MSSQL #

Microsoft SQL Server (MSSQL) is an enterprise relational database widely used in corporate environments, especially on Windows and .NET technology stacks. Python can interact with MSSQL through the pyodbc library, which uses the ODBC standard, making it flexible for connecting to various SQL Server versions — whether self-hosted, on a Windows server, or in Azure SQL Database. Understanding the syntax differences between MSSQL and other databases (such as the ? placeholder instead of %s) is key to avoiding hard-to-trace bugs when switching platforms.

Installation #

pip install pyodbc

Besides the Python library, you also need to install the ODBC Driver for SQL Server matching your operating system:

# macOS (using Homebrew)
brew install msodbcsql18

# Linux (Ubuntu/Debian)
curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
curl https://packages.microsoft.com/config/ubuntu/22.04/prod.list \
    | sudo tee /etc/apt/sources.list.d/mssql-release.list
sudo apt-get update
sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18

# Windows: download and install from
# https://learn.microsoft.com/sql/connect/odbc/download-odbc-driver-for-sql-server

The currently recommended ODBC Driver version is ODBC Driver 18 for SQL Server. Check the version installed on your system by running odbcinst -q -d on Linux/macOS, or look under “ODBC Data Sources” on Windows.

Creating a Connection #

MSSQL uses an ODBC-based connection string that’s slightly different from other database libraries. There are two common scenarios: SQL Server Authentication (username + password) and Windows Authentication (no password, using the current Windows account).

import pyodbc
import os

# ANTI-PATTERN: hardcoding credentials directly in code
conn = pyodbc.connect(
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=localhost;"
    "DATABASE=myapp;"
    "UID=sa;"
    "PWD=password123"  # ✗ -- don't do this
)

# CORRECT: read from environment variables
def build_connection_string() -> str:
    server   = os.getenv("MSSQL_HOST", "localhost")
    database = os.getenv("MSSQL_DB", "myapp")
    user     = os.getenv("MSSQL_USER", "sa")
    password = os.getenv("MSSQL_PASSWORD", "")
    driver   = os.getenv("MSSQL_DRIVER", "ODBC Driver 18 for SQL Server")
    
    return (
        f"DRIVER={{{driver}}};"
        f"SERVER={server};"
        f"DATABASE={database};"
        f"UID={user};"
        f"PWD={password};"
        "TrustServerCertificate=yes;"  # needed for local/dev connections
        "Encrypt=yes;"
    )

conn = pyodbc.connect(build_connection_string())
conn.autocommit = False  # manage transactions explicitly
print("Connection successful")
conn.close()

# Windows Authentication (no username/password)
conn_string_windows = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=localhost\\SQLEXPRESS;"
    "DATABASE=myapp;"
    "Trusted_Connection=yes;"
)
conn = pyodbc.connect(conn_string_windows)

Connection Context Manager #

import pyodbc
from contextlib import contextmanager
import os

@contextmanager
def db_connection():
    conn = pyodbc.connect(build_connection_string())
    conn.autocommit = False
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage
with db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT @@VERSION")
    version = cursor.fetchone()
    print("SQL Server Version:", version[0][:50])

Creating Tables #

MSSQL’s DDL syntax differs slightly from MySQL — use IDENTITY for auto-increment and NVARCHAR for Unicode strings.

def create_tables(conn):
    cursor = conn.cursor()
    
    # Check whether the table already exists before creating
    cursor.execute("""
        IF NOT EXISTS (
            SELECT * FROM sysobjects 
            WHERE name='pengguna' AND xtype='U'
        )
        CREATE TABLE pengguna (
            id          INT IDENTITY(1,1) PRIMARY KEY,
            nama        NVARCHAR(100) NOT NULL,
            email       NVARCHAR(150) NOT NULL,
            usia        INT,
            aktif       BIT DEFAULT 1,
            dibuat_pada DATETIME2 DEFAULT GETDATE(),
            CONSTRAINT UQ_pengguna_email UNIQUE (email)
        )
    """)
    
    cursor.execute("""
        IF NOT EXISTS (
            SELECT * FROM sysobjects 
            WHERE name='produk' AND xtype='U'
        )
        CREATE TABLE produk (
            id       INT IDENTITY(1,1) PRIMARY KEY,
            nama     NVARCHAR(200) NOT NULL,
            harga    DECIMAL(15, 2) NOT NULL,
            stok     INT DEFAULT 0,
            kategori NVARCHAR(100)
        )
    """)
    
    conn.commit()
    cursor.close()
    print("Tables created successfully.")

