Influx Manager Module¶

Functions:

`time_clause`(timestamp_ms)	If a timestamp is provided as a string (delta) or int, it converts it to a delta string or ISO format.
`influx_client`(url, token, org[, timeout])	Function to obtain an InfluxDB client in read mode.
`fetch_measurements`(client, bucket, org)	Function to fetch all the measurements from a specific bucket.
`fetch_measurement_tag_keys`(client, bucket, ...)	Function to fetch all the tags of a specific measurement in a specific bucket.
`fetch_measurement_field_keys`(client, bucket, ...)	Function to fetch all the fields of a specific measurement in a specific bucket.
`fetch_measurement_tag_values`(client, ...[, ...])	Function to fetch all the values a tag has in a specific measurement in a specific bucket during the specified time period.
`fetch_data`(client, bucket, measurement, org)	Function to retrieve data from a specific measurement in a specific bucket, filtered by specific tags and fields.
`delete_measurement`(client, org, bucket, ...)	Function to delete a measurement in InfluxDB 2.x.
`delete_entries_by_tag_value`(client, org, ...)	Function to delete entries in a specific measurement where a tag has a specific value.
`create_bulk_points`(client, bucket, org, ...)	Function to create a new measurement in InfluxDB 2.x.

time_clause(timestamp_ms: int) → str | None[source]¶

If a timestamp is provided as a string (delta) or int, it converts it to a delta string or ISO format.

Parameters:: timestamp_ms – A timestamp in milliseconds. If a string is passed, the expected format is ISO or Influx Delta: 1h, -1d, etc…
Returns:: A string in ISO format or delta.

influx_client(url: str, token: str, org: str, timeout: int = 10000) → influxdb_client.InfluxDBClient[source]¶

Function to obtain an InfluxDB client in read mode.

Parameters:

url (str) – InfluxDB URL. For example, “http://localhost:8086”
token (str) – Access token to InfluxDB.
org (str) – InfluxDB organization. For example, “your_org”
timeout (int) – Connection timeout in milliseconds.

Returns:

An InfluxDB client.

fetch_measurements(client: influxdb_client.InfluxDBClient, bucket: str, org: str) → set[source]¶

Function to fetch all the measurements from a specific bucket.

Parameters:

client – Client connected to InfluxDB.
bucket – InfluxDB bucket.
org – InfluxDB organization.

Returns:

List of measurements.

fetch_measurement_tag_keys(client: influxdb_client.InfluxDBClient, bucket: str, measurement: str, org: str) → set[source]¶

Function to fetch all the tags of a specific measurement in a specific bucket.

Parameters:

client – Client connected to InfluxDB.
bucket – Bucket name.
measurement – Measurement name. For example, “trades”.
org – Organization name. For example, “your_org”.

Returns:

Set of unique tags. Return example:

{'buyer_is_maker', 'stream', 'symbol'}

fetch_measurement_field_keys(client: influxdb_client.InfluxDBClient, bucket: str, measurement: str, org: str) → set[source]¶

Function to fetch all the fields of a specific measurement in a specific bucket.

Parameters:

client – Client connected to InfluxDB.
bucket – Bucket name.
measurement – Measurement name. For example, “trades”.
org – Organization name. For example, “your_org”.

Returns:

Set of unique fields. Return example:

{'buyer_order_id',
 'event_time',
 'price',
 'quantity',
 'quote_quantity',
 'seller_order_id',
 'trade_id',
 'trade_time'}

fetch_measurement_tag_values(client: influxdb_client.InfluxDBClient, measurement: str, tag: str, bucket: str, org: str, start_time: int = None, end_time: int = None) → set[source]¶

Function to fetch all the values a tag has in a specific measurement in a specific bucket during the specified time period.

Parameters:

client – Client connected to InfluxDB.
measurement – Measurement name. For example, “trades”.
tag – Tag name. For example, “stream”.
bucket – Bucket name.
org – Organization name. For example, “your_org”.
start_time – Timestamp in milliseconds for the start of the data range. Examples: 1631000000000, “2021-09-07T00:00:00Z”, “-1d”.
end_time – Timestamp in milliseconds for the end of the data range. Examples: 1631000000000, “2021-09-07T00:00:00Z”, “-1d”.

