Microsoft SQL Server 🔗

Supported drivers:

dbapi	default	driver	connection class
pymssql		`mssql+pymssql`	`pymssql._pymssql.Connection`

pymssql🔗

pymssql is the default dbapi driver for Microsoft SQL Server in pydapper.

Installation🔗

pippoetry

pip install pydapper[pymssql]

poetry add pydapper -E pymssql

DSN format🔗

TemplateExampleExample (Default Driver)

dsn = f"mssql+pymssql://{user}:{password}@{host}:{port}/{dbname}"

dsn = "mssql+pymssql://myuser:mypassword@localhost:1433/mydb"

dsn = "mssql://myuser:mypassword@localhost:1433/mydb"

Example - `connect`🔗

Warning

Exiting the with block closes the connection and rolls back any uncommitted work — commit explicitly or use a transaction() block. See Transactions below.

import pydapper

# exiting the block closes the connection, and pymssql's close() rolls back uncommitted
# work — commit explicitly or use commands.transaction(); see the Transactions section
with pydapper.connect("mssql://sa:pydapper!PYDAPPER@localhost:1434/pydapper") as commands:
    print(type(commands))
    # <class 'pydapper.mssql.pymssql.PymssqlCommands'>

    print(type(commands.connection))
    # <class 'pymssql._pymssql.Connection'>

    with commands.cursor() as raw_cursor:
        print(type(raw_cursor))
        # <class 'pymssql._pymssql.Cursor'>

Example - `using`🔗

Use pydapper with a custom connection pool.

from collections import deque

import pymssql

import pydapper


class SimplePool:
    """pymssql does not provide a pool interface, this is a simple example that should never be used in production"""

    def __init__(self, **connect_kwargs):
        self._connect_kwargs = connect_kwargs
        self._connections = deque()

    def getconn(self):
        if len(self._connections) == 0:
            return pymssql.connect(**self._connect_kwargs)
        return self._connections.pop()

    def putconn(self, conn):
        self._connections.append(conn)

    def __del__(self):
        for conn in self._connections:
            conn.close()


my_pool = SimplePool(server="localhost", port=1434, user="sa", password="pydapper!PYDAPPER", database="pydapper")

commands = pydapper.using(my_pool.getconn())
print(type(commands))
# <class 'pydapper.mssql.pymssql.PymssqlCommands'>

print(type(commands.connection))
# <class 'pymssql._pymssql.Connection'>

my_pool.putconn(commands.connection)

Transactions🔗

A non-autocommit pymssql connection (the default) holds an always-open BEGIN TRAN: the driver issues one when the connection opens and re-issues one after every commit() and rollback(), so the connection is never outside a transaction. Exiting with pydapper.connect(...) delegates to pymssql's context manager, which only closes the connection — and close() implicitly rolls back all uncommitted work. This is the classic silent-data-loss trap:

with pydapper.connect("mssql://user:password@localhost:1433/mydb") as commands:
    rows = commands.execute(insert_sql, params=task)
    assert rows == 1  # True — but nothing was committed
# exit closed the connection, which rolled the insert back

Fix it either way — commit explicitly before the block ends:

with pydapper.connect("mssql://user:password@localhost:1433/mydb") as commands:
    commands.execute(insert_sql, params=task)
    commands.commit()  # durable

or scope the work in a transaction() block, which commits on clean exit:

with pydapper.connect("mssql://user:password@localhost:1433/mydb") as commands:
    with commands.transaction():
        commands.execute(insert_sql, params=task)
    # committed

Two more pymssql quirks worth knowing:

autocommit is a method, not an attribute: commands.connection.autocommit(True). Turning autocommit on discards the currently open transaction via a ROLLBACK, so switch modes before doing work, not after.
You can also pass autocommit=True to connect()'s kwargs to make every statement durable immediately.

See Transactions and Context manager semantics for the cross-driver picture.

Microsoft SQL Server🔗