AuditFlow Python SDK

A lightweight Python SDK for sending logs to AuditFlow and streaming them back in real-time.

Features

• 🚀 Easy Integration - Drop-in sink for loguru, handler for stdlib logging
• 📤 Log Ingestion - Send logs from any Python application to AuditFlow
• 📡 Real-time Streaming - Stream logs back via WebSocket with flexible filtering
• ⚡ Async by Default - Non-blocking async transport with dedicated event loop
• 🔁 Automatic Retry - Built-in retry with exponential backoff
• 💾 Disk Fallback - Logs saved to disk when backend is unreachable
• 📦 Batching - Efficient batch sending (100 logs or 5 seconds)
• 🔌 Extensible - Designed to support multiple logging frameworks

Installation

# Basic installation (HTTP client only)
pip install auditflowlog

# With loguru integration
pip install auditflowlog[loguru]

# With WebSocket streaming
pip install auditflowlog[streaming]

# Everything
pip install auditflowlog[all]

Quick Start

1. Using with Loguru

from loguru import logger
from auditflowlog import AuditFlowSink

# Add AuditFlow sink to loguru (async by default for non-blocking log emission)
logger.add(
    AuditFlowSink(
        api_key="af_live_abc123...",
        bucket="payment-service"
    )
)

# Use loguru as normal - logs automatically go to AuditFlow
logger.bind(request_id="REQ-12345")
logger.info("Payment captured", order_id="ORD-789", amount=99.99)
logger.error("Payment failed", reason="insufficient_funds")

# Optional: Use sync client instead of async
logger.add(
    AuditFlowSink(
        api_key="af_live_abc123...",
        bucket="payment-service",
        use_async=False  # Use sync client if event loop is undesirable
    )
)

2. Direct Client Usage

from auditflowlog import AuditFlowClient, LogLevel

client = AuditFlowClient(
    api_key="af_live_abc123...",
    bucket="bucket_id_here"  # Bucket ID, not name
)

# Send a single log
client.send_log(
    level=LogLevel.INFO,
    message="User logged in",
    metadata={"user_id": "user_123", "ip": "192.168.1.1"}
)

# Send multiple logs in a batch
from auditflowlog import LogEntry

logs = [
    LogEntry(bucket_id="bucket_id_here", level=LogLevel.INFO, message="Request started"),
    LogEntry(bucket_id="bucket_id_here", level=LogLevel.INFO, message="Processing data"),
    LogEntry(bucket_id="bucket_id_here", level=LogLevel.INFO, message="Request completed"),
]
client.send_logs(logs=logs)

3. Real-time Log Streaming

Note: WebSocket streaming requires a JWT token from user login, not an API key.

import asyncio
from auditflowlog import LogStream, LogLevel

async def main():
    # Stream logs with JWT token (obtained from /api/v1/auth/login)
    stream = LogStream(
        jwt_token="your_jwt_token_here",
        bucket_id="bucket_id_here",
        level=LogLevel.ERROR,  # Only errors and critical
    )
    
    # Option 1: Callback-based
    def on_log(log):
        print(f"[{log['level']}] {log['timestamp']}: {log['message']}")
    
    await stream.connect(callback=on_log)
    await stream.run()  # Runs forever with auto-reconnect
    
    # Option 2: Async iterator
    async with stream:
        async for log in stream:
            print(f"[{log['level']}] {log['message']}")
            
            # Stop on specific condition
            if log['level'] == 'critical':
                break

asyncio.run(main())

4. Using with Standard Library Logging

import logging
from auditflowlog.sink import StdlibHandler

logger = logging.getLogger(__name__)

# Add AuditFlow handler
handler = StdlibHandler(
    api_key="af_live_abc123...",
    bucket="my-service"
)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Use stdlib logging as normal
logger.info("Application started", extra={"version": "1.0.0"})
logger.error("Database connection failed", extra={"host": "localhost"})

Advanced Usage

Filtering Streamed Logs

