Observability#

Instrumentation#

To enable observability features, you can attach an ObservabilityConfig to your database configuration or SQLSpec instance, or declare settings via extension_config.

Custom Statement Observers#

Create a custom observer to capture StatementEvent objects for logging, metrics, or alerting. Each observer is a callable that receives a StatementEvent after every SQL execution. Custom observers conform to the public sqlspec.observability.StatementObserver protocol.

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.observability import ObservabilityConfig, StatementEvent

# Collect events for testing or custom processing
captured_events: list[StatementEvent] = []

def my_observer(event: StatementEvent) -> None:
    """Custom observer that captures SQL events."""
    captured_events.append(event)
    if event.duration_s > 1.0:
        print(f"SLOW QUERY ({event.duration_s:.2f}s): {event.sql[:80]}")

# Wire the observer into the config
observability = ObservabilityConfig(statement_observers=(my_observer,), print_sql=False)

spec = SQLSpec(observability_config=observability)
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

with spec.provide_session(config) as session:
    session.execute("create table items (id integer primary key, name text)")
    session.execute("insert into items (name) values ('Widget')")
    session.select("select * from items")

# Observer received events for each execution
print(f"Captured {len(captured_events)} events")
for event in captured_events:
    print(f"  {event.operation} - {event.driver} - {event.duration_s:.4f}s")

StatementEvent fields include:

• sql -- the executed SQL string

• parameters -- bound parameters

• driver -- driver class name

• operation -- SQL operation type (SELECT, INSERT, etc.)

• duration_s -- execution time in seconds

• rows_affected -- number of rows affected

• correlation_id -- request correlation ID (if set)

• db_system -- database system identifier

• trace_id / span_id -- OpenTelemetry context (if enabled)

OpenTelemetry Tracing#

SQLSpec can automatically generate OpenTelemetry spans for every SQL query and migration command. This enables end-to-end distributed tracing across microservices and storage operations.

To enable tracing programmatically, use the enable_tracing helper:

from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.extensions.otel import enable_tracing
from sqlspec.observability import ObservabilityConfig

# Create a configuration with tracing enabled
observability = enable_tracing(
    base_config=ObservabilityConfig(),
    resource_attributes={"service.name": "my-service"},
    enable_spans=True,
)

# Attach to database configuration
config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    observability_config=observability,
)

Alternatively, configure OpenTelemetry declaratively via extension_config:

from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    extension_config={
        "otel": {
            "enabled": True,
            "resource_attributes": {"service.name": "my-service"},
            "enable_spans": True,
        }
    },
)

Generated spans include standard semantic attributes: - db.system (e.g., "postgresql", "sqlite", "duckdb") - db.statement (the sanitized SQL query) - db.operation (e.g., "SELECT", "INSERT", "UPDATE") - Migration command spans under command.upgrade, command.downgrade, etc.

Prometheus Metrics#

Expose Prometheus metrics for queries, execution duration histograms, and affected row counts.

To enable metrics programmatically, use the enable_metrics helper:

from sqlspec.extensions.prometheus import enable_metrics
from sqlspec.observability import ObservabilityConfig

# Enable Prometheus metrics
observability = enable_metrics(
    base_config=ObservabilityConfig(),
    namespace="myapp_sql",       # Metric prefix (default: 'sqlspec')
    subsystem="driver",          # Metric subsystem (default: 'driver')
    label_names=("db_system", "operation"),
)

Or declaratively via extension_config:

from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    extension_config={
        "prometheus": {
            "enabled": True,
            "namespace": "myapp_sql",
            "subsystem": "driver",
            "label_names": ("db_system", "operation"),
        }
    },
)

Exposed metrics: - {namespace}_{subsystem}_query_total: Counter of executed queries. - {namespace}_{subsystem}_query_duration_seconds: Histogram of execution duration. - {namespace}_{subsystem}_query_rows: Histogram of rows affected.

Pass subsystem="" if you wish to omit the subsystem prefix (e.g., myapp_sql_query_total).

OpenTelemetry tracing is span-based and does not register a statement observer. Use statement_observers for callback-style integrations such as metrics, audit sinks, or custom log emission; use TelemetryConfig or enable_tracing() for OpenTelemetry spans.

Lifecycle Hooks#

Lifecycle hooks allow applications to intercept connection pool events, session lifecycles, and statement execution phases:

from sqlspec.observability import ObservabilityConfig

def on_connection_created(connection: object) -> None:
    print(f"New connection established: {connection}")

def on_query_start(sql: str, params: dict) -> None:
    print(f"Starting execution: {sql}")

def on_query_error(error: Exception, sql: str, params: dict) -> None:
    print(f"Query error occurred: {error} in {sql}")

observability = ObservabilityConfig(
    lifecycle={
        "on_connection_create": [on_connection_created],
        "on_query_start": [on_query_start],
        "on_error": [on_query_error],
    }
)

Supported lifecycle hooks include:

• on_connection_create / on_connection_destroy: Connection lifecycle events.

• on_pool_create / on_pool_destroying / on_pool_destroy: Pool lifecycle events.

• on_session_start / on_session_end: Session boundary events.

• on_query_start: Invoked before statement execution with (sql, params).

• on_query_complete: Invoked after successful statement execution with (sql, params, result).

• on_error: Invoked on execution failure with (exception, sql, params).

Data Redaction#

Sensitive parameter and literal masking can be configured via RedactionConfig:

from sqlspec.observability import ObservabilityConfig, RedactionConfig

observability = ObservabilityConfig(
    redaction=RedactionConfig(
        mask_parameters=True,
        mask_literals=True,
        parameter_allow_list=("user_id", "status"),
    )
)

SQLCommenter#

