LogTide Python SDK

Official Python SDK for LogTide — self-hosted log management with async client, logging integration, batching, retry, circuit breaker, and middleware.

Features

• Sync & async clients — LogTideClient (requests) and AsyncLogTideClient (aiohttp)
• stdlib logging integration — drop-in LogTideHandler for existing logging setups
• structlog integration — LogTideProcessor to forward structlog events (optional, no hard dependency)
• Automatic batching with configurable size and interval
• Retry logic with exponential backoff
• Circuit breaker pattern for fault tolerance
• Payload limits — field truncation, base64 removal, field exclusion, max entry size
• Max buffer size with silent drop policy to prevent memory leaks
• Query API for searching and filtering logs
• Live tail with Server-Sent Events (SSE)
• Trace ID context for distributed tracing
• Global metadata added to all logs
• Structured exception serialization with parsed stack frames
• Internal metrics (logs sent, errors, latency, circuit breaker trips)
• Flask, Django, FastAPI & Starlette middleware for auto-logging HTTP requests
• Full Python 3.10+ support with type hints

Requirements

• Python 3.10 or higher

Installation

pip install logtide-sdk

Optional Dependencies

# Async client (AsyncLogTideClient)
pip install logtide-sdk[async]

# Flask middleware
pip install logtide-sdk[flask]

# Django middleware
pip install logtide-sdk[django]

# FastAPI middleware
pip install logtide-sdk[fastapi]

# Starlette middleware (standalone, without FastAPI)
pip install logtide-sdk[starlette]

# Install all extras
pip install logtide-sdk[async,flask,django,fastapi,starlette]

Quick Start

from logtide_sdk import LogTideClient, ClientOptions

client = LogTideClient(
    ClientOptions(
        api_url='http://localhost:8080',
        api_key='lp_your_api_key_here',
    )
)

client.info('api-gateway', 'Server started', {'port': 3000})
client.error('database', 'Connection failed', Exception('Timeout'))

# Graceful shutdown (also registered automatically via atexit)
client.close()

Configuration Options

Basic Options

Advanced Options

Payload Limits

PayloadLimitsOptions prevents 413 errors from oversized entries.

from logtide_sdk import LogTideClient, ClientOptions, PayloadLimitsOptions

client = LogTideClient(
    ClientOptions(
        api_url='http://localhost:8080',
        api_key='lp_your_api_key_here',
        payload_limits=PayloadLimitsOptions(
            max_field_size=5 * 1024,
            exclude_fields=['password', 'token'],
        ),
    )
)

Base64-encoded strings (data URIs or long base64 blobs) are automatically replaced with "[BASE64 DATA REMOVED]".

Sync Client

Logging Methods

client.debug('service', 'Debug message')
client.info('service', 'Info message', {'userId': 123})
client.warn('service', 'Warning message')
client.error('service', 'Error message', {'custom': 'data'})
client.critical('service', 'Critical message')

Exception Auto-Serialization

Pass an Exception directly to error() or critical() — it is serialized automatically:

try:
    raise RuntimeError('Database timeout')
except Exception as e:
    client.error('database', 'Query failed', e)

Generated metadata:

{
  "exception": {
    "type": "RuntimeError",
    "message": "Database timeout",
    "language": "python",
    "stacktrace": [{ "file": "app.py", "function": "run_query", "line": 42 }],
    "raw": "Traceback (most recent call last):\n  ..."
  }
}

Async Client

AsyncLogTideClient is the async equivalent, using aiohttp. Best used as an async context manager.

pip install logtide-sdk[async]

import asyncio
from logtide_sdk import AsyncLogTideClient, ClientOptions

async def main():
    async with AsyncLogTideClient(ClientOptions(
        api_url='http://localhost:8080',
        api_key='lp_your_api_key_here',
    )) as client:
        await client.info('my-service', 'Hello from async!')
        await client.error('my-service', 'Something failed', Exception('oops'))

asyncio.run(main())

Manual lifecycle (without context manager):

client = AsyncLogTideClient(options)
await client.start()   # starts background flush loop
try:
    await client.info('svc', 'message')
finally:
    await client.close()

All sync logging, query, stream, and metrics methods have async equivalents.

stdlib logging Integration

LogTideHandler is a standard logging.Handler — drop it into any existing logging setup.

import logging
from logtide_sdk import LogTideClient, ClientOptions, LogTideHandler

client = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
))

handler = LogTideHandler(client=client, service='my-service')
handler.setLevel(logging.WARNING)

logger = logging.getLogger(__name__)
logger.addHandler(handler)

# These are forwarded to LogTide automatically
logger.warning('Low disk space')
logger.error('Unhandled exception', exc_info=True)

Exception info is serialized with full structured stack frames when exc_info=True is used.

structlog Integration

LogTideProcessor is a structlog processor that forwards events to LogTide. structlog is not a hard dependency — install it separately if you use this integration.

import structlog
from logtide_sdk import LogTideClient, ClientOptions
from logtide_sdk.structlog import LogTideProcessor

client = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
))

structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        LogTideProcessor(client=client, service='my-service'),
        structlog.processors.JSONRenderer(),
    ],
)

log = structlog.get_logger()
log.info('user signed in', user_id=42, plan='pro')
log.error('payment failed', order_id='o_123', exc_info=True)

