FastMSSQL ⚡

FastMSSQL is an async Python library for Microsoft SQL Server (MSSQL), built in Rust. Unlike standard libaries, it uses a native SQL Server client—no ODBC required—simplifying installation on Windows, macOS, and Linux. Great for data ingestion, bulk inserts, and large-scale query workloads.

Features

• High performance: optimized for very high RPS and low overhead
• Rust core: memory‑safe and reliable, tuned Tokio runtime
• No ODBC: native SQL Server client, no external drivers needed
• Azure authentication: Service Principal, Managed Identity, and access token support (BETA)
• Connection pooling: bb8‑based, smart defaults (default max_size=20, min_idle=2)
• Async first: clean async/await API with async with context managers
• Strong typing: fast conversions for common SQL Server types
• Thread‑safe: safe to use in concurrent apps
• Cross‑platform: Windows, macOS, Linux
• Batch operations: high-performance bulk inserts and batch query execution
• Apache Arrow support

Installation

From PyPI (recommended)

pip install fastmssql

Optional dependencies

Apache Arrow support (for to_arrow() method):

pip install fastmssql[arrow]

Prerequisites

• Python 3.11 to 3.14
• Microsoft SQL Server (any recent version)

Quick start

Basic async usage

import asyncio
from fastmssql import Connection

async def main():
    conn_str = "Server=localhost;Database=master;User Id=myuser;Password=mypass"
    async with Connection(conn_str) as conn:
        # SELECT: use query() -> rows()
        result = await conn.query("SELECT @@VERSION as version")
        for row in result.rows():
            print(row['version'])

        # Pool statistics (tuple: connected, connections, idle, max_size, min_idle)
        connected, connections, idle, max_size, min_idle = await conn.pool_stats()
        print(f"Pool: connected={connected}, size={connections}/{max_size}, idle={idle}, min_idle={min_idle}")

asyncio.run(main())

Explicit Connection Management

When not utilizing Python's context manager (async with), FastMssql uses lazy connection initialization: if you call query() or execute() on a new Connection, the underlying pool is created if not already present.

For more control, you can explicitly connect and disconnect:

import asyncio
from fastmssql import Connection

async def main():
    conn_str = "Server=localhost;Database=master;User Id=myuser;Password=mypass"
    conn = Connection(conn_str)

    # Explicitly connect
    await conn.connect()
    assert await conn.is_connected()

    # Run queries
    result = await conn.query("SELECT 42 as answer")
    print(result.rows()[0]["answer"])  # -> 42

    # Explicitly disconnect
    await conn.disconnect()
    assert not await conn.is_connected()

asyncio.run(main())

Usage

Connection options

You can connect either with a connection string or individual parameters.

1. Connection string

import asyncio
from fastmssql import Connection

async def main():
    conn_str = "Server=localhost;Database=master;User Id=myuser;Password=mypass"
    async with Connection(connection_string=conn_str) as conn:
        rows = (await conn.query("SELECT DB_NAME() as db")).rows()
        print(rows[0]['db'])

asyncio.run(main())

1. Individual parameters

import asyncio
from fastmssql import Connection

async def main():
    async with Connection(
        server="localhost",
        database="master",
        username="myuser",
        password="mypassword"
    ) as conn:
        rows = (await conn.query("SELECT SUSER_SID() as sid")).rows()
        print(rows[0]['sid'])

asyncio.run(main())

Note: Windows authentication (Trusted Connection) is currently not supported. Use SQL authentication (username/password).

Azure Authentication (BETA)

🧪 This is a beta feature. Azure authentication functionality is experimental and may change in future versions.

FastMSSSQL supports Azure Active Directory (AAD) authentication for Azure SQL Database and Azure SQL Managed Instance. You can authenticate using Service Principals, Managed Identity, or access tokens.

Service Principal Authentication

import asyncio
from fastmssql import Connection, AzureCredential

async def main():
    # Create Azure credential using Service Principal
    azure_cred = AzureCredential.service_principal(
        client_id="your-client-id",
        client_secret="your-client-secret", 
        tenant_id="your-tenant-id"
    )
    
    async with Connection(
        server="yourserver.database.windows.net",
        database="yourdatabase",
        azure_credential=azure_cred
    ) as conn:
        result = await conn.query("SELECT GETDATE() as current_time")
        for row in result.rows():
            print(f"Connected! Current time: {row['current_time']}")

