road24-sdk 0.1.9

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, Redis, and FastStream 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]
pip install road24-sdk[faststream]

# 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 and FastStream require 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",
  "log_level": "INFO",
  "type": "http_request",
  "service_name": "payment-service",
  "attributes": { "..." : "..." }
}

Field	Description
timestamp	UTC ISO 8601 with milliseconds
log_level	INFO, WARNING, or ERROR (based on status code)
type	http_request, db_query, redis_command, or message
service_name	Value passed to init()
attributes	Type-specific fields (see below)

Integrations

FastAPI (HTTP Input)

Logs every incoming HTTP request. 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:

{
  "timestamp": "2025-01-15T10:30:45.123Z",
  "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,
    "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. Exception responses include error_type for debugging:

{"detail": "Internal server error", "error_type": "RuntimeError"}

Binary body handling: When the request/response has a non-text content type (e.g. multipart/form-data, application/octet-stream), the body is replaced with ***BINARY-REDACTED***.

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:

{
  "timestamp": "2025-01-15T10:30:45.456Z",
  "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,
    "headers": {},
    "request_body": "",
    "response_body": ""
  }
}

Instance-level patching:

integration = HttpxLoggingIntegration()

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:

{
  "timestamp": "2025-01-15T10:30:45.789Z",
  "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:

{
  "timestamp": "2025-01-15T10:30:46.012Z",
  "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")

---

FastStream (Message Broker)

Logs every consumed/published message via FastStream middleware. Supports RabbitMQ, Kafka, NATS, and Redis brokers.

from faststream.rabbit import RabbitBroker
import road24_sdk
from road24_sdk.integrations.faststream import FastStreamLoggingIntegration

road24_sdk.init(service_name="order-worker")

broker = RabbitBroker("amqp://localhost")
FastStreamLoggingIntegration().setup(broker)

Example log (consume):

{
  "timestamp": "2025-01-15T10:30:46.200Z",
  "log_level": "INFO",
  "type": "message",
  "service_name": "order-worker",
  "attributes": {
    "direction": "consume",
    "broker": "rabbitmq",
    "queue": "orders.created",
    "duration_seconds": 0.045,
    "message_body": "{\"order_id\": 42, \"amount\": 100.50}",
    "exchange": "orders",
    "routing_key": "orders.created"
  }
}

Example log (publish):

{
  "timestamp": "2025-01-15T10:30:46.250Z",
  "log_level": "INFO",
  "type": "message",
  "service_name": "order-worker",
  "attributes": {
    "direction": "publish",
    "broker": "rabbitmq",
    "queue": "notifications.send",
    "duration_seconds": 0.012,
    "message_body": "{\"user_id\": 42, \"type\": \"order_confirmed\"}"
  }
}

When an error occurs during message processing, error_class and error_message are added and log_level is set to ERROR.

Supported brokers: rabbitmq, kafka, nats, redis. Broker type is auto-detected from the broker class name, or can be set explicitly:

from road24_sdk._types import BrokerType

FastStreamLoggingIntegration(broker_type=BrokerType.KAFKA).setup(broker)

Options:

FastStreamLoggingIntegration(
    enable_logging=True,    # Structured JSON logs
    enable_metrics=True,    # Prometheus metrics
    broker_type=None,       # Auto-detect or set explicitly
)

Standalone logger (without FastStream middleware):

from road24_sdk._types import FastStreamDirection, BrokerType
from road24_sdk.loggers.faststream import FastStreamLogger

logger = FastStreamLogger(record_metrics=True)
logger.log(
    direction=FastStreamDirection.CONSUME,
    broker=BrokerType.RABBITMQ,
    queue="orders.created",
    duration_seconds=0.045,
    message_body='{"order_id": 42}',
    exchange="orders",
    routing_key="orders.created",
)

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.

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".

FastStream Metrics

Metric	Type	Labels
faststream_messages_total	Counter	direction, broker, queue
faststream_message_duration_seconds	Histogram	direction, broker, queue

direction: "consume" or "publish". broker: "rabbitmq", "kafka", "nats", or "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)

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:

• Sensitive fields matching patterns are replaced with ***REDACTED***: password, token, secret, api_key, auth, credential, private_key, access_key, session_id, cookie, bearer, pin, cvv, card_number

• Binary bodies (non-text content types like multipart/form-data, application/octet-stream) are replaced with ***BINARY-REDACTED***

• Large bodies exceeding 10,000 characters are truncated with ...[truncated]

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.