Typing Stubs for InfluxDB Client

Common errors

• error: Module 'influxdb_client' has no attribute 'InfluxDBClient' [attr-defined]

  cause This error often occurs when the `types-influxdb-client` package is installed, but the actual `influxdb-client` runtime library is missing or is installed in a way that the type checker cannot find it.
  fix Ensure the `influxdb-client` library is installed in your environment: `pip install influxdb-client`.

• error: Argument 'bucket' to 'write' of 'WriteApi' has incompatible type "str"; expected "str" [arg-type] (note: 'bucket' is a named argument)

  cause This Mypy error, despite appearing to match types, can indicate a version mismatch between the `influxdb-client` library and its corresponding type stubs, or a more subtle type incompatibility (e.g., Literal types). It can also occur if both `influxdb-client` and `types-influxdb-client` are installed and their internal types conflict.
  fix First, verify that `influxdb-client<1.46.0` is installed OR that `types-influxdb-client` is *not* installed. If the issue persists and versions are aligned, check for `Any` types masking the true issue or report a bug to the `typeshed` project.

• error: Cannot access member "query_api" for type "InfluxDBClient" [attr-defined]

  cause This `attr-defined` error typically suggests that the version of `influxdb-client` in use does not have the `query_api` method, or the type stubs incorrectly declare its presence/absence for your specific client version. This is common when the stub version doesn't align with the runtime library version.
  fix Ensure your `types-influxdb-client` version accurately reflects your installed `influxdb-client` version. For `influxdb-client>=1.46.0`, uninstall `types-influxdb-client` completely. If using an older `influxdb-client` version, downgrade `types-influxdb-client` to match.

Warnings

• breaking The `influxdb-client` library (runtime package) includes its own type annotations from version `1.46.0` onwards. If you are using `influxdb-client>=1.46.0`, you MUST uninstall `types-influxdb-client` to avoid duplicate type definitions and potential type-checking errors.
  Fix: If `influxdb-client>=1.46.0` is installed, uninstall this stub package: `pip uninstall types-influxdb-client`.
• gotcha This specific version of `types-influxdb-client` (`1.45.0.20241221`) is intended to provide accurate annotations for `influxdb-client==1.45.*`. Using it with significantly different versions of `influxdb-client` (e.g., `1.30.x` or `1.49.x`) may lead to incorrect type-checking results due to API mismatches.
  Fix: Always align the `types-influxdb-client` version with the `influxdb-client` version you are using. The stub package's version number (excluding the last part) usually indicates the target `influxdb-client` version (e.g., `1.45.0.x` for `influxdb-client==1.45.*`).
• gotcha This package, `types-influxdb-client`, requires `urllib3>=2` since its version `1.37.0.1`. If your project environment necessitates `urllib3<2`, you will encounter dependency conflicts or runtime issues.
  Fix: Ensure `urllib3>=2` is installed. If `urllib3<2` is strictly required, you must downgrade `types-influxdb-client` to a version `<1.37.0.1` (e.g., `pip install 'types-influxdb-client<1.37.0.1'`).

Install

• pip install types-influxdb-client Install package
• pip install influxdb-client Install peer dependency

Imports

• InfluxDBClient

  from influxdb_client import InfluxDBClient

• Point

  from influxdb_client import Point

• SYNCHRONOUS

  from influxdb_client.client.write_api import SYNCHRONOUS

Quickstart

This quickstart demonstrates basic InfluxDB client usage with type hints. It connects to an InfluxDB instance, writes a data point using a `Point` object, and then queries the data, showcasing how the type stubs aid in correct API usage and provide autocompletion and static checks.

import os
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

# Configuration (use environment variables or replace with actual values)
INFLUXDB_URL = os.environ.get('INFLUXDB_URL', 'http://localhost:8086')
INFLUXDB_TOKEN = os.environ.get('INFLUXDB_TOKEN', 'my-super-secret-token')
INFLUXDB_ORG = os.environ.get('INFLUXDB_ORG', 'my-org')
INFLUXDB_BUCKET = os.environ.get('INFLUXDB_BUCKET', 'my-bucket')

def write_and_query_data() -> None:
    try:
        with InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG) as client:
            write_api = client.write_api(write_options=SYNCHRONOUS)

            p = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3)
            write_api.write(bucket=INFLUXDB_BUCKET, record=p)
            print(f"Successfully wrote point: {p.to_line_protocol()}")

            query_api = client.query_api()
            tables = query_api.query(f'from(bucket:"{INFLUXDB_BUCKET}") |> range(start: -1h)')

            print("\nQuery Results:")
            for table in tables:
                for record in table.records:
                    print(f"  {record.get_measurement()}: {record.get_field()}={record.get_value()} @ {record.get_time()}")

    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    write_and_query_data()