Source code for sqlspec.observability._observer

"""Statement observer primitives for SQL execution events."""

import logging
from time import time
from typing import Any

from sqlspec.observability._config import LoggingConfig, StatementObserver
from sqlspec.utils.logging import SQL_LOGGER_NAME, get_logger

__all__ = (
    "SQL_LOGGER_NAME",
    "StatementEvent",
    "create_event",
    "create_statement_observer",
    "default_statement_observer",
    "format_statement_event",
)


logger = get_logger("sqlspec.observability")

sql_logger = get_logger(SQL_LOGGER_NAME)

_DEFAULT_LOGGING_CONFIG = LoggingConfig()




class StatementEvent:
    """Structured payload describing a SQL execution."""

    __slots__ = (
        "adapter",
        "bind_key",
        "correlation_id",
        "db_system",
        "driver",
        "duration_s",
        "execution_mode",
        "is_many",
        "is_script",
        "operation",
        "parameters",
        "prepared_statement",
        "rows_affected",
        "sampled",
        "span_id",
        "sql",
        "sql_hash",
        "sql_original_length",
        "sql_truncated",
        "started_at",
        "storage_backend",
        "trace_id",
        "transaction_state",
    )



    def __init__(
        self,
        *,
        sql: str,
        parameters: Any,
        driver: str,
        adapter: str,
        bind_key: "str | None",
        db_system: "str | None",
        operation: str,
        execution_mode: "str | None",
        is_many: bool,
        is_script: bool,
        rows_affected: "int | None",
        duration_s: float,
        started_at: float,
        correlation_id: "str | None",
        storage_backend: "str | None",
        sql_hash: "str | None",
        sql_truncated: bool,
        sql_original_length: "int | None",
        transaction_state: "str | None",
        prepared_statement: "bool | None",
        trace_id: "str | None",
        span_id: "str | None",
        sampled: bool = True,
    ) -> None:
        self.sql = sql
        self.parameters = parameters
        self.driver = driver
        self.adapter = adapter
        self.bind_key = bind_key
        self.db_system = db_system
        self.operation = operation
        self.execution_mode = execution_mode
        self.is_many = is_many
        self.is_script = is_script
        self.rows_affected = rows_affected
        self.duration_s = duration_s
        self.started_at = started_at
        self.correlation_id = correlation_id
        self.storage_backend = storage_backend
        self.sql_hash = sql_hash
        self.sql_truncated = sql_truncated
        self.sql_original_length = sql_original_length
        self.transaction_state = transaction_state
        self.prepared_statement = prepared_statement
        self.trace_id = trace_id
        self.span_id = span_id
        self.sampled = sampled

def __hash__(self) -> int: # pragma: no cover msg = "StatementEvent objects are mutable and unhashable" raise TypeError(msg)

def as_dict(self) -> "dict[str, Any]": """Return event payload as a dictionary.""" return { "sql": self.sql, "parameters": self.parameters, "driver": self.driver, "adapter": self.adapter, "bind_key": self.bind_key, "db_system": self.db_system, "operation": self.operation, "execution_mode": self.execution_mode, "is_many": self.is_many, "is_script": self.is_script, "rows_affected": self.rows_affected, "duration_s": self.duration_s, "started_at": self.started_at, "correlation_id": self.correlation_id, "storage_backend": self.storage_backend, "sql_hash": self.sql_hash, "sql_truncated": self.sql_truncated, "sql_original_length": self.sql_original_length, "transaction_state": self.transaction_state, "prepared_statement": self.prepared_statement, "trace_id": self.trace_id, "span_id": self.span_id, "sampled": self.sampled, }