mssql-python

Sync + async SQL Server adapter built on Microsoft's official mssql-python driver. Ships a T-SQL data dictionary, a Litestar session store, an events queue store, and migrations tracker. The SQL splitter also gains GO batch-separator handling so multi-batch T-SQL scripts execute correctly.

Sync Configuration

class sqlspec.adapters.mssql_python.MssqlPythonConfig

Bases: SyncDatabaseConfig[Connection, MssqlPythonConnectionPool, MssqlPythonDriver]

Configuration for mssql-python synchronous database connections.

driver_type

alias of MssqlPythonDriver

connection_type

alias of Connection

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)

create_connection()

Create a database connection.

Return type:

Connection

get_signature_namespace()

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.

Async Configuration

class sqlspec.adapters.mssql_python.MssqlPythonAsyncConfig

Bases: AsyncDatabaseConfig[Connection, MssqlPythonConnectionPool, MssqlPythonAsyncDriver]

Configuration for mssql-python async database connections.

driver_type

alias of MssqlPythonAsyncDriver

connection_type

alias of Connection

migration_tracker_type

alias of MssqlPythonAsyncMigrationTracker

__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)

async create_connection()

Create a database connection.

Return type:

Connection

get_signature_namespace()

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.

class sqlspec.adapters.mssql_python.MssqlPythonConnectionParams

Bases: TypedDict

mssql-python connection parameters.

class sqlspec.adapters.mssql_python.MssqlPythonPoolParams

Bases: MssqlPythonConnectionParams

mssql-python driver-level pooling parameters.

class sqlspec.adapters.mssql_python.MssqlPythonDriverFeatures

Bases: TypedDict

mssql-python driver feature flags.

Sync Driver

class sqlspec.adapters.mssql_python.MssqlPythonDriver

Bases: SyncDriverAdapterBase

mssql-python sync driver.

__init__(connection, statement_config=None, driver_features=None)

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)

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)

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)

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)

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)

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()

Begin a database transaction on the current connection.

Return type:

None

commit()

Commit the current transaction on the current connection.

Return type:

None

rollback()

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)

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()

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.

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

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

bulk_copy(target_table, rows, *, batch_size=64000, timeout=3600, column_mappings=None, keep_identity=False, check_constraints=True, table_lock=False, keep_nulls=False, fire_triggers=False, use_internal_transaction=False)

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

Return type:

int

Async Driver

class sqlspec.adapters.mssql_python.MssqlPythonAsyncDriver

Bases: AsyncDriverAdapterBase

Async wrapper around mssql-python's sync DB-API via asyncio.to_thread.

__init__(connection, statement_config=None, driver_features=None)

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: MssqlPythonAsyncDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

async dispatch_execute(cursor, statement)

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

async dispatch_execute_many(cursor, statement)

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

async dispatch_execute_script(cursor, statement)

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)

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)

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.

async begin()

Begin a database transaction on the current connection.

Return type:

None

async commit()

Commit the current transaction on the current connection.

Return type:

None

async rollback()

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)

Create and return an async context manager for cursor acquisition and cleanup.

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

Return type:

MssqlPythonAsyncCursor

handle_database_exceptions()

Handle database-specific exceptions and wrap them appropriately.

Return type:

MssqlPythonAsyncExceptionHandler

Returns:

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

async select_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)

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

async bulk_copy(target_table, rows, *, batch_size=64000, timeout=3600, column_mappings=None, keep_identity=False, check_constraints=True, table_lock=False, keep_nulls=False, fire_triggers=False, use_internal_transaction=False)

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

Return type:

int

Connection Pool

class sqlspec.adapters.mssql_python.MssqlPythonConnectionPool

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)

Data Dictionaries

class sqlspec.adapters.mssql_python.MssqlPythonSyncDataDictionary

Bases: _MssqlDataDictionaryMixin, SyncDataDictionaryBase

MSSQL sync data dictionary.

dialect: ClassVar[str] = 'mssql'

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

__init__()

get_version(driver)

Get SQL Server version information.

Return type:

MssqlVersionInfo | None

get_feature_flag(driver, feature)

Check whether SQL Server supports a feature.

Return type:

bool

get_optimal_type(driver, type_category)

Get optimal SQL Server type for a category.

Return type:

str

get_tables(driver, schema=None)

Get tables sorted by dependency order with catalog fallback.

Return type:

list[TableMetadata]

get_columns(driver, table=None, schema=None)

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

get_indexes(driver, table=None, schema=None)

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

get_foreign_keys(driver, table=None, schema=None)

Get foreign key metadata.

Return type:

list[ForeignKeyMetadata]

class sqlspec.adapters.mssql_python.MssqlPythonAsyncDataDictionary

Bases: _MssqlDataDictionaryMixin, AsyncDataDictionaryBase

MSSQL async data dictionary.

dialect: ClassVar[str] = 'mssql'

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

__init__()

async get_version(driver)

Get SQL Server version information.

Return type:

MssqlVersionInfo | None

async get_feature_flag(driver, feature)

Check whether SQL Server supports a feature.

Return type:

bool

async get_optimal_type(driver, type_category)

Get optimal SQL Server type for a category.

Return type:

str

async get_tables(driver, schema=None)

Get tables sorted by dependency order with catalog fallback.

Return type:

list[TableMetadata]

async get_columns(driver, table=None, schema=None)

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

async get_indexes(driver, table=None, schema=None)

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

async get_foreign_keys(driver, table=None, schema=None)

Get foreign key metadata.

Return type:

list[ForeignKeyMetadata]

Migration Trackers

class sqlspec.adapters.mssql_python.MssqlPythonSyncMigrationTracker

Bases: MssqlPythonMigrationTrackerMixin, SyncMigrationTracker

T-SQL sync migration tracker.

ensure_tracking_table(driver)

Create the migration tracking table if it does not exist.

Return type:

None

record_migration(driver, version, description, execution_time_ms, checksum)

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

Return type:

None

class sqlspec.adapters.mssql_python.MssqlPythonAsyncMigrationTracker

Bases: MssqlPythonMigrationTrackerMixin, AsyncMigrationTracker

T-SQL async migration tracker.

async ensure_tracking_table(driver)

Create the migration tracking table if it does not exist.

Return type:

None

async record_migration(driver, version, description, execution_time_ms, checksum)

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

Return type:

None