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.

Structured Events

The SDK supports rich structured event types:

HTTP Requests

Record completed HTTP requests with timing and status:

from app_logs_ai import record_http_request, HttpRequestEvent

record_http_request(logger, HttpRequestEvent(
    method="GET",
    route="/api/users",
    status=200,
    duration_ms=125.5,
    extra={"user_id": "user123"}  # Optional context
))

Exceptions

Record exceptions with stack traces:

from app_logs_ai import record_exception, ExceptionEvent

try:
    do_something()
except Exception as e:
    record_exception(logger, ExceptionEvent(
        error=e,
        method="POST",
        route="/api/process",
        status=500,
        extra={"request_id": "req-123"}
    ))

Metrics

Periodically emit process metrics (uptime, memory):

from app_logs_ai import start_metrics_heartbeat

stop = start_metrics_heartbeat(logger, interval_ms=15000)

# Later, to stop:
stop.set()

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, record_http_request, record_exception, HttpRequestEvent
import os
import time

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

@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    try:
        response = await call_next(request)
        duration_ms = (time.time() - start) * 1000
        
        # Record structured HTTP event
        record_http_request(logger, HttpRequestEvent(
            method=request.method,
            route=request.url.path,
            status=response.status_code,
            duration_ms=duration_ms
        ))
        return response
    except Exception as exc:
        duration_ms = (time.time() - start) * 1000
        record_exception(logger, ExceptionEvent(
            error=exc,
            method=request.method,
            route=request.url.path,
            duration_ms=duration_ms
        ))
        raise

@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):
    record_exception(logger, ExceptionEvent(
        "path": str(request.url),
        "error": str(exc),
    })
    return {"error": str(exc)}