from datetime import datetime, timedelta
from auditflowlog import LogStream, LogLevel

# Filter by multiple criteria
stream = LogStream(
    jwt_token="your_jwt_token_here",
    bucket_id="bucket_id_here",
    level=LogLevel.WARNING,  # WARNING and above
    start_time=datetime.now() - timedelta(hours=1),  # Last hour
    metadata_filters={
        "user_id": "user_123",  # Only logs for specific user
        "payment_status": "failed"
    }
)

async with stream:
    async for log in stream:
        print(log)

Async Client

import asyncio
from auditflowlog.client import AsyncAuditFlowClient, LogLevel

async def main():
    async with AsyncAuditFlowClient(
        api_key="af_live_abc123...",
        bucket="bucket_id_here"
    ) as client:
        await client.send_log(
            level=LogLevel.INFO,
            message="Async operation completed"
        )

asyncio.run(main())

Context Manager for Cleanup

from auditflowlog import AuditFlowClient

# Automatically closes connection on exit
with AuditFlowClient(api_key="...", bucket="bucket_id_here") as client:
    client.send_log(message="Processing batch")
    # ... more operations
# Connection closed here

Reliability Features

The SDK is designed for production use with built-in resilience:

Non-Blocking Architecture

• Fire-and-forget: Logs are enqueued immediately and processed in the background
• Async by default: Loguru sink uses AsyncAuditFlowClient with dedicated event loop
• No blocking I/O: Your application never waits for log transmission

Automatic Retry with Backoff

• Network errors trigger automatic retry (3 attempts)
• Exponential backoff: 2s, 4s, 8s (max 10s)
• Authentication (401) and rate limit (429) errors raise immediately

Disk Fallback

• Failed logs are saved to auditflow_fallback.log when backend is unreachable
• Automatic retry every 60 seconds when connectivity is restored
• Successfully sent logs are removed from fallback file

Efficient Batching

• Logs are batched automatically (100 items or 5 seconds)
• Reduces HTTP requests and network overhead
• Configurable queue size (default: 10,000 items)

Configuration Options

from auditflowlog import AuditFlowSink

sink = AuditFlowSink(
    api_key="af_live_abc123...",
    bucket="my-service",
    timeout=30.0,              # Request timeout
    enable_queue=True,          # Enable fire-and-forget queue
    queue_maxsize=10000,        # Max queue size
    use_async=True,             # Use async client (default)
    fallback_file="auditflow_fallback.log",  # Disk fallback path
)

Configuration

Environment Variables

You can configure common settings via environment variables:

export AUDITFLOW_API_KEY="af_live_abc123..."
export AUDITFLOW_BUCKET="default-service"

Then use them in your code:

import os
from auditflowlog import AuditFlowClient

client = AuditFlowClient(
    api_key=os.getenv("AUDITFLOW_API_KEY"),
    bucket=os.getenv("AUDITFLOW_BUCKET"),
)

API Reference

AuditFlowClient

Main synchronous client for log ingestion.

Constructor:

• api_key (str): Your AuditFlow API key (for log ingestion)
• bucket (str, optional): Default bucket ID
• timeout (float): Request timeout in seconds (default: 30.0)

Methods:

• send_log(message, level, bucket=None, source=None, timestamp=None, metadata=None) - Send a single log
• send_logs(logs) - Send multiple logs in batch (each LogEntry must have bucket_id)
• close() - Close HTTP connection

AuditFlowSink

Loguru sink for automatic log forwarding.

Constructor:

• api_key (str): Your AuditFlow API key (for log ingestion)
• bucket (str): Bucket ID for logs
• timeout (float): Request timeout (default: 30.0)
• include_extra (bool): Include loguru extra context (default: True)
• enable_queue (bool): Enable fire-and-forget queue (default: True, only applies to sync client)
• queue_maxsize (int): Maximum size of in-memory queue (default: 10000, only applies to sync client)
• use_async (bool): Use AsyncAuditFlowClient with dedicated event loop (default: True)
• auto_disable_on_auth_failure (bool): Auto-disable sink on authentication failures (default: True)

