app-logs-ai

Python SDK for AI Application Logs. Batches log entries and ships them to the ingest API.

Install

pip install app-logs-ai

Quick Start

from app_logs_ai import create_logger

logger = create_logger({
    "api_key": "your-api-key-here",
})

logger.info("Server started", {"port": 8000})
logger.error("Database connection failed", {"error": "ECONNREFUSED"})

Configuration

Options passed to create_logger():

• api_key (required): Your project's API key
• endpoint: Ingest endpoint URL (defaults to https://api-app-logs.up.railway.app/v1/ingest)
• batch_size: Flush when this many entries are buffered (default: 20)
• flush_interval_ms: Flush at most this often in milliseconds (default: 2000)
• source: Logical source label (default: "backend")
• max_buffer_size: Max entries kept in memory (default: 10000)
• max_retries: Send attempts per flush before re-queueing (default: 4)
• retry_backoff_ms: Base delay for exponential backoff (default: 500)
• gzip_threshold: Compress payloads larger than this (bytes, 0 to disable, default: 1024)
• persist_path: File path for crash-durable buffering (optional)
• flush_on_exit: Auto-flush on SIGTERM/SIGINT (default: True)
• fallback_to_stderr: Print undelivered logs to stderr (default: True)
• enable_metrics_heartbeat: Send periodic metrics (default: True)
• metrics_heartbeat_ms: Metrics interval in milliseconds (default: 15000)

Log Levels

The logger supports these levels:

• debug(message, attributes)
• info(message, attributes)
• warn(message, attributes)
• error(message, attributes)
• fatal(message, attributes)

The generic log(level, message, attributes) method is also available.

Durability

• Batching: Entries are buffered and sent every 2 seconds or when 20 entries accumulate
• Retries: Failed sends are retried with exponential backoff
• Persistence: With persist_path set, undelivered entries survive process restarts
• Graceful shutdown: Automatic flush on exit; call logger.flush() to ensure delivery

Example with FastAPI

See examples/fastapi-app/ for a complete example application.

from fastapi import FastAPI
from app_logs_ai import create_logger
import os

app = FastAPI()
logger = create_logger({"api_key": os.environ.get("APP_LOGS_KEY", "test-key")})

@app.get("/api/users")
async def get_users():
    logger.info("Fetching users", {"endpoint": "/api/users"})
    return {"users": []}

@app.exception_handler(Exception)
async def exception_handler(request, exc):
    logger.error("Unhandled exception", {
        "path": str(request.url),
        "error": str(exc),
    })
    return {"error": str(exc)}