road24-sdk 0.1.5

Shared logging and metrics library for Road24 FastAPI microservices

road24-sdk

Shared Python SDK for structured JSON logging and Prometheus metrics in Road24 microservices.

Provides auto-instrumentation for FastAPI, httpx, SQLAlchemy, and Redis with zero boilerplate.

Installation

# Core only
pip install road24-sdk

# With specific integrations
pip install road24-sdk[fastapi]
pip install road24-sdk[httpx]
pip install road24-sdk[sqlalchemy]
pip install road24-sdk[redis]

# All integrations
pip install road24-sdk[all]

Requires Python 3.12+.

Quick Start

import road24_sdk
from road24_sdk.integrations.fastapi import FastApiLoggingIntegration
from road24_sdk.integrations.httpx import HttpxLoggingIntegration
from road24_sdk.integrations.sqlalchemy import SqlalchemyLoggingIntegration
from road24_sdk.integrations.redis import RedisLoggingIntegration

road24_sdk.init(
    service_name="payment-service",
    log_level="INFO",
    integrations=[
        FastApiLoggingIntegration(),
        HttpxLoggingIntegration(),
        SqlalchemyLoggingIntegration(),
        RedisLoggingIntegration(),
    ],
)

After init(), all new httpx clients, SQLAlchemy engines, and Redis clients are automatically instrumented. FastAPI requires one extra call:

from fastapi import FastAPI

app = FastAPI()
FastApiLoggingIntegration().setup(app)

Log Format

Every log line is a single JSON object written to stdout:

{
  "timestamp": "2025-01-15T10:30:45.123Z",
  "trace_id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
  "log_level": "INFO",
  "type": "http_request",
  "service_name": "payment-service",
  "attributes": { "..." : "..." }
}

Field	Description
timestamp	UTC ISO 8601 with milliseconds
trace_id	32-char hex from x-trace-id header or auto-generated
log_level	INFO, WARNING, or ERROR (based on status code)
type	http_request, db_query, or redis_command
service_name	Value passed to init()
attributes	Type-specific fields (see below)

Integrations

FastAPI (HTTP Input)

Logs every incoming HTTP request. Adds x-trace-id response header. Catches and formats exceptions.

from fastapi import FastAPI
import road24_sdk
from road24_sdk.integrations.fastapi import FastApiLoggingIntegration

road24_sdk.init(
    service_name="user-service",
    integrations=[FastApiLoggingIntegration()],
)

app = FastAPI()
FastApiLoggingIntegration().setup(app)

Example log input:

{
  "timestamp": "2025-01-15T10:30:45.123Z",
  "trace_id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
  "log_level": "INFO",
  "type": "http_request",
  "service_name": "user-service",
  "attributes": {
    "direction": "input",
    "method": "GET",
    "url": "https://user-service.local/api/v1/users/42",
    "target": "/api/v1/users/42",
    "status_code": 200,
    "duration_seconds": 0.023,
    "request_headers": {},
    "request_body": "",
    "response_body": ""
  }
}

Log level mapping: 2xx/3xx -> INFO, 4xx -> WARNING, 5xx -> ERROR.

When an exception occurs, error_class and error_message are added to attributes:

{
  "attributes": {
    "direction": "input",
    "method": "GET",
    "url": "https://user-service.local/api/v1/process",
    "target": "/api/v1/process",
    "status_code": 500,
    "duration_seconds": 0.012,
    "request_headers": {},
    "request_body": "",
    "response_body": "",
    "error_class": "RuntimeError",
    "error_message": "Something went wrong"
  }
}

Exception responses include trace_id for debugging:

{"detail": "User not found", "trace_id": "a1b2c3d4e5f6a1b2..."}

Options:

FastApiLoggingIntegration(
    enable_logging=True,           # Structured JSON logs
    enable_metrics=True,           # Prometheus metrics
    enable_exception_handlers=True # Catch and format exceptions
)

Skipped paths: /metrics, /docs, /redoc, /openapi.json

---

httpx (HTTP Output)

Logs every outgoing HTTP request made via httpx.AsyncClient.

import httpx
import road24_sdk
from road24_sdk.integrations.httpx import HttpxLoggingIntegration

road24_sdk.init(
    service_name="payment-service",
    integrations=[HttpxLoggingIntegration()],
)

# All new clients are auto-instrumented after init()
async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com/users/42")

Example log output:

{
  "timestamp": "2025-01-15T10:30:45.456Z",
  "trace_id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
  "log_level": "INFO",
  "type": "http_request",
  "service_name": "payment-service",
  "attributes": {
    "direction": "output",
    "method": "GET",
    "url": "https://api.example.com/users/42",
    "target": "/users/42",
    "status_code": 200,
    "duration_seconds": 0.145,
    "request_headers": {},
    "request_body": "",
    "response_body": ""
  }
}

Alternative usage (instance-level):

integration = HttpxLoggingIntegration()

# Patch a specific client
client = httpx.AsyncClient()
integration.setup(client)

# Or use the factory method
client = integration.create_client(base_url="https://api.example.com")

