How to write and query InfluxDB with Python

Writing and querying InfluxDB from Python proves that application code can send time series points and read them back from the same database. The InfluxDB 3 Python client fits ingestion jobs, service checks, and automation scripts that need one client for both operations.

The influxdb3-python package provides the influxdb_client_3 module. It writes Point objects through the InfluxDB write API and queries rows through Flight using SQL or InfluxQL, returning a PyArrow table that Python code can inspect or format.

A running InfluxDB 3 Core server, an existing python_client database, and a token with read and write permission are required. The older v2 Python client can still write to InfluxDB 3 through compatibility endpoints, but it cannot query through /api/v2/query, so use the v3 client when one script must write and query.

1. Create a virtual environment for the Python project.

   $ python3 -m venv .venv

2. Activate the virtual environment.

   $ source .venv/bin/activate

3. Install the InfluxDB 3 Python client.

   $ python -m pip install influxdb3-python
   ##### snipped #####
   Successfully installed influxdb3-python

4. Set the connection variables for the target InfluxDB database.

   $ export INFLUX_HOST='http://127.0.0.1:8181'
   $ export INFLUX_DATABASE='python_client'
   $ export INFLUX_TOKEN='AUTH_TOKEN'

   Use a token with read and write permission on the target database. Keep raw tokens out of shared transcripts, screenshots, shell history, and committed scripts.
   Related: How to create an InfluxDB 3 Core database
   Related: How to create an InfluxDB 3 Core admin token

5. Create the Python script that writes one point and queries it back.

   write-query.py

   import os
   from datetime import datetime, timezone
    
   from influxdb_client_3 import InfluxDBClient3, Point
    
   host = os.environ["INFLUX_HOST"]
   token = os.environ["INFLUX_TOKEN"]
   database = os.environ["INFLUX_DATABASE"]
    
   point = (
       Point("weather")
       .tag("room", "lab")
       .field("temperature", 22.4)
       .time(datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc))
   )
    
   with InfluxDBClient3(host=host, token=token, database=database) as client:
       client.write(point, write_precision="s")
       table = client.query(
           """
           SELECT time, room, temperature
           FROM weather
           WHERE room = 'lab'
           ORDER BY time
           """
       )
    
   for row in table.select(["time", "room", "temperature"]).to_pylist():
       print(f"{row['time']} room={row['room']} temperature={row['temperature']}")

   The timestamp is set explicitly so the query returns one predictable row instead of relying on the current clock.

6. Run the script and confirm the printed row matches the point.

   $ python write-query.py
   2026-01-01 00:00:00 room=lab temperature=22.4

   An empty result usually means the database name, token scope, table name, tag predicate, or target host does not match the write.
   Related: How to troubleshoot InfluxDB query errors