Python SQL Server CRUD Tutorial with pyodbc: 10 Examples

• Run files 01–09 individually for focused learning; file 10 repeats the complete workflow.
• SQL values are parameterized, while the trusted database identifier is interpolated only for CREATE DATABASE.
• The current full demo commits each helper independently, so it is a sequence of transactions—not one all-or-nothing transaction.

Python SQL Server CRUD Learning Path

2. Shared Driver and Connection Pattern

SERVER = r"localhost\WINCC"
DB_NAME = "SQF_DB"
DRIVER_NAME = "ODBC Driver 17 for SQL Server"

if DRIVER_NAME not in pyodbc.drivers():
    raise RuntimeError(
        f"{DRIVER_NAME} is not installed. Installed drivers: {pyodbc.drivers()}"
    )

Each file verifies the driver before connecting. The master connection creates the database; the database connection handles table/CRUD work. Windows authentication uses the identity running Python.

MASTER_CONN = (
    f"DRIVER={DRIVER};SERVER={SERVER};DATABASE=master;"
    "Trusted_Connection=yes;TrustServerCertificate=yes;Connection Timeout=5;"
)
DB_CONN = (
    f"DRIVER={DRIVER};SERVER={SERVER};DATABASE={DB_NAME};"
    "Trusted_Connection=yes;TrustServerCertificate=yes;Connection Timeout=5;"
)

TrustServerCertificate=yes is useful in a controlled lab with an untrusted/self-signed certificate. Production should deploy a trusted SQL Server certificate. The code is also repeated in every file; a shared db_config.py would prevent configuration drift.

3. File 01—Create the Database

with pyodbc.connect(MASTER_CONN, autocommit=True) as conn:
    exists = conn.cursor().execute("SELECT DB_ID(?)", DB_NAME).fetchone()[0]
    if exists is None:
        conn.cursor().execute(f"CREATE DATABASE [{DB_NAME}]")

Connecting to master is necessary because SQF_DB might not exist. CREATE DATABASE requires execution outside a user transaction, so the master connection uses autocommit. DB_ID(?) safely binds the database name as a value for the check.

The DDL identifier cannot be supplied through a normal ? value parameter, so it is interpolated. This is acceptable only because DB_NAME is a trusted constant. If configurable, validate it against a strict identifier allowlist and quote it correctly.

4. File 02—Create dbo.SCRIPT

CREATE TABLE dbo.SCRIPT
(
    ScriptID INT IDENTITY(1,1) PRIMARY KEY,
    ScriptName NVARCHAR(100) NOT NULL,
    ScriptText NVARCHAR(MAX) NOT NULL,
    IsActive BIT NOT NULL DEFAULT(1),
    CreatedAt DATETIME2(0) NOT NULL DEFAULT SYSDATETIME()
);

The schema demonstrates identity keys, Unicode text, Boolean status and a server-generated timestamp. OBJECT_ID(..., N'U') prevents duplicate creation. It does not validate or upgrade an existing table with incorrect columns/types; production schema changes require migrations.

Insert Data into SQL Server with Python pyodbc

Single insert with generated identity

INSERT INTO dbo.SCRIPT (ScriptName, ScriptText, IsActive)
OUTPUT INSERTED.ScriptID
VALUES (?, ?, ?)

OUTPUT INSERTED.ScriptID reliably returns the identity created by the statement. Values are bound separately, so apostrophes inside ScriptText do not break SQL and untrusted input is not interpreted as SQL syntax.

Batch insert

rows = [
    ("Temperature Check", "SELECT 25.4 AS TemperatureC", True),
    ("Alarm Reset", "PRINT 'Alarm reset'", False),
    ("Production Check", "SELECT GETDATE() AS CheckedAt", True),
]
cur.fast_executemany = True
cur.executemany(sql, rows)
conn.commit()

executemany() applies the same parameterized statement to each tuple. fast_executemany can reduce round trips with the Microsoft driver. The returned count is len(rows), representing rows submitted after success—not a server-returned affected-row count.

Both demonstrations add duplicate logical examples when rerun. Add a unique business key, MERGE/upsert policy, test-batch cleanup or explicit idempotency rule when repetition must not duplicate data.

6. File 05—Read and Display Records

SELECT ScriptID, ScriptName, ScriptText, IsActive, CreatedAt
FROM dbo.SCRIPT
ORDER BY ScriptID

The SELECT returns all rows in identity order, prints each pyodbc Row object and handles an empty result. For real applications, select only required columns, add filters/paging, map rows into dictionaries or domain objects, and avoid printing sensitive script text to uncontrolled logs.

7. File 06—Update by Primary Key

UPDATE dbo.SCRIPT
SET ScriptText = ?, IsActive = ?
WHERE ScriptID = ?

get_or_create_sample() ensures a row exists, then the update targets one primary-key value and returns cursor.rowcount. Confirm the affected count before reporting success. In a multi-user system, consider optimistic concurrency—such as comparing a rowversion—so one user does not silently overwrite another user’s change.

