mssql-python#

Configuration#

class sqlspec.adapters.mssql_python.MssqlPythonConfig[source]#

Bases: SyncDatabaseConfig[Connection, MssqlPythonConnectionPool, MssqlPythonDriver]

Configuration for mssql-python synchronous database connections.

driver_type#

alias of MssqlPythonDriver

migration_tracker_type#

alias of MssqlPythonSyncMigrationTracker

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, observability_config=None, **kwargs)[source]#

create_connection()[source]#

Create a database connection.

Return type:

Connection

get_signature_namespace()[source]#

Get the signature namespace for this database configuration.

Returns a dictionary of type names to objects (classes, functions, or other callables) that should be registered with Litestar's signature namespace to prevent serialization attempts on database-specific structures.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to objects.

Connection Parameters#

class sqlspec.adapters.mssql_python.MssqlPythonConnectionParams[source]#

Bases: TypedDict

mssql-python connection parameters.

Pool Parameters#

class sqlspec.adapters.mssql_python.MssqlPythonPoolParams[source]#

Bases: MssqlPythonConnectionParams

mssql-python driver-level pooling parameters.

Driver Features#

class sqlspec.adapters.mssql_python.MssqlPythonDriverFeatures[source]#

Bases: TypedDict

mssql-python driver feature flags.

Driver#

class sqlspec.adapters.mssql_python.MssqlPythonDriver[source]#

Bases: SyncDriverAdapterBase

mssql-python sync driver.

__init__(connection, statement_config=None, driver_features=None)[source]#

Initialize driver adapter with connection and configuration.

Parameters:

• connection¶ (Connection) -- Database connection instance

• statement_config¶ (StatementConfig | None) -- Statement configuration for the driver

• driver_features¶ (dict[str, typing.Any] | None) -- Driver-specific features like extensions, secrets, and connection callbacks

• observability¶ -- Optional runtime handling lifecycle hooks, observers, and spans

property data_dictionary: MssqlPythonSyncDataDictionary#

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

dispatch_execute(cursor, statement)[source]#

Execute a single SQL statement.

Must be implemented by each driver for database-specific execution logic.

Parameters:

• cursor¶ (Cursor) -- Database cursor/connection object

• statement¶ (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data

dispatch_execute_many(cursor, statement)[source]#

Execute SQL with multiple parameter sets (executemany).

Must be implemented by each driver for database-specific executemany logic.

Parameters:

• cursor¶ (Cursor) -- Database cursor/connection object

• statement¶ (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data for the many operation

dispatch_execute_script(cursor, statement)[source]#

Execute a SQL script containing multiple statements.

Default implementation splits the script and executes statements individually. Drivers can override for database-specific script execution methods.

Parameters:

• cursor¶ (Cursor) -- Database cursor/connection object

• statement¶ (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution data including statement counts

collect_rows(cursor, fetched)[source]#

Collect rows from cursor after fetchall for the direct execution path.

Adapters should override this method to provide optimized row collection that bypasses full dispatch_execute overhead.

Parameters:

• cursor¶ (Cursor) -- Database cursor with description metadata.

• fetched¶ (list[typing.Any]) -- Rows returned from cursor.fetchall().

Return type:

tuple[list[typing.Any], list[str], int]

Returns:

Tuple of (data, column_names, row_count).

Raises:

NotImplementedError -- If the adapter does not implement this method.

resolve_rowcount(cursor)[source]#

Resolve the number of affected rows from cursor for the direct execution path.

Adapters should override this method to provide optimized rowcount resolution that bypasses full dispatch_execute overhead.

Parameters:

cursor¶ (Cursor) -- Database cursor with rowcount metadata.

Return type:

int

Returns:

Number of affected rows, or 0 when unknown.

Raises:

NotImplementedError -- If the adapter does not implement this method.

begin()[source]#

Begin a database transaction on the current connection.

Return type:

None

commit()[source]#

Commit the current transaction on the current connection.

Return type:

None

rollback()[source]#

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)[source]#

Create and return a context manager for cursor acquisition and cleanup.

Returns a context manager that yields a cursor for database operations. Concrete implementations handle database-specific cursor creation and cleanup.

Return type:

MssqlPythonCursor

handle_database_exceptions()[source]#

Handle database-specific exceptions and wrap them appropriately.

Return type:

MssqlPythonExceptionHandler

Returns:

Exception handler with deferred exception pattern for mypyc compatibility. The handler stores mapped exceptions in pending_exception rather than raising from __exit__ to avoid ABI boundary violations.

dispatch_select_stream(statement, chunk_size)[source]#

Return a native mssql-python row stream backed by fetchmany().

Return type:

Optional[SyncRowStream[dict[str, typing.Any]]]

create_savepoint(name)[source]#

Create a savepoint within the current transaction.

Return type:

None

release_savepoint(name)[source]#

Release a previously created savepoint.

Return type:

None

rollback_to_savepoint(name)[source]#

Roll back the current transaction to a previously created savepoint.

Return type:

None

set_migration_session_schema(schema)[source]#

Point the database user's default schema at the migration schema, remembering the prior one.

Return type:

None

reset_migration_session_schema()[source]#

Restore the user's default schema captured by set_migration_session_schema and commit it.

Return type:

None

has_schema(schema)[source]#

Return whether the specified schema exists.

Return type:

bool

select_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)[source]#

Execute a query and return native mssql-python Arrow results.

bulk_copy(target_table, rows, *, batch_size=0, timeout=30, column_mappings=None, keep_identity=False, check_constraints=False, table_lock=False, keep_nulls=False, fire_triggers=False, use_internal_transaction=False)[source]#

Bulk insert rows via mssql-python cursor.bulkcopy().

Return type:

MssqlPythonBulkCopyResult

load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None, batch_size=0, timeout=30, table_lock=False, check_constraints=False, fire_triggers=False, keep_identity=False, keep_nulls=False, use_internal_transaction=False, column_mappings=None)[source]#