SQLSpec supports the Google SQLCommenter spec, which appends structured /* key='value' */ comments to SQL statements for query attribution. This lets database logs, query planners, and APM tools trace queries back to the application code that issued them.

Enable it on your StatementConfig:

from sqlspec.core import StatementConfig

config = StatementConfig(enable_sqlcommenter=True)

The db_driver attribute is set automatically from the adapter's dialect (e.g. postgresql, sqlite, mysql). This produces queries like:

SELECT * FROM users /* db_driver='postgresql',framework='litestar',route='%2Fusers' */

All framework extensions automatically register SQLCommenter middleware that populates request-scoped attributes (route, action, framework, and controller for Litestar). The Litestar plugin installs its SQLCommenter and correlation middleware at the outermost position of the middleware stack, ahead of any middleware the application itself registers, so a correlation ID is available even for requests that authentication or session middleware rejects. To include these in the SQL comments, enable sqlcommenter_enable_context:

config = StatementConfig(
    enable_sqlcommenter=True,
    sqlcommenter_enable_context=True,
)

You can add custom static attributes that appear on every query:

config = StatementConfig(
    enable_sqlcommenter=True,
    sqlcommenter_attributes={"app_name": "my-service", "deployment": "prod"},
)

OpenTelemetry traceparent can be auto-populated from the current span:

config = StatementConfig(
    enable_sqlcommenter=True,
    sqlcommenter_enable_traceparent=True,
)

To disable the middleware for a specific extension, set enable_sqlcommenter_middleware to False:

extension_config={"litestar": {"enable_sqlcommenter_middleware": False}}

Correlation Tracking#

SQLSpec can track a correlation ID across your application to link SQL logs with specific requests.

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

spec = SQLSpec()
spec.add_config(
    SqliteConfig(
        connection_config={"database": str(tmp_path / "observability.db")},
        extension_config={"litestar": {"enable_correlation_middleware": True}},
    )
)

with CorrelationContext.context("req-123") as correlation_id:
    print(correlation_id)

Logging & Sampling#

You can configure detailed SQL logging and sampling to reduce noise in production.

from sqlspec.observability import ObservabilityConfig, SamplingConfig

sampling = SamplingConfig(sample_rate=0.1, force_sample_on_error=True, deterministic=True)
observability = ObservabilityConfig(sampling=sampling, print_sql=False)

Logger Hierarchy#

SQLSpec uses a hierarchical logger namespace that allows fine-grained control over log levels. This enables you to configure SQL execution logs independently from internal debug logs.

sqlspec                              # Root logger for all SQLSpec logs
├── sqlspec.sql                      # SQL execution logs (SELECT, INSERT, etc.)
├── sqlspec.pool                     # Connection pool operations (acquire, release, recycle)
├── sqlspec.cache                    # Cache operations (hit, miss, evict)
├── sqlspec.driver                   # Driver base class operations
├── sqlspec.core
│   ├── sqlspec.core.compiler        # SQL compilation
│   ├── sqlspec.core.splitter        # Statement splitting
│   └── sqlspec.core.statement       # Statement processing
├── sqlspec.adapters
│   ├── sqlspec.adapters.asyncpg     # AsyncPG adapter
│   ├── sqlspec.adapters.psycopg     # Psycopg adapter
│   └── ...                          # Other adapters
└── sqlspec.observability
    └── sqlspec.observability.lifecycle  # Lifecycle events

Common Configuration Patterns:

import logging

# Pattern 1: Debug cache while keeping SQL logs at INFO
logging.getLogger("sqlspec").setLevel(logging.WARNING)
logging.getLogger("sqlspec.cache").setLevel(logging.DEBUG)

# Pattern 2: Show SQL queries, suppress internal logs
logging.getLogger("sqlspec").setLevel(logging.WARNING)
logging.getLogger("sqlspec.sql").setLevel(logging.INFO)

# Pattern 3: Debug connection pool while keeping other logs quiet
logging.getLogger("sqlspec").setLevel(logging.WARNING)
logging.getLogger("sqlspec.pool").setLevel(logging.DEBUG)

# Pattern 4: Disable all SQLSpec logs
logging.getLogger("sqlspec").setLevel(logging.CRITICAL)

Using the SQL_LOGGER_NAME constant:

from sqlspec.observability import SQL_LOGGER_NAME

# Configure SQL logging level
logging.getLogger(SQL_LOGGER_NAME).setLevel(logging.INFO)

cache.miss extra_fields={'cache_namespace': 'statement', 'cache_size': 0}
cache.hit  extra_fields={'cache_namespace': 'expression', 'cache_size': 42}

SELECT  driver=AsyncpgDriver bind_key=primary duration_ms=3.5 rows=5 sql='SELECT ...'
INSERT  driver=AsyncpgDriver bind_key=primary duration_ms=1.2 rows=1 sql='INSERT ...'

pool.connection.recycle  adapter=sqlite pool_id=a1b2c3d4 database=:memory: reason=exceeded_recycle_time
pool.connection.close.timeout  adapter=aiosqlite pool_id=e5f6g7h8 connection_id=abc timeout_seconds=10.0
pool.extension.load.failed  adapter=duckdb pool_id=i9j0k1l2 extension=httpfs error='...'

from sqlspec.utils.logging import POOL_LOGGER_NAME

# Enable pool debug logs for connection troubleshooting
logging.getLogger(POOL_LOGGER_NAME).setLevel(logging.DEBUG)

Cloud Log Formatters#

For cloud environments (like GCP or AWS), structured logging is essential.

from sqlspec.observability import AWSLogFormatter, GCPLogFormatter, ObservabilityConfig

gcp_logs = ObservabilityConfig(cloud_formatter=GCPLogFormatter())
aws_logs = ObservabilityConfig(cloud_formatter=AWSLogFormatter())

Related Guides#

• Observability for API details.