with db_connection() as conn:
    create_tables(conn)

Important differences from MySQL:
IDENTITY(1,1) replaces AUTO_INCREMENT
NVARCHAR for Unicode strings (supports non-Latin characters), VARCHAR is ASCII only
BIT replaces BOOLEAN
DATETIME2 replaces TIMESTAMP — higher precision
GETDATE() replaces CURRENT_TIMESTAMP

CRUD Operations #

The Placeholder Difference — Important! #

# ANTI-PATTERN: using %s like in MySQL/PostgreSQL
cursor.execute("INSERT INTO pengguna (nama) VALUES (%s)", ("Budi",))  # ✗ -- errors in pyodbc

# CORRECT: pyodbc uses question marks (?) as placeholders
cursor.execute("INSERT INTO pengguna (nama) VALUES (?)", ("Budi",))   # ✓

Inserting Data #

def tambah_pengguna(conn, nama: str, email: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # Use OUTPUT INSERTED.id to get the newly created ID
    cursor.execute(
        """
        INSERT INTO pengguna (nama, email, usia)
        OUTPUT INSERTED.id
        VALUES (?, ?, ?)
        """,
        (nama, email, usia)
    )
    
    baris = cursor.fetchone()
    id_baru = baris[0] if baris else None
    cursor.close()
    return id_baru

def tambah_banyak_pengguna(conn, daftar: list[tuple]) -> int:
    cursor = conn.cursor()
    cursor.executemany(
        "INSERT INTO pengguna (nama, email, usia) VALUES (?, ?, ?)",
        daftar
    )
    jumlah = cursor.rowcount
    cursor.close()
    return jumlah

with db_connection() as conn:
    id1 = tambah_pengguna(conn, "Budi Santoso", "[email protected]", 28)
    print(f"New user ID: {id1}")
    
    data_baru = [
        ("Sari Dewi", "[email protected]", 25),
        ("Andi Prasetyo", "[email protected]", 32),
    ]
    tambah_banyak_pengguna(conn, data_baru)

Reading Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    cursor = conn.cursor()
    cursor.execute("""
        SELECT id, nama, email, usia, aktif
        FROM pengguna
        WHERE aktif = 1
        ORDER BY dibuat_pada DESC
    """)
    
    kolom = [desc[0] for desc in cursor.description]
    hasil = [dict(zip(kolom, baris)) for baris in cursor.fetchall()]
    cursor.close()
    return hasil

def ambil_pengguna_by_id(conn, pengguna_id: int) -> dict | None:
    cursor = conn.cursor()
    cursor.execute(
        "SELECT id, nama, email, usia FROM pengguna WHERE id = ?",
        (pengguna_id,)
    )
    baris = cursor.fetchone()
    if not baris:
        return None
    
    kolom = [desc[0] for desc in cursor.description]
    cursor.close()
    return dict(zip(kolom, baris))

# Pagination with OFFSET-FETCH (SQL Server 2012+)
def ambil_pengguna_halaman(conn, halaman: int = 1, per_halaman: int = 10) -> list[dict]:
    cursor = conn.cursor()
    offset = (halaman - 1) * per_halaman
    
    cursor.execute("""
        SELECT id, nama, email, usia
        FROM pengguna
        ORDER BY id
        OFFSET ? ROWS
        FETCH NEXT ? ROWS ONLY
    """, (offset, per_halaman))
    
    kolom = [desc[0] for desc in cursor.description]
    hasil = [dict(zip(kolom, baris)) for baris in cursor.fetchall()]
    cursor.close()
    return hasil

with db_connection() as conn:
    semua = ambil_semua_pengguna(conn)
    for p in semua:
        print(f"[{p['id']}] {p['nama']} — {p['email']}")

Updating and Deleting #

def update_pengguna(conn, pengguna_id: int, nama: str, usia: int) -> bool:
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE pengguna SET nama = ?, usia = ? WHERE id = ?",
        (nama, usia, pengguna_id)
    )
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

def hapus_pengguna(conn, pengguna_id: int) -> bool:
    cursor = conn.cursor()
    cursor.execute("DELETE FROM pengguna WHERE id = ?", (pengguna_id,))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