Load Arrow tables or streams using native BulkCopy options.

Stream sources without schema metadata require explicit column mappings. Overwrite deletes existing rows after validating the source shape; errors while consuming a stream can still leave a partially completed load.

Return type:

StorageBridgeJob

load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)[source]#

Load staged artifacts from storage into SQL Server via BulkCopy.

Return type:

StorageBridgeJob

Connection Pool#

class sqlspec.adapters.mssql_python.MssqlPythonConnectionPool[source]#

Bases: object

Small SQLSpec pool facade over mssql-python's driver-level pooling.

__init__(*, connection_string, connect_kwargs=None, max_size=100, idle_timeout=600, enabled=True, on_connection_create=None)[source]#

Data Dictionary#

class sqlspec.adapters.mssql_python.MssqlPythonSyncDataDictionary[source]#

Bases: _MssqlDataDictionaryMixin, SyncDataDictionaryBase

MSSQL sync data dictionary.

dialect: ClassVar[str] = 'mssql'#

Dialect identifier. Must be defined by subclasses as a class attribute.

__init__()[source]#

get_metadata_capabilities(driver, domains=None)[source]#

Get SQL Server data-dictionary capability profile.

Return type:

MetadataCapabilityProfile

get_system_metadata_capabilities(driver, domains=None)[source]#

Get SQL Server opt-in system metadata capability disclosures.

Return type:

tuple[SystemMetadataCapability, ...]

get_version(driver)[source]#

Get SQL Server version information.

Return type:

MssqlVersionInfo | None

get_feature_flag(driver, feature)[source]#

Check whether SQL Server supports a feature.

Return type:

bool

get_optimal_type(driver, type_category)[source]#

Get optimal SQL Server type for a category.

Return type:

str

get_tables(driver, schema=None)[source]#

Get tables sorted by dependency order with catalog fallback.

Return type:

list[TableMetadata]

get_columns(driver, table=None, schema=None)[source]#

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

get_indexes(driver, table=None, schema=None)[source]#

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

get_foreign_keys(driver, table=None, schema=None)[source]#

Get foreign key metadata.

Return type:

list[ForeignKeyMetadata]

get_ddl(driver, object_name, schema=None, *, object_type='table', include_dependencies=True, prefer_native=True, redact=True)[source]#

Generate SQL Server table DDL from sys catalog rows.

Return type:

DDLResult

get_system_metadata(driver, request=None, **kwargs)[source]#

Get opt-in SQL Server system metadata with sensitive columns redacted by default.

Return type:

SystemMetadataResult

Migrations#

class sqlspec.adapters.mssql_python.MssqlPythonSyncMigrationTracker[source]#

Bases: MssqlPythonMigrationTrackerMixin, SyncMigrationTracker

T-SQL sync migration tracker.

ensure_tracking_table(driver)[source]#

Create the migration tracking table if it does not exist.

Return type:

None

record_migration(driver, version, description, execution_time_ms, checksum)[source]#

Record a successfully applied migration with T-SQL-compatible metadata.

Return type:

None

Extensions#

class sqlspec.adapters.mssql_python.events.MssqlPythonEventQueueStore[source]#

Bases: BaseEventQueueStore[MssqlPythonConfig]

T-SQL DDL hooks for the event queue store.

class sqlspec.adapters.mssql_python.litestar.MssqlPythonStore[source]#

Bases: BaseSQLSpecStore[MssqlPythonConfig]

SQL Server-backed session store using mssql-python sync sessions.

__init__(config)[source]#

Initialize the session store.

Parameters:

config¶ (MssqlPythonConfig) -- SQLSpec database configuration.

async create_table()[source]#

Create the session table if it doesn't exist.

Return type:

None

async get(key, renew_for=None)[source]#

Get a session value by key.