Returns:

Set of unique symbols. Return example:

streams = fetch_tag_values(client=client, bucket=influx_bucket, measurement="trades", org=influx_org, tag="stream")
print(streams)

{'duskusdt@trade', 'dotusdt@trade', 'multiusdt@trade', 'fluxusdt@trade', 'scrtusdt@trade', 'frontusdt@trade', 'tfuelusdt@trade',
 'hftusdt@trade', 'alcxusdt@trade', 'rsrusdt@trade', 'joeusdt@trade', 'tkousdt@trade', 'vthousdt@trade', 'galusdt@trade',
  'osmousdt@trade','...}

fetch_data(client: influxdb_client.InfluxDBClient, bucket: str, measurement: str, org: str, tags_filter: dict = {}, fields_in_ret: list = [], tags_in_ret: list = [], start_time=None, end_time=None) → list[source]¶

Function to retrieve data from a specific measurement in a specific bucket, filtered by specific tags and fields.

Parameters:

client – Client connected to InfluxDB
bucket – Bucket name
measurement – Measurement name. For example, “trades”
org – Organization name. For example, “your_org”
tags_filter – Dictionary of tags to filter by. For example, {“symbol”: “BTC”, “stream”: “trades”}
fields_in_ret – List of fields to select. For example, [“field1”, “field2”]
tags_in_ret – List of tags to include in the results. For example, [“stream”]
start_time – Timestamp in milliseconds for the start of the data range. Examples: 1631000000000, “2021-09-07T00:00:00Z”, “-1d”
end_time – Timestamp in milliseconds for the end of the data range. Examples: 1631000000000, “2021-09-07T00:00:00Z”, “-1d”

Returns:

List of results. Return example:

data = fetch_data(client=client,
                  bucket=influx_bucket,
                  measurement="trades",
                  org=influx_org,
                  tags={"stream": "btcusdt@trade", "buyer_is_maker": True},
                  fields=fields,
                  tag_keys=tags)
[{'result': '_result',
  'table': 0,
  '_time': datetime.datetime(2023, 9, 14, 16, 0, 2, 454000, tzinfo=tzutc()),
  '_measurement': 'trades',
  'buyer_is_maker': 'True',
  'stream': 'btcusdt@trade',
  'buyer_order_id': 22340720310,
  'price': 26693.97,
  'quantity': 0.00085,
  'quote_quantity': 22.6898745,
  'seller_order_id': 22340720976,
  'trade_id': 3212341083},
  ...]

delete_measurement(client: influxdb_client.InfluxDBClient, org: str, bucket: str, measurement: str, days: int = 30)[source]¶

Function to delete a measurement in InfluxDB 2.x. The data is not deleted directly, but rather the data within it is deleted. The process is asynchronous, so it may take some time to complete.

Parameters:

client – Client connected to InfluxDB.
org – InfluxDB organization.
bucket – InfluxDB bucket.
measurement – Name of the measurement to delete.
days – Number of days to delete.

Returns:

None

delete_entries_by_tag_value(client: influxdb_client.InfluxDBClient, org: str, bucket: str, measurement: str, tag_key: str, tag_value: str, days: int = 30)[source]¶

Function to delete entries in a specific measurement where a tag has a specific value.

Parameters:

client – Client connected to InfluxDB.
org – InfluxDB organization.
bucket – InfluxDB bucket.
measurement – Name of the measurement to delete.
tag_key – Name of the tag.
tag_value – Value of the tag.
days – Number of days to delete.

Returns:

None

create_bulk_points(client: influxdb_client.InfluxDBClient, bucket: str, org: str, measurement: str, data_list: list[dict])[source]¶

Function to create a new measurement in InfluxDB 2.x.

Parameters:

client – Client connected to InfluxDB.
bucket – InfluxDB bucket.
org – InfluxDB organization.
measurement – Measurement to create.

data_list –

Data to insert. In the format of a list of dictionaries with expected keys “fields”, “tags” and “time”. For example:

data_list = [
    {"fields": {"field1": 1, "field2": 2},
     "tags": {"tag1": "tag1", "tag2": "tag2"},
     "time": 1234567890000000000}
]

Returns:

None