Non-reserved event fields (e.g. user_id, plan) are sent as metadata, the log level is mapped to LogTide's levels, and exc_info is serialized into structured exception metadata. Metadata values that aren't natively JSON-serializable (Pydantic models, dataclasses, sets, datetime, UUID, …) are handled automatically.

Trace ID Context

Manual Trace ID

client.set_trace_id('request-123')

client.info('api', 'Request received')
client.info('db', 'Querying users')
client.info('api', 'Response sent')

client.set_trace_id(None)  # clear

Scoped Trace ID (Context Manager)

with client.with_trace_id('request-456'):
    client.info('api', 'Processing in context')
    client.warn('cache', 'Cache miss')
# Trace ID automatically restored after block

Auto-Generated Trace ID

with client.with_new_trace_id():
    client.info('worker', 'Background job started')
    client.info('worker', 'Job completed')

Query API

Basic Query

from datetime import datetime, timedelta
from logtide_sdk import QueryOptions, LogLevel

result = client.query(
    QueryOptions(
        service='api-gateway',
        level=LogLevel.ERROR,
        from_time=datetime.now() - timedelta(hours=24),
        to_time=datetime.now(),
        limit=100,
        offset=0,
    )
)

print(f"Found {result.total} logs")
for log in result.logs:
    print(log)

Full-Text Search

result = client.query(QueryOptions(q='timeout', limit=50))

Get Logs by Trace ID

logs = client.get_by_trace_id('trace-123')

Aggregated Statistics

from logtide_sdk import AggregatedStatsOptions

stats = client.get_aggregated_stats(
    AggregatedStatsOptions(
        from_time=datetime.now() - timedelta(days=7),
        to_time=datetime.now(),
        interval='1h',
    )
)

for service in stats.top_services:
    print(f"{service['service']}: {service['count']} logs")

Live Streaming (SSE)

stream() runs in a background daemon thread and returns immediately with a stop function.

def handle_log(log):
    print(f"[{log['time']}] {log['level']}: {log['message']}")

stop = client.stream(
    on_log=handle_log,
    on_error=lambda e: print(f"Stream error: {e}"),
    filters={'service': 'api-gateway', 'level': 'error'},
)

# ... later, to stop:
stop()

Async streaming runs as a cancellable coroutine:

task = asyncio.create_task(client.stream(on_log=handle_log))
# ... later:
task.cancel()

Metrics

metrics = client.get_metrics()

print(f"Logs sent:              {metrics.logs_sent}")
print(f"Logs dropped:           {metrics.logs_dropped}")
print(f"Errors:                 {metrics.errors}")
print(f"Retries:                {metrics.retries}")
print(f"Avg latency:            {metrics.avg_latency_ms:.1f}ms")
print(f"Circuit breaker trips:  {metrics.circuit_breaker_trips}")

print(client.get_circuit_breaker_state())  # CLOSED | OPEN | HALF_OPEN

client.reset_metrics()

Middleware Integration

Flask

from flask import Flask
from logtide_sdk import LogTideClient, ClientOptions
from logtide_sdk.middleware import LogTideFlaskMiddleware

app = Flask(__name__)
client = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
))

LogTideFlaskMiddleware(
    app,
    client=client,
    service_name='flask-api',
    log_requests=True,
    log_responses=True,
    skip_paths=['/metrics'],
)

Django

# settings.py
from logtide_sdk import LogTideClient, ClientOptions

LOGTIDE_CLIENT = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
))
LOGTIDE_SERVICE_NAME = 'django-api'

MIDDLEWARE = [
    'logtide_sdk.middleware.LogTideDjangoMiddleware',
    # ...
]

FastAPI

from fastapi import FastAPI
from logtide_sdk import LogTideClient, ClientOptions
from logtide_sdk.middleware import LogTideFastAPIMiddleware

app = FastAPI()
client = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
))

app.add_middleware(LogTideFastAPIMiddleware, client=client, service_name='fastapi-api')

Starlette (standalone)

pip install logtide-sdk[starlette]

from starlette.applications import Starlette
from logtide_sdk import LogTideClient, ClientOptions
from logtide_sdk.middleware import LogTideStarletteMiddleware

app = Starlette()
client = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
))

app.add_middleware(LogTideStarletteMiddleware, client=client, service_name='starlette-api')

All middleware auto-logs requests, responses (with duration and status code), and errors (with serialized exception metadata). Health check paths (/health, /healthz) are skipped by default.

Examples

See the examples/ directory for complete working examples:

• basic.py - Simple usage
• advanced.py - All advanced features

Best Practices

Use Global Metadata

client = LogTideClient(ClientOptions(
    api_url='http://localhost:8080',
    api_key='lp_your_api_key_here',
    global_metadata={
        'env': os.getenv('APP_ENV', 'production'),
        'version': '2.0.0',
        'region': 'eu-west-1',
    },
))

Monitor Metrics in Production

import threading

def _monitor():
    while True:
        m = client.get_metrics()
        if m.logs_dropped > 0:
            print(f"WARNING: {m.logs_dropped} logs dropped")
        if m.circuit_breaker_trips > 0:
            print("ERROR: Circuit breaker tripped")
        time.sleep(60)

threading.Thread(target=_monitor, daemon=True).start()

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

MIT License — see LICENSE for details.

Links

• LogTide Website
• Documentation
• GitHub Issues