Return type:

bytes | None

async set(key, value, expires_in=None)[source]#

Store a session value.

Return type:

None

async delete(key)[source]#

Delete a session by key.

Return type:

None

async delete_all()[source]#

Delete all sessions from the store.

Return type:

None

async exists(key)[source]#

Check if a session key exists and is not expired.

Return type:

bool

async expires_in(key)[source]#

Get the time in seconds until the session expires.

Return type:

int | None

async delete_expired()[source]#

Delete all expired sessions.

Return type:

int

class sqlspec.adapters.mssql_python.adk.MssqlPythonADKStore[source]#

Bases: BaseSyncADKStore[MssqlPythonConfig]

Synchronous mssql-python ADK session/event store.

__init__(config)[source]#

Initialize the ADK store.

Parameters:

config¶ (MssqlPythonConfig) -- SQLSpec database configuration.

Notes

Reads configuration from config.extension_config["adk"]: - session_table: Sessions table name (default: "adk_session") - events_table: Events table name (default: "adk_event") - app_state_table: App-scoped state table name (default: "adk_app_state") - user_state_table: User-scoped state table name (default: "adk_user_state") - metadata_table: Internal metadata table name (default: "adk_internal_metadata") - owner_id_column: Optional owner FK column DDL (default: None)

create_tables()[source]#

Create ADK tables (idempotent T-SQL) and DD-gated indexes.

Return type:

None

create_session(session_id, app_name, user_id, state, owner_id=None)[source]#

Create a new ADK session.

Return type:

StoredSession

get_session(app_name, user_id, session_id, *, renew_for=None)[source]#

Return a scoped session or None if absent.

Return type:

StoredSession | None

update_session_state(app_name, user_id, session_id, state)[source]#

Replace a session's durable state.

Return type:

None

list_sessions(app_name, user_id=None, *, order_by='update_time', descending=True, limit=None, offset=None)[source]#

List ADK sessions for an application, optionally scoped to a user.

Return type:

list[StoredSession]

delete_session(app_name, user_id, session_id)[source]#

Delete a session. Event rows cascade through the FK.

Return type:

None

append_event(event_record)[source]#

Append an event to a session.

Return type:

None

append_event_and_update_state(event_record, app_name, user_id, session_id, state, *, app_state=None, user_state=None)[source]#

Atomically append an event and update durable session/scoped state.

Return type:

StoredSession

get_events(app_name, user_id, session_id, after_timestamp=None, limit=None)[source]#

Return events for a scoped session ordered by event timestamp.

Return type:

list[StoredEvent]

delete_expired_events(before, app_name=None)[source]#

Delete events older than before.

Return type:

int

delete_idle_sessions(updated_before, app_name=None)[source]#

Delete sessions whose update_time is older than updated_before.

Return type:

int

delete_idle_user_states(updated_before, app_name=None)[source]#

Delete user state rows whose update_time is older than updated_before.

Return type:

int

get_app_state(app_name)[source]#

Return app-scoped state.

Return type:

dict[str, typing.Any] | None

get_user_state(app_name, user_id)[source]#

Return user-scoped state.

Return type:

dict[str, typing.Any] | None

upsert_app_state(app_name, state)[source]#

Insert or replace app-scoped state.

Return type:

None

upsert_user_state(app_name, user_id, state)[source]#

Insert or replace user-scoped state.

Return type:

None

get_metadata(key)[source]#

Return an ADK metadata value.

Return type:

str | None

set_metadata(key, value)[source]#

Set an ADK metadata value.

Return type:

None

class sqlspec.adapters.mssql_python.adk.MssqlPythonADKMemoryStore[source]#

Bases: BaseSyncADKMemoryStore[MssqlPythonConfig]

SQL Server ADK memory store using mssql-python.

__init__(config)[source]#

Initialize the ADK memory store.

Parameters:

config¶ (MssqlPythonConfig) -- SQLSpec database configuration.

create_tables()[source]#

Create the memory table (idempotent T-SQL) and DD-gated indexes.

Return type:

None

insert_memory_entries(entries, owner_id=None)[source]#

Bulk insert memory entries with event-id deduplication.

Return type:

int

search_entries(query, app_name, user_id, limit=None, scope_filter='all', embedding=None)[source]#

Search memory entries by text query.

Return type:

list[StoredMemory]

delete_entries_by_session(session_id)[source]#

Delete all memory entries for a specific session.

Return type:

int

delete_entries_older_than(days, app_name=None, scope=None)[source]#

Delete memory entries older than the retention window.

Return type:

int

Extension Settings#

Use these types inside extension_config["adk"].

class sqlspec.adapters.mssql_python.adk.MssqlPythonADKConfig[source]#

Bases: ADKConfig

mssql-python ADK extension settings.

native_json: NotRequired[bool]#

Force native SQL Server JSON columns when True, or NVARCHAR(MAX) when False.