The selected sample is simply the lowest ScriptID; this is deterministic but may modify a row unrelated to the update lesson if the table contains real data. A training-specific row name or batch ID is safer.

8. File 07—Delete a Disposable Row

new_id = cursor.execute(
    "INSERT ... OUTPUT INSERTED.ScriptID VALUES (?, ?, ?)", ...
).fetchone()[0]

cursor.execute("DELETE FROM dbo.SCRIPT WHERE ScriptID = ?", new_id)

The file creates a row specifically for deletion, captures its identity, then deletes exactly that row. This is safer than choosing an arbitrary existing row. The insert and delete are committed separately; if deletion fails, the disposable row remains. For zero net change, perform both operations in one transaction and roll back the demonstration intentionally.

9. File 08—Guarded TRUNCATE TABLE

RUN_TRUNCATE = False

def truncate_scripts(conn, confirm=False):
    if not confirm:
        raise ValueError("TRUNCATE cancelled because confirm=False.")
    conn.cursor().execute("TRUNCATE TABLE dbo.SCRIPT")
    conn.commit()

Two gates protect the operation: the global flag defaults to false and the function requires confirm=True. TRUNCATE removes all rows, normally resets the identity seed, uses schema-level locking and requires stronger permission than ordinary DELETE. It can be blocked by certain foreign-key relationships and other dependencies.

10. File 09—Test the Connection

SELECT
    @@SERVERNAME AS ServerName,
    DB_NAME() AS DatabaseName,
    SYSTEM_USER AS LoginName,
    @@VERSION AS VersionInfo

This is an excellent diagnostic query: it proves the actual server, database, login and SQL Server version. However, the script calls create_database() before testing, so it is not strictly read-only. A pure connection-test utility should connect to an expected existing database and fail without creating anything.

@@SERVERNAME may not match the connection string exactly after server renaming or alias use; treat the output as diagnostic evidence rather than string-equality validation alone.

11. File 10—Complete CRUD Demonstration

The final file combines setup, single insert, batch insert, SELECT, update, delete, optional truncate and final SELECT.

welcome_id = insert_script(conn, "Full CRUD Welcome", ...)
insert_many(conn, [
    ("Full CRUD Temperature", ...),
    ("Full CRUD Alarm", ...),
])
rows = select_scripts(conn)
alarm_id = next(
    row.ScriptID for row in reversed(rows)
    if row.ScriptName == "Full CRUD Alarm"
)
update_script(conn, welcome_id, "PRINT 'System ready'", True)
delete_script(conn, alarm_id)

Finding the newest matching alarm by reversing the ScriptID-ordered rows works for this single-process demo. In concurrent use, another identical row could interfere. Capture inserted identities directly—such as using OUTPUT INSERTED.ScriptID for each required row—or add an unambiguous batch ID.

The next(...) expression raises StopIteration if no matching row exists. Although the script inserts that row first, explicit handling would produce a clearer diagnostic.

pyodbc Transactions, Commit and Rollback Explained

The full CRUD demo wraps operations in try/except pyodbc.Error and calls conn.rollback() on failure. However, every helper function calls conn.commit() internally. Therefore:

• The database/table setup is committed.
• The single insert is committed.
• The batch insert is committed.
• The update and delete are committed separately.
• A failure later can roll back only the current uncommitted work—not earlier commits.

with pyodbc.connect(DB_CONN) as conn:
    try:
        create_table_without_commit(conn)
        new_id = insert_without_commit(conn, ...)
        update_without_commit(conn, new_id, ...)
        conn.commit()             # one transaction boundary
    except Exception:
        conn.rollback()
        raise

Database creation remains separate because it uses the autocommit master connection.

13. Refactoring and Production Design

All ten files repeat driver checks, connection strings, database creation and often table creation. Repetition is helpful when each lesson must run alone, but maintainable applications should separate responsibilities:

sql_server_demo/
├── config.py          # validated settings
├── connection.py      # connection factories
├── schema.py          # database/table migrations
├── repository.py      # insert/select/update/delete
├── diagnostics.py     # read-only connection checks
├── cli.py             # commands and destructive confirmations
└── tests/             # isolated integration tests

Recommended changes

• Load environment-specific configuration from a controlled source.
• Use trusted certificates instead of globally trusting the server certificate.
• Use separate deployment and runtime accounts with least privilege.
• Let the top-level workflow define transaction boundaries.
• Add unique keys or batch IDs to make retries idempotent.
• Use logging with operation name, row count, duration and exception details.
• Catch expected pyodbc.Error cases and re-raise after rollback.
• Add integration tests against a disposable database.
• Remove unused MASTER_CONN/RUN_TRUNCATE from files that do not need them—or explain that uniform configuration is intentional.
• Use migration scripts to compare/upgrade existing schemas.

14. Troubleshooting Checklist