Method
add_data_point
(
self,
field_name: str,
field_value: str | int | float,
tags: typing.Mapping[
str,
str | int | float,
]
| None = None,
)
Source from the content-addressed store, hash-verified
| 73 | return cls.get_downsampled_bucket(DownsampleSize.FIFTEEN_MINUTES) |
| 74 | |
| 75 | def add_data_point( |
| 76 | self, |
| 77 | field_name: str, |
| 78 | field_value: str | int | float, |
| 79 | tags: typing.Mapping[ |
| 80 | str, |
| 81 | str | int | float, |
| 82 | ] |
| 83 | | None = None, |
| 84 | ) -> None: |
| 85 | point = Point(self.name) |
| 86 | point.field(field_name, field_value) |
| 87 | |
| 88 | if tags is not None: |
| 89 | for tag_key, tag_value in tags.items(): |
| 90 | point = point.tag(tag_key, tag_value) |
| 91 | |
| 92 | self.records.append(point) |
| 93 | |
| 94 | def write(self) -> None: |
| 95 | """Persist collected data points to InfluxDB""" |