asyncio.run(main())

Managed Identity Authentication

For Azure resources (VMs, Function Apps, App Service, etc.):

import asyncio
from fastmssql import Connection, AzureCredential

async def main():
    # System-assigned managed identity
    azure_cred = AzureCredential.managed_identity()
    
    # Or user-assigned managed identity
    # azure_cred = AzureCredential.managed_identity(client_id="user-assigned-identity-client-id")
    
    async with Connection(
        server="yourserver.database.windows.net",
        database="yourdatabase",
        azure_credential=azure_cred
    ) as conn:
        result = await conn.query("SELECT USER_NAME() as user_name")
        for row in result.rows():
            print(f"Connected as: {row['user_name']}")

asyncio.run(main())

Access Token Authentication

If you already have an access token from another Azure service:

import asyncio
from fastmssql import Connection, AzureCredential

async def main():
    # Use a pre-obtained access token
    access_token = "your-access-token"
    azure_cred = AzureCredential.access_token(access_token)
    
    async with Connection(
        server="yourserver.database.windows.net",
        database="yourdatabase",
        azure_credential=azure_cred
    ) as conn:
        result = await conn.query("SELECT 1 as test")
        print("Connected with access token!")

asyncio.run(main())

Default Azure Credential

Uses the Azure credential chain (environment variables → managed identity → Azure CLI → Azure PowerShell):

import asyncio
from fastmssql import Connection, AzureCredential

async def main():
    # Use default Azure credential chain
    azure_cred = AzureCredential.default()
    
    async with Connection(
        server="yourserver.database.windows.net",
        database="yourdatabase",
        azure_credential=azure_cred
    ) as conn:
        result = await conn.query("SELECT 1 as test")
        print("Connected with default credentials!")

asyncio.run(main())

Prerequisites for Azure Authentication:

• Azure SQL Database or Azure SQL Managed Instance
• Service Principal with appropriate SQL Database permissions
• For Managed Identity: Azure resource with managed identity enabled
• For Default credential: Azure CLI installed and authenticated (az login)

See examples/azure_auth_example.py for comprehensive usage examples.

Working with data

import asyncio
from fastmssql import Connection

async def main():
    async with Connection("Server=.;Database=MyDB;User Id=sa;Password=StrongPwd;") as conn:
        # SELECT (returns rows)
        users = (await conn.query(
            "SELECT id, name, email FROM users WHERE active = 1"
        )).rows()
        for u in users:
            print(f"User {u['id']}: {u['name']} ({u['email']})")

        # INSERT / UPDATE / DELETE (returns affected row count)
        inserted = await conn.execute(
            "INSERT INTO users (name, email) VALUES (@P1, @P2)",
            ["Jane", "jane@example.com"],
        )
        print(f"Inserted {inserted} row(s)")

        updated = await conn.execute(
            "UPDATE users SET last_login = GETDATE() WHERE id = @P1",
            [123],
        )
        print(f"Updated {updated} row(s)")

asyncio.run(main())

Parameters use positional placeholders: @P1, @P2, ... Provide values as a list in the same order.

Batch operations

For high-throughput scenarios, use batch methods to reduce network round-trips:

import asyncio
from fastmssql import Connection

async def main_fetching():
    # Replace with your actual connection string
    async with Connection("Server=.;Database=MyDB;User Id=sa;Password=StrongPwd;") as conn:

        # --- 1. Prepare Data for Demonstration ---
        columns = ["name", "email", "age"]
        data_rows = [
            ["Alice Johnson", "alice@example.com", 28],
            ["Bob Smith", "bob@example.com", 32],
            ["Carol Davis", "carol@example.com", 25],
            ["David Lee", "david@example.com", 35],
            ["Eva Green", "eva@example.com", 29]
        ]
        await conn.bulk_insert("users", columns, data_rows)

        # --- 2. Execute Query and Retrieve the Result Object ---
        print("\n--- Result Object Fetching (fetchone, fetchmany, fetc