Methods:

• enable() - Enable the sink to start sending logs again
• disable() - Disable the sink to stop sending logs
• is_enabled() - Check if the sink is currently enabled
• close() - Close the client and cleanup resources

LogStream

Async WebSocket client for real-time log streaming.

Note: Requires JWT token from user authentication, not API key.

Constructor:

• jwt_token (str): JWT token from user login (for WebSocket streaming)
• bucket_id (str): Bucket ID to stream logs from
• level (LogLevel, optional): Filter by minimum log level
• start_time (datetime, optional): Filter by start time
• end_time (datetime, optional): Filter by end time
• metadata_filters (dict, optional): Filter by metadata fields
• reconnect (bool): Auto-reconnect on disconnect (default: True)
• reconnect_interval (float): Seconds between reconnects (default: 5.0)

Methods:

• connect(callback=None) - Connect to WebSocket
• disconnect() - Disconnect from WebSocket
• run() - Run stream with callback (blocking)
• Supports async context manager and async iterator

LogLevel

Enum for log severity levels:

• LogLevel.DEBUG
• LogLevel.INFO
• LogLevel.WARNING
• LogLevel.ERROR
• LogLevel.CRITICAL

Error Handling

The SDK provides specific exception types:

from auditflowlog.exceptions import (
    AuditFlowError,        # Base exception
    AuthenticationError,   # Invalid API key
    RateLimitError,       # Rate limit exceeded
    ConnectionError,      # WebSocket connection failed
    StreamError,          # Streaming error
)

try:
    client.send_log(message="test")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limit exceeded, retry later")
except AuditFlowError as e:
    print(f"AuditFlow error: {e}")

Circuit Breaker Pattern

The SDK includes a circuit breaker to prevent wasting resources on invalid API keys. When authentication failures reach a threshold (default: 3), the circuit opens and stops sending requests for a cooldown period (default: 60 seconds).

Client Circuit Breaker:

from auditflowlog import AuditFlowClient

# Configure circuit breaker thresholds
client = AuditFlowClient(
    api_key="your_api_key",
    bucket_id="my-bucket",
    circuit_breaker_threshold=3,  # Failures before opening
    circuit_breaker_timeout=60.0  # Seconds before retry
)

# Check circuit status
if client.is_circuit_open():
    print("Circuit is open, requests paused")

# Manually reset circuit breaker
client.reset_circuit_breaker()

Sink Auto-Disable:

The AuditFlowSink can automatically disable itself when authentication failures occur:

from loguru import logger
from auditflowlog import AuditFlowSink

sink = AuditFlowSink(
    api_key="your_api_key",
    bucket_id="my-bucket",
    auto_disable_on_auth_failure=True  # Default: True
)

logger.add(sink)

# If API key is invalid, sink will auto-disable
# Re-enable after fixing API key:
sink.enable()

# Or manually disable:
sink.disable()

# Check status
if sink.is_enabled():
    print("Sink is active")

Examples

See the examples/ directory for complete working examples:

• examples/loguru_example.py - Loguru integration
• examples/stdlib_example.py - Standard library logging
• examples/streaming_example.py - Real-time log streaming
• examples/async_example.py - Async client usage

Development

Setup

git clone https://github.com/jamesadewara/auditflowlog-py.git
cd auditflowlog-py
python -m venv venv
source venv/bin/activate  # or venv\Scripts\activate on Windows
pip install -e ".[all]"

Building

pip install build
python -m build

This creates:

• dist/auditflowlog-0.1.0-py3-none-any.whl
• dist/auditflowlog-0.1.0.tar.gz

Publishing to PyPI

pip install twine
python -m twine upload dist/*

License

APACHE License - see LICENSE file for details.

Contributing

Contributions welcome! Please open an issue or submit a pull request.

Links

• AuditFlow Backend
• Documentation
• Issue Tracker