---

SQLAlchemy (Database)

Logs every database query with operation type, table name, and duration.

from sqlalchemy.ext.asyncio import create_async_engine
import road24_sdk
from road24_sdk.integrations.sqlalchemy import SqlalchemyLoggingIntegration

road24_sdk.init(
    service_name="order-service",
    integrations=[SqlalchemyLoggingIntegration()],
)

# All engines are auto-instrumented after init()
engine = create_async_engine("postgresql+asyncpg://...")

Example log output:

{
  "timestamp": "2025-01-15T10:30:45.789Z",
  "trace_id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
  "log_level": "INFO",
  "type": "db_query",
  "service_name": "order-service",
  "attributes": {
    "direction": "sqlalchemy",
    "operation": "select",
    "table": "orders",
    "duration_seconds": 0.008,
    "statement": "SELECT orders.id, orders.total FROM orders WHERE orders.user_id = ?"
  }
}

Supported operations: select, insert, update, delete, other. Table name is auto-extracted from SQL. Statements longer than 500 characters are truncated.

Instance-level patching:

integration = SqlalchemyLoggingIntegration()
integration.setup(engine)  # Works with both sync and async engines

---

Redis

Logs every Redis command with operation, key, and duration.

from redis.asyncio import Redis
import road24_sdk
from road24_sdk.integrations.redis import RedisLoggingIntegration

road24_sdk.init(
    service_name="cache-service",
    integrations=[RedisLoggingIntegration()],
)

# All Redis clients are auto-instrumented after init()
redis = Redis(host="localhost")
await redis.set("user:42:name", "Alice")
await redis.get("user:42:name")

Example log output:

{
  "timestamp": "2025-01-15T10:30:46.012Z",
  "trace_id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
  "log_level": "INFO",
  "type": "redis_command",
  "service_name": "cache-service",
  "attributes": {
    "direction": "redis",
    "operation": "set",
    "key": "user:42:name",
    "duration_seconds": 0.002,
    "command_args": "value=Alice"
  }
}

Supported operations: get, set, delete, expire, hget, hset, lpush, rpush, lpop, rpop, publish, subscribe, other.

Alternative: explicit wrapper with typed methods:

from road24_sdk.loggers.redis import LoggedRedis

logged = LoggedRedis(redis_client, record_metrics=True)
await logged.set("key", "value", ex=300)
await logged.get("key")
await logged.delete("key")
await logged.hset("hash", "field", "value")
await logged.lpush("list", "item1", "item2")

Prometheus Metrics

All integrations automatically record Prometheus metrics. Expose them with any Prometheus-compatible endpoint.

HTTP Metrics

Metric	Type	Labels
http_requests_total	Counter	method, domain, status_code, direction
http_request_duration_seconds	Histogram	method, domain, status_code, direction

direction: "input" (FastAPI) or "output" (httpx). domain: extracted hostname from URL (e.g. api.example.com).

Database Metrics

Metric	Type	Labels
db_queries_total	Counter	operation, direction
db_query_duration_seconds	Histogram	operation, direction

operation: select, insert, update, delete, other. direction: "sqlalchemy".

Redis Metrics

Metric	Type	Labels
redis_commands_total	Counter	operation, direction
redis_command_duration_seconds	Histogram	operation, direction

direction: "redis".

URL and Key Normalization

Dynamic segments are replaced with {id} to prevent metric cardinality explosion:

Type	Raw	Normalized
URL path	/api/v1/users/42	/api/v1/users/{id}
URL path	/orders/550e8400-e29b-41d4-a716-446655440000	/orders/{id}
Redis key	user:42:sessions	user:{id}:sessions
Redis key	cache:abc123def456	cache:abc123def456 (short hex not replaced)

Trace ID Propagation

1. Incoming request has x-trace-id header -> SDK uses it
2. No header -> SDK generates a random 32-char hex ID
3. Response includes x-trace-id header
4. All logs within the request share the same trace_id
5. Trace context is propagated across async operations via ContextVar

Configuration

road24_sdk.init(
    service_name="my-service",  # Appears in every log line
    log_level="INFO",           # DEBUG, INFO, WARNING, ERROR
    debug=False,                # True = less library silencing
    log_format="json",          # Log output format
    integrations=[...],         # List of integrations to activate
)

Per-Integration Options

Every integration accepts enable_logging and enable_metrics:

# Metrics only, no log lines
HttpxLoggingIntegration(enable_logging=False, enable_metrics=True)

# Logs only, no metrics
SqlalchemyLoggingIntegration(enable_logging=True, enable_metrics=False)

Body Sanitization

Request/response bodies are automatically sanitized. Fields matching sensitive patterns are replaced with ***REDACTED***:

password, token, secret, api_key, auth, credential, private_key, access_key, session_id, cookie, bearer, pin, cvv, card_number

Library Silencing

In production (debug=False), noisy library loggers are silenced: uvicorn, sqlalchemy, httpx, httpcore, asyncio.

In debug mode (debug=True), only uvicorn.access and asyncio are silenced.