auditflowlog 0.0.1

Multi-tenant audit logging and AI-assisted monitoring - Python SDK

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 Support - Both sync and async APIs available
• 🔌 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
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")

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

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
• include_extra (bool): Include loguru extra context (default: True)

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}")

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