influx-rust 0.3.0

High-performance InfluxDB query interface for Python

influx-rust

High-performance InfluxDB query library for Python, powered by Rust.

Why influx-rust?

10x faster InfluxDB queries compared to native Python implementations. By leveraging Rust's performance with PyO3 bindings, influx-rust eliminates JSON serialization overhead and interpreter bottlenecks while maintaining a familiar Python API.

Performance Comparison

Implementation	Query Time (40+ sources, 24h range)	Improvement
Python (influxdb-client)	~38-40 seconds	baseline
influx-rust	~3-4 seconds	10x faster ✅

Real-world performance measured on production queries with 40+ data sources, aggregations, and time-windowed data.

Features

• 🚀 10x faster than Python native InfluxDB clients
• 🔄 Drop-in replacement for existing influxdb-client async queries
• 🦀 Rust-powered performance with zero-copy deserialization
• 🐍 Python-friendly interface - both async/await and synchronous
• 📦 Pre-built wheels for Linux, macOS, and Windows
• 🔒 Type-safe Rust implementation with comprehensive error handling
• ⚡ Tokio async runtime for concurrent query execution
• 🔓 GIL-free I/O - releases Python's Global Interpreter Lock during queries

Installation

pip install influx-rust

Requires Python 3.8 or higher.

Quick Start

Async Version (for async/await code)

from influx_rust import get_influx_data_async

# For use in async functions
results = await get_influx_data_async(
    url="https://your-influxdb.com",
    token="your-token",
    org="your-org",
    query='''
        from(bucket: "sensors")
        |> range(start: -24h)
        |> filter(fn: (r) => r._measurement == "temperature")
        |> aggregateWindow(every: 30m, fn: mean)
    '''
)

# Returns list of dictionaries
for record in results:
    print(record)  # {'_time': '2024-01-29T10:00:00Z', '_value': 23.5, ...}

Synchronous Version (for non-async code)

from influx_rust import get_influx_data

# For use in regular synchronous functions
results = get_influx_data(
    url="https://your-influxdb.com",
    token="your-token",
    org="your-org",
    query='''
        from(bucket: "sensors")
        |> range(start: -24h)
        |> filter(fn: (r) => r._measurement == "temperature")
        |> aggregateWindow(every: 30m, fn: mean)
    '''
)

# Returns list of dictionaries (same as async version)
for record in results:
    print(record)  # {'_time': '2024-01-29T10:00:00Z', '_value': 23.5, ...}

Usage Examples

Basic Async Query

import asyncio
from influx_rust import get_influx_data_async

async def fetch_temperature_data():
    data = await get_influx_data_async(
        url="https://influx.example.com",
        token="your_token_here",
        org="my_org",
        query='from(bucket: "sensors") |> range(start: -1h)'
    )
    return data

# Run the async function
results = asyncio.run(fetch_temperature_data())
print(f"Retrieved {len(results)} records")

Basic Synchronous Query

For non-async contexts (Flask, Django, Dramatiq workers, simple scripts):

from influx_rust import get_influx_data

# No async/await needed - just call the function directly
def fetch_temperature_data():
    data = get_influx_data(
        url="https://influx.example.com",
        token="your_token_here",
        org="my_org",
        query='from(bucket: "sensors") |> range(start: -1h)'
    )
    return data

# Direct function call - no asyncio.run() needed
results = fetch_temperature_data()
print(f"Retrieved {len(results)} records")

Note: The synchronous version (get_influx_data) still benefits from Rust's performance and properly releases Python's GIL during I/O operations, allowing other threads to run.

Complex Query with Aggregations

# Multi-source query with joins and aggregations
query = '''
from(bucket: "agua_mar")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "oxygen")
  |> filter(fn: (r) => contains(value: r.source, set: ["sensor1", "sensor2", "sensor3"]))
  |> aggregateWindow(every: 30m, fn: mean)
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
'''

data = await get_influx_data_async(
    url=INFLUXDB_URL,
    token=INFLUXDB_TOKEN,
    org=INFLUXDB_ORG,
    query=query
)

Integration with Existing Python Code

Drop-in replacement for influxdb-client:

# BEFORE: Using influxdb-client (slow)
from influxdb_client import InfluxDBClient
from influxdb_client.client.write_api import SYNCHRONOUS

async def get_data_old_way():
    client = InfluxDBClient(url=url, token=token, org=org)
    query_api = client.query_api()
    tables = query_api.query(query)
    # ... process tables ...

# AFTER: Using influx-rust (fast)
from influx_rust import get_influx_data_async

