From cb637c29cb024ff14f4e08f6897bfb4e888504cd Mon Sep 17 00:00:00 2001 From: bboger Date: Thu, 17 Feb 2022 10:36:35 +0100 Subject: [PATCH] feat: add modifiable __str__ to Point. --- influxdb_client/client/write/point.py | 22 ++++++++++-- tests/test_point.py | 52 +++++++++++++++++---------- 2 files changed, 54 insertions(+), 20 deletions(-) diff --git a/influxdb_client/client/write/point.py b/influxdb_client/client/write/point.py index c9fcbfbc..9f0dda24 100644 --- a/influxdb_client/client/write/point.py +++ b/influxdb_client/client/write/point.py @@ -1,6 +1,5 @@ """Point data structure to represent LineProtocol.""" - import math from builtins import int from datetime import datetime, timedelta @@ -54,6 +53,8 @@ class Point(object): Ref: http://bit.ly/influxdata-point """ + __str___rep = None + @staticmethod def measurement(measurement): """Create a new Point with specified measurement name.""" @@ -146,7 +147,6 @@ def __init__(self, measurement_name): self._name = measurement_name self._time = None self._write_precision = DEFAULT_WRITE_PRECISION - pass def time(self, time, write_precision=DEFAULT_WRITE_PRECISION): """ @@ -195,6 +195,24 @@ def write_precision(self): """Get precision.""" return self._write_precision + @classmethod + def set_str_rep(cls, rep_function): + """Set the string representation for all Points.""" + cls.__str___rep = rep_function + + def __str__(self): + """ + Create string representation of this Point. + + Can be set via `Point.set_str_rep`. Defaults to `to_line_protocol` + Example: + .. code-block:: python + Point.set_str_rep(lambda p: f'{p._name} - {p._tags} - {p._fields} - {p._time}') + """ + if self.__str___rep is None: + return self.to_line_protocol() + return self.__str___rep() + def _append_tags(tags): _return = [] diff --git a/tests/test_point.py b/tests/test_point.py index 53fd0cc5..b599a844 100644 --- a/tests/test_point.py +++ b/tests/test_point.py @@ -11,6 +11,18 @@ class PointTest(unittest.TestCase): + def test_ToStr(self): + point = Point.measurement("h2o").tag("location", "europe").field("level", 2.2) + expected_str = "h2o,location=europe level=2.2" + self.assertEqual(expected_str, str(point)) + + def my_str_rep(p: Point) -> str: + return f'{p._name} - {p._tags} - {p._fields} - {p._time}' + + Point.set_str_rep(my_str_rep) + + self.assertEqual(my_str_rep(point), str(point)) + def test_MeasurementEscape(self): point = Point.measurement("h2 o").tag("location", "europe").tag("", "warn").field("level", 2) self.assertEqual(point.to_line_protocol(), "h2\\ o,location=europe level=2i") @@ -36,17 +48,17 @@ def test_TagEmptyValue(self): self.assertEqual("h2o,location=europe level=2i", point.to_line_protocol()) def test_TagEscapingKeyAndValue(self): - point = Point.measurement("h\n2\ro\t_data") \ .tag("new\nline", "new\nline") \ .tag("carriage\rreturn", "carriage\nreturn") \ .tag("t\tab", "t\tab") \ .field("level", 2) - self.assertEqual("h\\n2\\ro\\t_data,carriage\\rreturn=carriage\\nreturn,new\\nline=new\\nline,t\\tab=t\\tab level=2i", point.to_line_protocol()) + self.assertEqual( + "h\\n2\\ro\\t_data,carriage\\rreturn=carriage\\nreturn,new\\nline=new\\nline,t\\tab=t\\tab level=2i", + point.to_line_protocol()) def test_EqualSignEscaping(self): - point = Point.measurement("h=2o") \ .tag("l=ocation", "e=urope") \ .field("l=evel", 2) @@ -391,22 +403,24 @@ def test_backslash(self): def test_numpy_types(self): from influxdb_client.extras import np - point = Point.measurement("h2o")\ - .tag("location", "europe")\ - .field("np.float1", np.float(1.123))\ - .field("np.float2", np.float16(2.123))\ - .field("np.float3", np.float32(3.123))\ - .field("np.float4", np.float64(4.123))\ - .field("np.int1", np.int8(1))\ - .field("np.int2", np.int16(2))\ - .field("np.int3", np.int32(3))\ - .field("np.int4", np.int64(4))\ - .field("np.uint1", np.uint8(5))\ - .field("np.uint2", np.uint16(6))\ - .field("np.uint3", np.uint32(7))\ + point = Point.measurement("h2o") \ + .tag("location", "europe") \ + .field("np.float1", np.float(1.123)) \ + .field("np.float2", np.float16(2.123)) \ + .field("np.float3", np.float32(3.123)) \ + .field("np.float4", np.float64(4.123)) \ + .field("np.int1", np.int8(1)) \ + .field("np.int2", np.int16(2)) \ + .field("np.int3", np.int32(3)) \ + .field("np.int4", np.int64(4)) \ + .field("np.uint1", np.uint8(5)) \ + .field("np.uint2", np.uint16(6)) \ + .field("np.uint3", np.uint32(7)) \ .field("np.uint4", np.uint64(8)) - self.assertEqual("h2o,location=europe np.float1=1.123,np.float2=2.123,np.float3=3.123,np.float4=4.123,np.int1=1i,np.int2=2i,np.int3=3i,np.int4=4i,np.uint1=5i,np.uint2=6i,np.uint3=7i,np.uint4=8i", point.to_line_protocol()) + self.assertEqual( + "h2o,location=europe np.float1=1.123,np.float2=2.123,np.float3=3.123,np.float4=4.123,np.int1=1i,np.int2=2i,np.int3=3i,np.int4=4i,np.uint1=5i,np.uint2=6i,np.uint3=7i,np.uint4=8i", + point.to_line_protocol()) def test_from_dictionary_custom_measurement(self): dictionary = { @@ -457,7 +471,9 @@ def test_from_dictionary_custom_fields(self): record_measurement_key="name", record_tag_keys=["location", "version"], record_field_keys=["pressure", "temperature"]) - self.assertEqual("sensor_pt859,location=warehouse_125,version=2021.06.05.5874 pressure=125i,temperature=10i 1632208639", point.to_line_protocol()) + self.assertEqual( + "sensor_pt859,location=warehouse_125,version=2021.06.05.5874 pressure=125i,temperature=10i 1632208639", + point.to_line_protocol()) def test_from_dictionary_tolerant_to_missing_tags_and_fields(self): dictionary = {