or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client.mddata-management.mddatabase-operations.mddataframe-client.mdindex.mdlegacy.md
tile.json

tessl/pypi-influxdb

InfluxDB client library for time series database operations with comprehensive API for data management and querying

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/influxdb@5.3.x

To install, run

npx @tessl/cli install tessl/pypi-influxdb@5.3.0

index.mddocs/

InfluxDB

A comprehensive Python client library for InfluxDB, the open-source time series database. Provides full functionality for connecting to InfluxDB instances, writing time series data points, querying data with InfluxQL, managing databases and retention policies, and working with time series measurements. The library supports both standard JSON-based operations and DataFrame integration via pandas for data analysis workflows, with built-in authentication, connection pooling, batch operations, and comprehensive error handling for robust production deployments.

Package Information

  • Package Name: influxdb
  • Language: Python
  • Installation: pip install influxdb
  • Dependencies: requests, python-dateutil, pytz, six
  • Optional Dependencies: pandas (for DataFrameClient)

Core Imports

from influxdb import InfluxDBClient, DataFrameClient, SeriesHelper

For exception handling:

from influxdb.exceptions import InfluxDBClientError, InfluxDBServerError

For line protocol utilities:

from influxdb.line_protocol import make_line, make_lines

Basic Usage

Quick Start with InfluxDBClient

from influxdb import InfluxDBClient
import datetime

# Connect to InfluxDB
client = InfluxDBClient(host='localhost', port=8086, username='root', password='root')

# Create database
client.create_database('mydb')
client.switch_database('mydb')

# Write data points
json_body = [
    {
        "measurement": "cpu_load_short",
        "tags": {
            "host": "server01",
            "region": "us-west"
        },
        "time": "2023-09-07T07:18:24Z",
        "fields": {
            "value": 0.64
        }
    }
]

client.write_points(json_body)

# Query data
result = client.query('SELECT * FROM cpu_load_short')
print("Result: {0}".format(result))

# Close connection
client.close()

Context Manager Usage

from influxdb import InfluxDBClient

with InfluxDBClient(host='localhost', port=8086, database='mydb') as client:
    # Write data
    client.write_points([
        {
            "measurement": "temperature",
            "tags": {"location": "room1"},
            "fields": {"value": 23.5},
            "time": "2023-09-07T07:18:24Z"
        }
    ])
    
    # Query data
    results = client.query('SELECT * FROM temperature')

Architecture

The InfluxDB Python client follows a layered architecture:

  • InfluxDBClient: Core client with complete InfluxDB API support
  • DataFrameClient: Extends InfluxDBClient for pandas DataFrame integration
  • SeriesHelper: Bulk data insertion with configurable auto-commit
  • ResultSet: Query result processing with filtering and iteration
  • Line Protocol: Efficient data serialization for high-throughput writes
  • Legacy Support: Backward compatibility with InfluxDB 0.8.x via influxdb08 module

Capabilities

Core Client Operations

Comprehensive InfluxDB client with connection management, authentication, database operations, data querying and writing, user management, retention policies, continuous queries, and UDP support.

class InfluxDBClient:
    def __init__(self, host='localhost', port=8086, username='root', password='root', 
                 database=None, ssl=False, verify_ssl=False, timeout=None, retries=3,
                 use_udp=False, udp_port=4444, proxies=None, pool_size=10, path='',
                 cert=None, gzip=False, session=None, headers=None, socket_options=None): ...
    
    def query(self, query, params=None, bind_params=None, epoch=None, 
              expected_response_code=200, database=None, raise_errors=True, 
              chunked=False, chunk_size=0, method="GET"): ...
    
    def write_points(self, points, time_precision=None, database=None, 
                     retention_policy=None, tags=None, batch_size=None, 
                     protocol='json', consistency=None): ...
    
    def ping(self): ...
    def close(self): ...

Core Client

DataFrame Integration

Pandas DataFrame integration for data analysis workflows, enabling direct DataFrame-to-InfluxDB operations and query results as DataFrames.

class DataFrameClient(InfluxDBClient):
    def write_points(self, dataframe, measurement, tags=None, tag_columns=None, 
                     field_columns=None, time_precision=None, database=None, 
                     retention_policy=None, batch_size=None, protocol='line', 
                     numeric_precision=None): ...
    
    def query(self, query, params=None, bind_params=None, epoch=None, 
              expected_response_code=200, database=None, raise_errors=True, 
              chunked=False, chunk_size=0, method="GET", dropna=True, 
              data_frame_index=None): ...

DataFrame Client

Data Management Utilities

Bulk data insertion, line protocol formatting, and query result processing utilities for efficient data operations.

class SeriesHelper:
    def __init__(self, **kw): ...
    def commit(cls, client=None): ...

def make_line(measurement, tags=None, fields=None, time=None, precision=None): ...
def make_lines(data, precision=None): ...

class ResultSet:
    def __init__(self, series, raise_errors=True): ...
    def get_points(self, measurement=None, tags=None): ...

Data Management

Database Operations

Complete database lifecycle management including database creation/deletion, user management, retention policies, continuous queries, and measurement operations.

def create_database(self, dbname): ...
def drop_database(self, dbname): ...
def get_list_database(self): ...

def create_user(self, username, password, admin=False): ...
def drop_user(self, username): ...
def get_list_users(self): ...

def create_retention_policy(self, name, duration, replication, database=None, 
                           default=False, shard_duration="0s"): ...
def get_list_retention_policies(self, database=None): ...

Database Operations

Legacy InfluxDB 0.8 Support

Backward compatibility support for InfluxDB 0.8.x servers with dedicated client implementations.

from influxdb.influxdb08 import InfluxDBClient as LegacyClient
# Legacy API with different method signatures for InfluxDB 0.8.x

Legacy Support

Error Handling

The library provides specific exception types for different error conditions:

class InfluxDBClientError(Exception):
    """Raised when an error occurs in the request."""
    def __init__(self, content, code=None): ...

class InfluxDBServerError(Exception):  
    """Raised when a server error occurs (5xx status codes)."""
    def __init__(self, content): ...

Connection Configuration

SSL/TLS Configuration

client = InfluxDBClient(
    host='influxdb.example.com',
    port=8086,
    ssl=True,
    verify_ssl=True,
    cert='/path/to/client.crt'
)

Connection Pooling

client = InfluxDBClient(
    host='localhost',
    pool_size=20,  # Number of connections in pool
    timeout=30     # Request timeout in seconds
)

UDP Configuration

client = InfluxDBClient(
    host='localhost',
    use_udp=True,
    udp_port=4444
)

# Send UDP packets
client.send_packet(packet_data, protocol='json')