async def get_data_new_way():
    # Same result, 10x faster
    results = await get_influx_data_async(
        url=url,
        token=token,
        org=org,
        query=query
    )
    return results

Error Handling

from influx_rust import get_influx_data_async

try:
    data = await get_influx_data_async(
        url="https://influx.example.com",
        token="invalid_token",
        org="my_org",
        query="from(bucket: 'test') |> range(start: -1h)"
    )
except Exception as e:
    print(f"Query failed: {e}")
    # Handle authentication errors, network issues, etc.

How It Works

influx-rust uses PyO3 to create Python bindings for a high-performance Rust implementation:

1. Rust Core: Uses the official influxdb2 Rust client
2. Zero-Copy Deserialization: Directly processes InfluxDB responses without intermediate JSON strings
3. Async Runtime: Powered by Tokio for efficient concurrent operations
4. Python Bindings: PyO3 exposes Rust functions as native Python async functions

Why It's Faster

Bottleneck	Python (influxdb-client)	influx-rust
JSON serialization	❌ Double serialization (InfluxDB → JSON → Python)	✅ Zero-copy deserialization
Interpreter overhead	❌ Python GIL and interpreter	✅ Compiled Rust (no GIL)
Memory allocations	❌ Intermediate string buffers	✅ Direct struct mapping
Async runtime	❌ Python asyncio	✅ Tokio (native threads)

Development

Building from Source

# Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# Clone repository
git clone https://github.com/your-org/influx-rust.git
cd influx-rust

# Create virtual environment
python -m venv .venv
source .venv/bin/activate

# Install maturin
pip install maturin

# Build and install in development mode
maturin develop --release

Running Tests

# Build release version
cargo build --release

# Run Rust tests
cargo test

# Test Python integration
python -c "from influx_rust import get_influx_data_async; print('✅ Import successful')"

Performance Testing

Compare performance with Python implementation:

# Set environment variables
export INFLUXDB_URL="https://your-influxdb.com"
export INFLUXDB_TOKEN="your-token"
export INFLUXDB_ORG="your-org"

# Run comparison script
./compare_performance.sh

API Reference

get_influx_data_async

async def get_influx_data_async(
    url: str,
    token: str,
    org: str,
    query: str
) -> list[dict[str, str]]

Execute an InfluxDB Flux query asynchronously.

Parameters:

• url (str): InfluxDB server URL (e.g., https://influx.example.com)
• token (str): Authentication token
• org (str): Organization name
• query (str): Flux query string

Returns:

• list[dict[str, str]]: List of records as dictionaries. Each dictionary contains InfluxDB fields like _time, _value, _measurement, etc.

Raises:

• Exception: On authentication errors, network failures, or invalid queries

Requirements

• Python: 3.8 or higher
• Operating Systems: Linux (x86_64, aarch64), macOS (x86_64, arm64), Windows (x86_64)

Pre-built wheels are available for all supported platforms. No Rust toolchain required for installation.

Deployment

Docker

When using in Docker, simply install from PyPI:

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .

# influx-rust installs from pre-built wheel (no Rust needed!)
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
CMD ["python", "app.py"]

No need to install Rust in your Docker container!

Production Considerations

• Connection Pooling: influx-rust reuses connections internally
• Timeout: Default timeout is 30 seconds (configurable in future versions)
• Memory: Significantly lower memory usage vs Python client (no intermediate buffers)
• Logging: Enable debug logs with RUST_LOG=debug environment variable

Roadmap

• Configurable timeouts
• Connection pooling configuration
• Write API support (currently read-only)
• Streaming queries for large datasets
• Custom error types instead of generic exceptions
• Sync version of the API (non-async)

Benchmarks

Real-world production query (AquaChile agua_mar monitoring):

Query: 40+ sensors, 24h range, aggregations, joins
Records: ~12,450 records

Python (influxdb-client):
  ⏱️  Query execution: 38.2s
  📊 Memory usage: ~450MB

influx-rust:
  ⏱️  Query execution: 3.8s  (10x faster)
  📊 Memory usage: ~180MB  (2.5x less)

Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

License

MIT License - see LICENSE for details.

Credits

Built with:

• PyO3 - Rust bindings for Python
• influxdb2 - Official InfluxDB Rust client
• Tokio - Async runtime for Rust
• Maturin - Build and publish Rust-based Python packages

Developed by AquaChile DevOps for high-performance aquaculture monitoring.

Support

• 📫 Issues: GitHub Issues
• 📖 Documentation: GitHub Wiki
• 💬 Discussions: GitHub Discussions