Stored Procedures #

Calling stored procedures is a feature often used in SQL Server enterprise environments. pyodbc supports it through EXEC syntax or using callproc on some drivers.

-- Create the stored procedure in SQL Server first
CREATE PROCEDURE GetPenggunaByEmail
    @Email NVARCHAR(150)
AS
BEGIN
    SELECT id, nama, email, usia
    FROM pengguna
    WHERE email = @Email
END

def panggil_sp_get_pengguna(conn, email: str) -> dict | None:
    cursor = conn.cursor()
    
    # Way 1: using EXEC
    cursor.execute("EXEC GetPenggunaByEmail ?", (email,))
    baris = cursor.fetchone()
    
    if not baris:
        return None
    
    kolom = [desc[0] for desc in cursor.description]
    cursor.close()
    return dict(zip(kolom, baris))

with db_connection() as conn:
    pengguna = panggil_sp_get_pengguna(conn, "[email protected]")
    if pengguna:
        print(f"Found: {pengguna['nama']}")

Transactions #

def proses_order(conn, pengguna_id: int, produk_id: int, jumlah: int) -> int | None:
    cursor = conn.cursor()
    
    try:
        # Check stock with a row-level lock
        cursor.execute(
            "SELECT stok FROM produk WITH (UPDLOCK) WHERE id = ?",
            (produk_id,)
        )
        baris = cursor.fetchone()
        
        if not baris or baris[0] < jumlah:
            raise ValueError(f"Insufficient stock. Available: {baris[0] if baris else 0}")
        
        # Decrease stock
        cursor.execute(
            "UPDATE produk SET stok = stok - ? WHERE id = ?",
            (jumlah, produk_id)
        )
        
        # Create the order record, get the new ID via OUTPUT
        cursor.execute(
            """
            INSERT INTO orders (pengguna_id, produk_id, jumlah, total)
            OUTPUT INSERTED.id
            SELECT ?, ?, ?, (harga * ?) FROM produk WHERE id = ?
            """,
            (pengguna_id, produk_id, jumlah, jumlah, produk_id)
        )
        
        order_id = cursor.fetchone()[0]
        conn.commit()
        return order_id
        
    except Exception as e:
        conn.rollback()
        print(f"Transaction rolled back: {e}")
        return None
    finally:
        cursor.close()

Error Handling #

import pyodbc

def tambah_pengguna_aman(conn, nama: str, email: str) -> int | None:
    cursor = conn.cursor()
    try:
        cursor.execute(
            """
            INSERT INTO pengguna (nama, email)
            OUTPUT INSERTED.id
            VALUES (?, ?)
            """,
            (nama, email)
        )
        conn.commit()
        baris = cursor.fetchone()
        return baris[0] if baris else None
    
    except pyodbc.IntegrityError as e:
        conn.rollback()
        # Error 2627: Violation of UNIQUE constraint
        # Error 2601: Cannot insert duplicate key
        if "2627" in str(e) or "2601" in str(e):
            print(f"Email '{email}' is already registered.")
        else:
            print(f"Integrity error: {e}")
        return None
    
    except pyodbc.Error as e:
        conn.rollback()
        sqlstate = e.args[0]
        print(f"Database error [{sqlstate}]: {e}")
        return None
    
    finally:
        cursor.close()

Summary #

? placeholder, not %s — pyodbc uses the question mark ? as the parameter placeholder, unlike MySQL (%s) and PostgreSQL (%s or $1).
OUTPUT INSERTED.id — use this clause to get the ID of the newly inserted row, replacing lastrowid, which isn’t always reliable in pyodbc.
NVARCHAR not VARCHAR — always use NVARCHAR for string columns to support Unicode characters (including accented Indonesian letters).
IDENTITY(1,1) — the MSSQL equivalent of MySQL’s AUTO_INCREMENT for auto-increment columns.
OFFSET-FETCH — use it for pagination in SQL Server 2012+, replacing the LIMIT/OFFSET found in MySQL/PostgreSQL.
WITH (UPDLOCK) — use this hint when doing read-before-update inside a transaction to prevent race conditions.
TrustServerCertificate=yes — add it to the connection string for local/development connections; in production use a valid certificate.
Don’t hardcode credentials — always read host, user, password from environment variables.
autocommit = False — set it explicitly after creating the connection for full control over transactions.

← Previous: MySQL Next: Oracle →