Source code for sqlspec.adapters.mssql_python.config

"""mssql-python database configuration."""

import asyncio
import warnings
from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast

from typing_extensions import NotRequired

from sqlspec.adapters.mssql_python._typing import (
    MSSQL_PYTHON_MODULE,
    MssqlPythonAsyncSessionContext,
    MssqlPythonConnection,
    MssqlPythonSessionContext,
)
from sqlspec.adapters.mssql_python.core import apply_driver_features, build_connection_config, default_statement_config
from sqlspec.adapters.mssql_python.driver import MssqlPythonAsyncDriver, MssqlPythonDriver
from sqlspec.adapters.mssql_python.migrations import MssqlPythonAsyncMigrationTracker, MssqlPythonSyncMigrationTracker
from sqlspec.config import AsyncDatabaseConfig, ExtensionConfigs, SyncDatabaseConfig
from sqlspec.driver._async import AsyncPoolConnectionContext, AsyncPoolSessionFactory
from sqlspec.driver._sync import SyncPoolConnectionContext, SyncPoolSessionFactory
from sqlspec.utils.config_tools import normalize_connection_config

if TYPE_CHECKING:
    from collections.abc import Callable
    from types import TracebackType

    from sqlspec.core import StatementConfig
    from sqlspec.observability import ObservabilityConfig

__all__ = (
    "MssqlPythonAsyncConfig",
    "MssqlPythonConfig",
    "MssqlPythonConnectionParams",
    "MssqlPythonConnectionPool",
    "MssqlPythonDriverFeatures",
    "MssqlPythonPoolParams",
)

_POOLING_PARAMS: "tuple[int, int, bool] | None" = None


class MssqlPythonConnectionParams(TypedDict):
    """mssql-python connection parameters."""

    connection_string: NotRequired[str]
    server: NotRequired[str]
    address: NotRequired[str]
    addr: NotRequired[str]
    port: NotRequired[int]
    database: NotRequired[str]
    db: NotRequired[str]
    uid: NotRequired[str]
    user: NotRequired[str]
    username: NotRequired[str]
    pwd: NotRequired[str]
    password: NotRequired[str]
    authentication: NotRequired[str]
    trusted_connection: NotRequired[bool | str]
    trust_server_certificate: NotRequired[bool | str]
    trust: NotRequired[bool | str]
    encrypt: NotRequired[bool | str]
    hostname_in_certificate: NotRequired[str]
    hostnameincertificate: NotRequired[str]
    server_certificate: NotRequired[str]
    servercertificate: NotRequired[str]
    server_spn: NotRequired[str]
    serverspn: NotRequired[str]
    multi_subnet_failover: NotRequired[bool | str]
    multisubnetfailover: NotRequired[bool | str]
    application_intent: NotRequired[str]
    applicationintent: NotRequired[str]
    connect_retry_count: NotRequired[int]
    connectretrycount: NotRequired[int]
    connect_retry_interval: NotRequired[int]
    connectretryinterval: NotRequired[int]
    keep_alive: NotRequired[int]
    keepalive: NotRequired[int]
    keep_alive_interval: NotRequired[int]
    keepaliveinterval: NotRequired[int]
    ip_address_preference: NotRequired[str]
    ipaddresspreference: NotRequired[str]
    packet_size: NotRequired[int]
    packetsize: NotRequired[int]
    connection_timeout: NotRequired[int]
    login_timeout: NotRequired[int]
    command_timeout: NotRequired[int]
    autocommit: NotRequired[bool]
    attrs_before: NotRequired[dict[int, int | str | bytes]]
    timeout: NotRequired[int]
    native_uuid: NotRequired[bool]
    extra: NotRequired[dict[str, Any]]
class MssqlPythonPoolParams(MssqlPythonConnectionParams): """mssql-python driver-level pooling parameters.""" pool_size: NotRequired[int] pool_idle_timeout: NotRequired[int] pool_enabled: NotRequired[bool]