Insert SQL Server Data with Python & pyodbc

Quick answer

Use pyodbc.connect() to open SQL Server, verify the target database and schema, prepare an INSERT ... VALUES (?, ...) statement, pass row tuples through cursor.executemany(), and call connection.commit() only after the complete batch succeeds. Parameter markers keep data separate from SQL syntax and let the driver convert compatible Python values.

Validate the database, table and all required columns before inserting.
Use parameterized statements—never concatenate untrusted values into SQL.
Treat the 100 generated rows as test data and protect production databases from accidental reruns.

Python pyodbc INSERT: Single Rows, Batches and Transactions

This guide targets the practical query “insert data into SQL Server using Python.” It verifies the destination schema, generates typed records, binds values with question-mark parameters, uses executemany and fast_executemany for batches, and explains commit and rollback behavior.

1. Insert Data into SQL Server: Python Workflow

This practical creates 100 sample records representing six stages of an SQF furnace cycle: charging, heating, soaking, oil quenching, cooling and charge discharge. Before writing anything, the program confirms that the database, target table and expected columns exist.

SQL INSERT • Parameterized Batch Architecture

Python Events → pyodbc → SQL Server INSERT

Generate furnace-event values in Python, validate the SQL target, bind values with parameters, insert as a transaction and verify the committed row count.

Connect

Open the intended SQL Server/database with the configured driver.

pyodbc.connect(...)

Verify Target

Confirm database, table and required columns before writing.

DB_ID / OBJECT_ID column check

Generate Events

Build realistic Python records for the SQF process stages.

records = [...] len(records) = 100

Parameterized INSERT

Keep SQL syntax separate from Python data values.

VALUES (?, ?, ?, ...)

Batch + Transaction

Insert multiple rows efficiently and commit only after success.

cursor.executemany(...) connection.commit()

Confirm

Count/read rows after commit to verify the write operation.

SELECT COUNT(*) Inserted OK

Python valuesdatetime / int / float / str

? parametersSafe value binding

executemanyBatch insert

commitPersist transaction

Architecture:Connect→Verify→Generate→Bind Parameters→Insert / Commit→Confirm

Code-rendered diagram: the write path makes the transaction boundary and post-insert verification visible to learners.

An SQL INSERT changes persistent data. Unlike the previous read-only example, this workflow requires explicit authorization, careful target verification and a clear recovery plan.

Lab Manual Index: 8 Guided Exercises

Every section of this tutorial carries its own lab manual — objective, procedure, a line-by-line explanation, copy-paste practice code, the exact console output to expect, a checkpoint and a troubleshooting table. Work through them in order; each lab builds on the checkpoint before it. Total bench time is about 65 minutes.

2. Requirements and Safety Conditions

Python 3.10 or later and pyodbc installed with python -m pip install pyodbc.
Microsoft ODBC Driver 17 for SQL Server.
Windows access to SOFTWELL\WINCC.
Database SQF_DB and table dbo.tblEvent.
Permissions to read metadata, insert rows and read the verification count.
Compatible SQL column types for the Python datetime, time strings, integers, decimals and status text.

Important: run dummy-data code only against an approved training or test database. Repeating the script inserts another 100 rows because the supplied table design does not include an idempotency key or duplicate check. Back up important data and confirm the server/database names before execution.

3. Configure the SQL Server Connection

01_connection.py

SERVER = r"SOFTWELL\WINCC"
DATABASE = "SQF_DB"
TABLE = "dbo.tblEvent"
DRIVER = "ODBC Driver 17 for SQL Server"


def connect(database_name):
    return pyodbc.connect(
        f"DRIVER={{{DRIVER}}};"
        f"SERVER={SERVER};"
        f"DATABASE={database_name};"
        "Trusted_Connection=yes;"
        "TrustServerCertificate=yes;",
        timeout=15,
    )

The raw string preserves the named-instance backslash. Triple braces in the f-string produce the braces ODBC expects around the driver name. Trusted_Connection=yes uses the current Windows identity.

TrustServerCertificate=yes skips certificate-chain validation while retaining encryption when the driver negotiates it. It is convenient in some controlled internal labs, but production systems should deploy a trusted SQL Server certificate and use normal validation.

Lab Manual 3 — Configure and Prove the SQL Server Connection

8 min

Objective

Prove that Python can reach the exact SQL Server instance and database you intend to write to. Nothing else in this practical is safe until this step prints the server name you expect.

Procedure

Open ODBC Data Sources (64-bit) on the lab PC, go to the Drivers tab and copy the driver name exactly as Windows shows it.
Replace SERVER with your own instance in PCNAME\INSTANCE form and DATABASE with your database name.
Save the snippet below as 01_connection.py in your project folder.
Run python 01_connection.py and read the printed server name before continuing.

What each line does

Code	Meaning in this lab
r"SOFTWELL\WINCC"	Raw string. Without the leading `r`, Python reads `\W` as an escape sequence and the named instance breaks.
DRIVER	Must match the installed ODBC driver character for character. On Driver 18 machines this becomes `ODBC Driver 18 for SQL Server`.
f"DRIVER={{{DRIVER}}};"	Three braces produce one literal `{`, the variable value, and one literal `}` — ODBC requires the driver name inside braces.
Trusted_Connection=yes	Uses the Windows account that runs the script, so no password is stored in source code.
TrustServerCertificate=yes	Skips certificate-chain validation. Acceptable inside a closed training lab, never on a plant or production server.
timeout=15	Login timeout in seconds. The script fails fast instead of freezing a report PC or HMI station.

Copy-paste practice code

01_connection_test.py

import pyodbc

print("Installed ODBC drivers:")
for driver in pyodbc.drivers():
    print("  -", driver)

connection = connect("master")
cursor = connection.cursor()
cursor.execute("SELECT @@SERVERNAME, DB_NAME(), SYSTEM_USER")
server_name, database_name, login_name = cursor.fetchone()

print("Server  :", server_name)
print("Database:", database_name)
print("Login   :", login_name)
connection.close()

Paste this below the connect() function from Section 3 and run the file as it is.

Expected result

console output

Installed ODBC drivers:
  - SQL Server
  - ODBC Driver 17 for SQL Server
Server  : SOFTWELL\WINCC
Database: master
Login   : SOFTWELL\Trainer

Checkpoint

Pass when: The server name printed by SQL Server matches your SERVER constant, and the login shown is the Windows account you expect to own the inserted rows.

If it fails

Symptom	What to check
IM002 — data source name not found	The driver string is misspelled, or you installed the 32-bit driver and are running 64-bit Python. Compare with the list the script just printed.
08001 — SQL Server does not exist	Wrong instance name, SQL Server Browser service stopped, or TCP/IP disabled in SQL Server Configuration Manager.
28000 — login failed for user	The Windows account has no SQL login. Ask the DBA to create the login and grant it rights on the training database only.

4. Verify the Database and Table

02_verify_database_table.py

def database_exists():
    connection = connect("master")
    cursor = connection.cursor()
    cursor.execute(
        "SELECT COUNT(*) FROM sys.databases WHERE name = ?",
        DATABASE,
    )
    result = cursor.fetchone()[0] > 0
    connection.close()
    return result


def table_exists(connection):
    cursor = connection.cursor()
    cursor.execute("""
        SELECT COUNT(*)
        FROM INFORMATION_SCHEMA.TABLES
        WHERE TABLE_SCHEMA = 'dbo'
          AND TABLE_NAME = 'tblEvent'
    """)
    return cursor.fetchone()[0] > 0

The database check connects to master because the target database might not exist. The table check runs only after connecting to SQF_DB. The database name is a bound value, while the schema and table are fixed identifiers in the metadata query.

Lab Manual 4 — Verify the Database and Table Exist

6 min

Objective

Confirm the destination exists before an INSERT is prepared, so a wrong instance produces a clear message instead of a driver exception halfway through a batch.

Procedure

Run the Section 4 function against master to test for SQF_DB.
Connect to the target database and test the table with OBJECT_ID.
Write down the row count printed before the insert — Lab 8 and Lab 9 compare against it.

What each line does

Code	Meaning in this lab
connect("master")	`master` always exists, so it is the safe place to ask whether another database exists.
WHERE name = ?	A parameter marker is used even for a metadata lookup, so the habit carries into the real INSERT.
cursor.fetchone()[0]	`fetchone()` returns one row; index `[0]` takes the first column, the count.
connection.close()	The `master` connection is released before the working connection is opened.
OBJECT_ID(?, 'U')	Returns the object id of a user table, or `None` when the table is missing or not visible to your login.

Copy-paste practice code

02_verify_target.py

connection = connect(DATABASE)
cursor = connection.cursor()

cursor.execute("SELECT OBJECT_ID(?, 'U')", TABLE)
object_id = cursor.fetchone()[0]
print("Table found" if object_id else "TABLE MISSING", "->", TABLE)

if object_id:
    cursor.execute(f"SELECT COUNT(*) FROM {TABLE}")
    print("Rows before insert:", cursor.fetchone()[0])

connection.close()

TABLE is a fixed constant in your own script, never user input, so it is safe to place in the f-string. Values always stay in parameters.

Expected result

console output

Table found -> dbo.tblEvent
Rows before insert: 0

Checkpoint

Pass when: The table is reported as found and the starting row count is recorded in your lab notebook.

If it fails

Symptom	What to check
Invalid object name 'dbo.tblEvent'	You are connected to the wrong database, or the table sits in a different schema. Print `DB_NAME()` to confirm.
Table reports missing but exists in SSMS	Your login can connect but has no rights on the object, so `OBJECT_ID` returns NULL. Ask for SELECT and INSERT on the table.
Count is far higher than expected	The script was run before. Identify and remove only your own test batch before repeating.

5. Validate All Required Columns

03_validate_columns.py

REQUIRED_COLUMNS = [
    "DT",
    "TM",
    "SQF_No",
    "ChargeNo",
    "Event_From",
    "Event_To",
    "Temp_Set",
    "Temp_Act",
    "Cp_Set",
    "Cp_Act",
    "Oil_Set",
    "Oil_Act",
    "Jacket_Set",
    "Jacket_Act",
    "Fan_Status",
]


def get_missing_columns(connection):
    cursor = connection.cursor()
    cursor.execute("""
        SELECT COLUMN_NAME
        FROM INFORMATION_SCHEMA.COLUMNS
        WHERE TABLE_SCHEMA = 'dbo'
          AND TABLE_NAME = 'tblEvent'
    """)
    available_columns = {
        row.COLUMN_NAME for row in cursor.fetchall()
    }
    return [
        name
        for name in REQUIRED_COLUMNS
        if name not in available_columns
    ]

Checking names prevents an insert from starting against an incomplete schema. A stronger production validator should also compare data types, lengths, precision, scale, nullability, identity/default behavior and insert permissions—not only column names.

Lab Manual 5 — Validate All 15 Required Columns

7 min

Objective

Catch a schema mismatch as a readable message — a named missing column — instead of a truncated driver error thrown in the middle of a 100-row batch.

Procedure

Run the Section 5 function and confirm an empty missing list.
Print the full column definition of the table and compare data types with the values Python will send.
Note any NOT NULL column that is absent from REQUIRED_COLUMNS — it must have a default or the insert fails.

What each line does

Code	Meaning in this lab
REQUIRED_COLUMNS	Order matters. This list defines the column order in the INSERT statement and therefore the value order in every tuple.
INFORMATION_SCHEMA.COLUMNS	ANSI-standard catalog view listing every column of every table the login can see.
{row.COLUMN_NAME for row in ...}	A set comprehension, so each name lookup is fast and duplicates cannot occur.
if name not in available_columns	Returns only the names the table does not have, which becomes the error message.

Copy-paste practice code

03_validate_columns.py

connection = connect(DATABASE)
missing = get_missing_columns(connection)

if missing:
    print("MISSING COLUMNS:", ", ".join(missing))
else:
    print("All", len(REQUIRED_COLUMNS), "required columns are present.")

cursor = connection.cursor()
cursor.execute("""
    SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, IS_NULLABLE
    FROM INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_SCHEMA = 'dbo' AND TABLE_NAME = 'tblEvent'
    ORDER BY ORDINAL_POSITION
""")

print(f"{"COLUMN":<12}{"TYPE":<12}{"LEN":<7}NULL?")
for name, data_type, length, nullable in cursor.fetchall():
    print(f"{name:<12}{data_type:<12}{str(length):<7}{nullable}")

connection.close()

On Python 3.10 and 3.11, replace the inner double quotes in that header f-string with single quotes; nested same-type quotes are only allowed from Python 3.12.

Expected result

console output

All 15 required columns are present.
COLUMN      TYPE        LEN    NULL?
DT          date        None   YES
TM          time        None   YES
SQF_No      int         None   YES
ChargeNo    varchar     20     YES
Event_From  varchar     50     YES
Event_To    varchar     50     YES
Temp_Set    float       None   YES
Temp_Act    float       None   YES

Checkpoint

Pass when: Every required column is present, the text column lengths are longer than the strings you will generate, and no unexpected NOT NULL column exists outside the list.

If it fails

Symptom	What to check
A name is listed as missing but visible in SSMS	Check spelling and case, and confirm the schema is `dbo` and not a personal schema.
String data, right truncation later	A `varchar` column is shorter than the generated text. Compare lengths here, before inserting.
Arithmetic overflow later	A `decimal(p,s)` column cannot hold the generated precision. Round the Python values or widen the column.

6. Generate Realistic Furnace Records

generate_record() converts a record number and timestamp into one event tuple. Modular arithmetic drives a repeating 20-record furnace cycle, rotates furnace numbers 1–3 and assigns one charge number per 10 records.

Cycle index	Process stage	Temperature pattern	Fan
0–3	Charging	Rises from approximately 100 °C	OFF
4–8	Heating	Ramps toward 850 °C	ON
9–12	Soaking	Held near 850 °C	ON
13–15	Oil quenching	Falls from high temperature	ON
16–18	Cooling	Falls toward 100 °C	ON
19	Charge discharged	Approximately 92–99 °C	OFF

04_generate_records.py

start_time = (
    datetime.now().replace(microsecond=0)
    - timedelta(minutes=NUMBER_OF_RECORDS - 1)
)

for number in range(1, NUMBER_OF_RECORDS + 1):
    record_time = start_time + timedelta(minutes=number - 1)
    records.append(generate_record(number, record_time))

The list spans 100 one-minute timestamps ending near the current time. random.uniform() adds realistic variation and round(..., 2) produces two-decimal values. For reproducible test runs, call random.seed(known_value) before generation.

Lab Manual 6 — Generate and Inspect the Furnace Records

8 min

Objective

Produce 100 tuples of 15 values representing a repeating 20-step furnace cycle, and inspect a small sample so that any type or ordering mistake is caught in Python rather than by the database.

Procedure

Set a fixed random seed so the lab is repeatable and two trainees can compare results.
Generate five records instead of a hundred and print each tuple with its length.
Confirm the value order matches REQUIRED_COLUMNS exactly, left to right.

What each line does

Code	Meaning in this lab
datetime.now().replace(microsecond=0)	Drops microseconds so `time` and `datetime` columns accept the value without rounding surprises.
- timedelta(minutes=NUMBER_OF_RECORDS - 1)	Back-dates the batch so the first row is 99 minutes old and the last row lands near the current time.
range(1, NUMBER_OF_RECORDS + 1)	Produces 1-based record numbers, which drive the cycle stage through modular arithmetic.
generate_record(number, record_time)	Returns one tuple of 15 values in the same order as `REQUIRED_COLUMNS`.
random.uniform(...) / round(..., 2)	Adds realistic process noise and trims it to two decimals so float columns store clean values.
random.seed(42)	Optional but recommended in a lab: makes the generated batch identical on every run.

Copy-paste practice code

04_generate_preview.py

import random
from datetime import datetime, timedelta

random.seed(42)

preview = []
start_time = datetime.now().replace(microsecond=0) - timedelta(minutes=4)

for number in range(1, 6):
    record_time = start_time + timedelta(minutes=number - 1)
    preview.append(generate_record(number, record_time))

for index, row in enumerate(preview, start=1):
    print(f"Row {index}: {len(row)} values")
    for column_name, value in zip(REQUIRED_COLUMNS, row):
        print(f"   {column_name:<12} {type(value).__name__:<10} {value}")
    print("-" * 46)

Nothing is written to SQL Server in this lab. It is a dry run that proves the data shape before any INSERT is prepared.

Expected result

console output

Row 1: 15 values
   DT           date       2026-09-19
   TM           str        11:04:00
   SQF_No       int        1
   ChargeNo     str        CH-001
   Event_From   str        Charging
   Event_To     str        Heating
   Temp_Set     float      850.0
   Temp_Act     float      104.36
   Fan_Status   str        OFF
----------------------------------------------

Checkpoint

Pass when: Every row reports exactly 15 values, each Python type is compatible with the SQL type printed in Lab 5, and the stage names follow the documented cycle order.

If it fails

Symptom	What to check
A row reports 14 or 16 values	`generate_record()` was edited without updating `REQUIRED_COLUMNS` and the VALUES markers. All three must stay in step.
Timestamps are identical	The loop is not advancing `record_time`. Confirm `timedelta(minutes=number - 1)` is inside the loop.
Values change on every run	The seed line is missing or placed after generation. Call `random.seed()` first.

7. Create a Parameterized pyodbc INSERT Query

05_insert_query.sql

INSERT INTO dbo.tblEvent
(
    DT,
    TM,
    SQF_No,
    ChargeNo,
    Event_From,
    Event_To,
    Temp_Set,
    Temp_Act,
    Cp_Set,
    Cp_Act,
    Oil_Set,
    Oil_Act,
    Jacket_Set,
    Jacket_Act,
    Fan_Status
)
VALUES
(
    ?, ?, ?, ?, ?,
    ?, ?, ?, ?, ?,
    ?, ?, ?, ?, ?
)

The fifteen question marks correspond positionally to the fifteen tuple values returned by generate_record(). Parameterization improves safety and type handling. Table and column identifiers cannot use ordinary value parameters, so keep them fixed or validate them against a strict allowlist.

Lab Manual 7 — Build the Parameterized INSERT Safely

6 min

Objective

Write an INSERT statement in which SQL syntax and data values never mix, and build it from the column list so the marker count can never drift out of step with the tuples.

Procedure

Read the hand-written statement in Section 7 and count the markers against the column list.
Replace it with the generated version below so one list controls columns, markers and tuple order.
Print the finished statement once and keep it in your lab notes.

What each line does

Code	Meaning in this lab
INSERT INTO dbo.tblEvent (...)	Naming the columns explicitly means the statement keeps working if someone adds a column to the table later.
VALUES (?, ?, ...)	pyodbc uses qmark paramstyle. Markers are positional — the first `?` takes the first tuple value.
? never quotes	You never wrap a marker in quotes. The driver sends the value with its type, so quoting would make it a literal question mark.
Identifiers are not parameters	Table and column names cannot be bound as values. Keep them as fixed constants or check them against a strict allowlist.

Copy-paste practice code

05_build_query.py

column_list = ", ".join(REQUIRED_COLUMNS)
marker_list = ", ".join(["?"] * len(REQUIRED_COLUMNS))

insert_query = (
    f"INSERT INTO {TABLE} ({column_list}) "
    f"VALUES ({marker_list})"
)

print(insert_query)
print("Columns:", len(REQUIRED_COLUMNS), "| Markers:", marker_list.count("?"))

Generating the statement from REQUIRED_COLUMNS removes the most common cause of failed batch inserts — a column added in one place and forgotten in the other.

Expected result

console output

INSERT INTO dbo.tblEvent (DT, TM, SQF_No, ChargeNo, Event_From, Event_To, Temp_Set, Temp_Act, Cp_Set, Cp_Act, Oil_Set, Oil_Act, Jacket_Set, Jacket_Act, Fan_Status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
Columns: 15 | Markers: 15

Checkpoint

Pass when: Column count, marker count and tuple length are all 15, and the printed statement contains no Python values.

If it fails

Symptom	What to check
COUNT field incorrect or syntax error	The number of markers does not match the number of values in the tuple. Compare both counts as the snippet prints them.
Values land in the wrong columns	The tuple order differs from the column order. Both come from `REQUIRED_COLUMNS` once you use the generated statement.
An f-string was used for values	Never format values into the SQL text. Values belong in the parameter list passed to `execute()` or `executemany()`.

8. Insert Multiple Rows with executemany and Transactions

06_batch_insert.py

cursor = connection.cursor()
cursor.fast_executemany = True
cursor.executemany(insert_query, records)
connection.commit()

executemany() applies the same prepared statement to every tuple. With Microsoft’s ODBC driver, fast_executemany can greatly reduce round trips. commit() makes the transaction permanent only after the complete call succeeds.

Transaction improvement: the source code reports errors and closes the connection, and an uncommitted transaction is normally rolled back on close. For explicit intent, call connection.rollback() inside the exception path whenever a connection exists, then log how many rows were intended and the batch identifier.

Lab Manual 8 — Batch Insert, Commit and Rollback Drill

10 min

Objective

Write the full batch in a single transaction, measure the effect of fast_executemany, and confirm that a failed batch rolls back completely rather than leaving partial rows in the table.

Procedure

Record the row count before the insert.
Run the timed batch insert below and note the elapsed time.
Repeat with cursor.fast_executemany = False and compare the two timings.
Run the rollback drill: append one deliberately invalid tuple, run again, and confirm the count does not change.

What each line does

Code	Meaning in this lab
cursor.fast_executemany = True	Enables driver-side batching on Microsoft ODBC drivers. Typically several times faster than row-by-row execution.
cursor.executemany(query, records)	Runs the same prepared statement for every tuple in the list.
connection.commit()	Nothing is permanent until this line runs. Call it once, after the whole batch succeeds.
connection.rollback()	Discards the entire open transaction so no partial batch survives a failure.
finally: connection.close()	Closing without a commit also discards the transaction, but an explicit rollback documents the intent.

Copy-paste practice code

06_batch_insert_timed.py

import time
import pyodbc

connection = connect(DATABASE)
cursor = connection.cursor()
cursor.fast_executemany = True

cursor.execute(f"SELECT COUNT(*) FROM {TABLE}")
rows_before = cursor.fetchone()[0]

started = time.perf_counter()
try:
    cursor.executemany(insert_query, records)
    connection.commit()
    elapsed = time.perf_counter() - started
    print(f"Committed {len(records)} rows in {elapsed:.2f} seconds")
except pyodbc.Error as error:
    connection.rollback()
    print("Rolled back — no rows written. Reason:", error)

cursor.execute(f"SELECT COUNT(*) FROM {TABLE}")
rows_after = cursor.fetchone()[0]
print("Before:", rows_before, "| After:", rows_after, "| Written:", rows_after - rows_before)
connection.close()

Run this only against an approved training database. Each successful run adds another 100 rows because the table has no idempotency key.

Expected result

console output

Committed 100 rows in 0.19 seconds
Before: 0 | After: 100 | Written: 100

Checkpoint

Pass when: Written equals 100 on a good run, and equals 0 on the rollback drill. A partial number means the batch is not running inside one transaction.

If it fails

Symptom	What to check
String data, right truncation	A generated string is longer than its varchar column. Compare with the lengths printed in Lab 5.
Memory spikes on a large batch	`fast_executemany` buffers the whole batch. Insert in chunks of 1,000 to 5,000 rows for very large loads.
Rows appear even though an error was raised	Autocommit is switched on, or commit is being called inside the loop. Keep one commit after the batch.
The insert is very slow	`fast_executemany` is off, or the driver is the legacy 'SQL Server' driver rather than ODBC Driver 17/18.

9. Main Execution and Post-Insert Verification

Check for SQF_DB through master.
Connect to the target database.
Verify dbo.tblEvent.
Compare actual column names with the required list.
Generate 100 timestamped test rows.
Insert and commit the batch.
Run SELECT COUNT(*) and display the new table total.
Close the connection in finally.

The final count proves the table is readable and reports its total size, but it does not independently prove that exactly this batch was inserted. A stronger check records a batch ID or captures before/after counts inside the same controlled process.

Lab Manual 9 — Run the Sequence and Verify the Batch in SQL

8 min

Objective

Run the complete sequence once and verify the result from the database side, using row counts, a sample of the newest rows and a stage-wise summary.

Procedure

Run the program and note the count it prints.
Open SSMS or Azure Data Studio against the same database.
Run the three verification queries below and compare them with the console output.
Clean up only your own test batch when the lab is finished.

What each line does

Code	Meaning in this lab
Check for SQF_DB via master	Fails early and clearly if the instance is wrong.
Verify dbo.tblEvent	Confirms the object, not only the database.
Compare column names	Stops the run before any value is bound.
Generate, insert, commit	One transaction for the whole batch.
SELECT COUNT(*)	Proves the table is readable and reports its size, though it does not by itself prove which rows are yours — a batch id does that.
finally: close	The connection is released on both the success and the failure path.

Copy-paste practice code

07_verify_batch.sql

-- 1. How many rows does the table hold now?
SELECT COUNT(*) AS TotalRows FROM dbo.tblEvent;

-- 2. Look at the ten newest events
SELECT TOP 10 DT, TM, SQF_No, ChargeNo, Event_From, Event_To,
       Temp_Set, Temp_Act, Fan_Status
FROM dbo.tblEvent
ORDER BY DT DESC, TM DESC;

-- 3. Stage-wise summary of the inserted cycle
SELECT Event_From AS Stage,
       COUNT(*) AS Records,
       MIN(Temp_Act) AS MinTemp,
       MAX(Temp_Act) AS MaxTemp
FROM dbo.tblEvent
GROUP BY Event_From
ORDER BY Records DESC;

Run these in SSMS against the same database the script used. Query 3 is the quickest way to confirm that all six furnace stages were generated.

Expected result

console output

TotalRows
---------
100

Stage              Records  MinTemp  MaxTemp
Heating            25       318.44   849.21
Soaking            20       846.02   854.88
Cooling            15       101.77   431.60
Charging           20        98.12   286.45
Oil Quenching      15       112.09   792.33
Charge Discharged   5        92.41    99.06

Cleanup (optional)

08_cleanup_test_batch.sql

-- STEP 1 — always review the count before deleting anything
SELECT COUNT(*) AS RowsToDelete
FROM dbo.tblEvent
WHERE DT = CAST(GETDATE() AS DATE);

-- STEP 2 — run only after the count above matches your test batch
DELETE FROM dbo.tblEvent
WHERE DT = CAST(GETDATE() AS DATE);

Run the SELECT first, every time. A DELETE without a reviewed WHERE clause is the fastest way to lose a plant table.

Checkpoint

Pass when: The total matches the console figure, all six stage names appear, and each stage's temperature range matches the cycle table in Section 6.

If it fails

Symptom	What to check
Count is 100 higher than expected	The script was run twice. Remove only the extra batch using the cleanup query below.
Rows exist but temperatures are NULL	The tuple order slipped. Re-run Lab 6 and compare value order against `REQUIRED_COLUMNS`.
Newest rows are not at the top	Order by both `DT` and `TM`; ordering by date alone leaves the minutes unsorted.

10. Complete Python Program

insert_events.py

"""
Check database, table and columns, then insert 100 dummy records.

Install once:
python -m pip install pyodbc
"""

import random
from datetime import datetime, timedelta

import pyodbc


# ============================================================
# SQL SERVER SETTINGS
# ============================================================

SERVER = r"SOFTWELL\WINCC"
DATABASE = "SQF_DB"
TABLE = "dbo.tblEvent"
DRIVER = "ODBC Driver 17 for SQL Server"
NUMBER_OF_RECORDS = 100

# Required table columns
REQUIRED_COLUMNS = [
    "DT",
    "TM",
    "SQF_No",
    "ChargeNo",
    "Event_From",
    "Event_To",
    "Temp_Set",
    "Temp_Act",
    "Cp_Set",
    "Cp_Act",
    "Oil_Set",
    "Oil_Act",
    "Jacket_Set",
    "Jacket_Act",
    "Fan_Status",
]


# ============================================================
# CREATE SQL SERVER CONNECTION
# ============================================================

def connect(database_name):
    """Connect to the selected SQL Server database."""
    return pyodbc.connect(
        f"DRIVER={{{DRIVER}}};"
        f"SERVER={SERVER};"
        f"DATABASE={database_name};"
        "Trusted_Connection=yes;"
        "TrustServerCertificate=yes;",
        timeout=15,
    )


# ============================================================
# CHECK DATABASE
# ============================================================

def database_exists():
    """Return True when SQF_DB exists."""
    connection = connect("master")
    cursor = connection.cursor()
    cursor.execute(
        "SELECT COUNT(*) FROM sys.databases WHERE name = ?",
        DATABASE,
    )
    result = cursor.fetchone()[0] > 0
    connection.close()
    return result


# ============================================================
# CHECK TABLE
# ============================================================

def table_exists(connection):
    """Return True when dbo.tblEvent exists."""
    cursor = connection.cursor()
    cursor.execute("""
        SELECT COUNT(*)
        FROM INFORMATION_SCHEMA.TABLES
        WHERE TABLE_SCHEMA = 'dbo'
          AND TABLE_NAME = 'tblEvent'
    """)
    return cursor.fetchone()[0] > 0


# ============================================================
# CHECK REQUIRED COLUMNS
# ============================================================

def get_missing_columns(connection):
    """Return a list of missing table columns."""
    cursor = connection.cursor()
    cursor.execute("""
        SELECT COLUMN_NAME
        FROM INFORMATION_SCHEMA.COLUMNS
        WHERE TABLE_SCHEMA = 'dbo'
          AND TABLE_NAME = 'tblEvent'
    """)
    available_columns = {
        row.COLUMN_NAME for row in cursor.fetchall()
    }
    return [
        column
        for column in REQUIRED_COLUMNS
        if column not in available_columns
    ]


# ============================================================
# GENERATE ONE FURNACE RECORD
# ============================================================

def generate_record(record_number, record_time):
    """Generate one realistic SQF furnace record."""

    # One complete furnace cycle contains 20 records
    cycle = (record_number - 1) % 20

    # Furnace number cycles through 1, 2 and 3
    sqf_no = ((record_number - 1) % 3) + 1

    # One charge number is used for every 10 records
    charge_sequence = ((record_number - 1) // 10) + 1
    charge_no = (
        f"CHG-{record_time:%Y%m%d}-"
        f"{charge_sequence:03d}"
    )

    # --------------------------------------------------------
    # Stage 1: Charging
    # --------------------------------------------------------
    if cycle <= 3:
        event_from = "Furnace Ready"
        event_to = "Charging Started"
        temp_set = 850.0
        temp_act = 100 + cycle * 80 + random.uniform(-4, 4)
        cp_set = 0.80
        cp_act = random.uniform(0.18, 0.25)
        oil_set = 80.0
        oil_act = random.uniform(32, 38)
        jacket_set = 40.0
        jacket_act = random.uniform(29, 32)
        fan_status = "OFF"

    # --------------------------------------------------------
    # Stage 2: Heating
    # --------------------------------------------------------
    elif cycle <= 8:
        event_from = "Charging Completed"
        event_to = "Heating"
        temp_set = 850.0
        temp_act = (
            420
            + (cycle - 4) * 100
            + random.uniform(-4, 4)
        )
        cp_set = 0.80
        cp_act = (
            0.60
            + (cycle - 4) * 0.04
            + random.uniform(-0.02, 0.02)
        )
        oil_set = 80.0
        oil_act = random.uniform(42, 50)
        jacket_set = 40.0
        jacket_act = random.uniform(33, 36)
        fan_status = "ON"

    # --------------------------------------------------------
    # Stage 3: Soaking
    # --------------------------------------------------------
    elif cycle <= 12:
        event_from = "Heating"
        event_to = "Soaking"
        temp_set = 850.0
        temp_act = random.uniform(846, 853)
        cp_set = 0.80
        cp_act = random.uniform(0.78, 0.82)
        oil_set = 80.0
        oil_act = random.uniform(55, 62)
        jacket_set = 40.0
        jacket_act = random.uniform(36, 39)
        fan_status = "ON"

    # --------------------------------------------------------
    # Stage 4: Oil quenching
    # --------------------------------------------------------
    elif cycle <= 15:
        event_from = "Soaking Completed"
        event_to = "Oil Quenching"
        temp_set = 850.0
        temp_act = (
            820
            - (cycle - 13) * 80
            + random.uniform(-4, 4)
        )
        cp_set = 0.80
        cp_act = random.uniform(0.73, 0.79)
        oil_set = 80.0
        oil_act = random.uniform(73, 79)
        jacket_set = 40.0
        jacket_act = random.uniform(38, 41)
        fan_status = "ON"

    # --------------------------------------------------------
    # Stage 5: Cooling
    # --------------------------------------------------------
    elif cycle <= 18:
        event_from = "Oil Quenching"
        event_to = "Cooling"
        temp_set = 100.0
        temp_act = (
            400
            - (cycle - 16) * 120
            + random.uniform(-4, 4)
        )
        cp_set = 0.20
        cp_act = random.uniform(0.22, 0.28)
        oil_set = 80.0
        oil_act = random.uniform(76, 81)
        jacket_set = 40.0
        jacket_act = random.uniform(39, 42)
        fan_status = "ON"

    # --------------------------------------------------------
    # Stage 6: Charge discharged
    # --------------------------------------------------------
    else:
        event_from = "Cooling Completed"
        event_to = "Charge Discharged"
        temp_set = 100.0
        temp_act = random.uniform(92, 99)
        cp_set = 0.20
        cp_act = random.uniform(0.18, 0.22)
        oil_set = 80.0
        oil_act = random.uniform(70, 75)
        jacket_set = 40.0
        jacket_act = random.uniform(37, 40)
        fan_status = "OFF"

    return (
        record_time,
        record_time.strftime("%H:%M:%S"),
        sqf_no,
        charge_no,
        event_from,
        event_to,
        round(temp_set, 2),
        round(temp_act, 2),
        round(cp_set, 2),
        round(cp_act, 2),
        round(oil_set, 2),
        round(oil_act, 2),
        round(jacket_set, 2),
        round(jacket_act, 2),
        fan_status,
    )


# ============================================================
# GENERATE 100 RECORDS
# ============================================================

def generate_records():
    """Generate 100 records at one-minute intervals."""
    records = []

    start_time = (
        datetime.now().replace(microsecond=0)
        - timedelta(minutes=NUMBER_OF_RECORDS - 1)
    )

    for number in range(1, NUMBER_OF_RECORDS + 1):
        record_time = start_time + timedelta(
            minutes=number - 1
        )
        records.append(
            generate_record(number, record_time)
        )

    return records


# ============================================================
# INSERT RECORDS
# ============================================================

def insert_records(connection, records):
    """Insert all generated records into dbo.tblEvent."""
    insert_query = """
        INSERT INTO dbo.tblEvent
        (
            DT,
            TM,
            SQF_No,
            ChargeNo,
            Event_From,
            Event_To,
            Temp_Set,
            Temp_Act,
            Cp_Set,
            Cp_Act,
            Oil_Set,
            Oil_Act,
            Jacket_Set,
            Jacket_Act,
            Fan_Status
        )
        VALUES
        (
            ?, ?, ?, ?, ?,
            ?, ?, ?, ?, ?,
            ?, ?, ?, ?, ?
        )
    """

    cursor = connection.cursor()
    cursor.fast_executemany = True
    cursor.executemany(
        insert_query,
        records,
    )
    connection.commit()


# ============================================================
# MAIN PROGRAM
# ============================================================

connection = None

try:
    print("=" * 65)
    print("SQF DATABASE VERIFICATION AND DUMMY INSERT")
    print("=" * 65)

    # Check 1: Database
    if not database_exists():
        raise RuntimeError(
            f"Database [{DATABASE}] is not available."
        )

    print(f"Database [{DATABASE}] verified.")
    connection = connect(DATABASE)

    # Check 2: Table
    if not table_exists(connection):
        raise RuntimeError(
            f"Table [{TABLE}] is not available."
        )

    print(f"Table [{TABLE}] verified.")

    # Check 3: Columns
    missing_columns = get_missing_columns(connection)

    if missing_columns:
        raise RuntimeError(
            "Missing columns: " + ", ".join(missing_columns)
        )

    print("All required columns verified.")

    # Generate and insert records
    dummy_records = generate_records()
    insert_records(
        connection,
        dummy_records,
    )

    print(
        f"{NUMBER_OF_RECORDS} dummy records "
        "inserted successfully."
    )

    # Display total number of records
    cursor = connection.cursor()
    cursor.execute("SELECT COUNT(*) FROM dbo.tblEvent")
    total_records = cursor.fetchone()[0]

    print(f"Total records in table: {total_records}")
    print("=" * 65)
    print("PROGRAM COMPLETED SUCCESSFULLY")
    print("=" * 65)

except Exception as error:
    print("Program stopped:")
    print(error)

finally:
    if connection is not None:
        connection.close()
        print("SQL Server connection closed.")

Lab Manual 10 — Run the Complete Program End to End

12 min

Objective

Execute the finished program on a clean machine so the whole workflow — connect, verify, generate, insert, confirm — runs in one command.

Procedure

Create a project folder and a virtual environment so pyodbc is not installed system-wide.
Install pyodbc inside the environment.
Save the Section 10 listing as insert_events.py.
Edit only four constants: SERVER, DATABASE, TABLE and DRIVER.
Run the program and compare the console output with the expected output below.

What each line does

Code	Meaning in this lab
python -m venv .venv	Creates an isolated environment so lab packages never disturb other projects on the PC.
.venv\Scripts\activate	Activates the environment on Windows. The prompt changes to show `(.venv)`.
pip install pyodbc	Installs the ODBC binding. The ODBC driver itself is a separate Windows installation.
NUMBER_OF_RECORDS	Start with a small number such as 5 on a new database, then raise it to 100.
if __name__ == "__main__":	Lets the file be imported by another script or a unit test without inserting anything.

Copy-paste practice code

setup_and_run.bat

cd C:\SoftwellLabs\python-sql
python -m venv .venv
.venv\Scripts\activate
python -m pip install --upgrade pip
python -m pip install pyodbc
python insert_events.py

On macOS or Linux, activate with source .venv/bin/activate and install the Microsoft ODBC driver for your distribution first.

Expected result

console output

Database SQF_DB found.
Table dbo.tblEvent found.
All required columns present.
Generated 100 records.
Inserted 100 records successfully.
Total rows in dbo.tblEvent: 100

Checkpoint

Pass when: All six lines appear in order, the inserted count matches NUMBER_OF_RECORDS, and the total row count agrees with the SQL query from Lab 9.

If it fails

Symptom	What to check
ModuleNotFoundError: No module named 'pyodbc'	The virtual environment is not active, or the install ran against a different Python. Run `where python` and confirm it points inside `.venv`.
The script stops after the column check	A required column is missing. Lab 5 names it exactly.
It runs but nothing appears in SSMS	You are querying a different instance or database than the script used. Confirm with `SELECT @@SERVERNAME, DB_NAME()`.
Rows double on every run	Expected behaviour — the table has no unique key. Add a batch id before reusing this pattern anywhere real.

11. Troubleshooting and Production Improvements

Problem	Likely cause	Action
Driver not found	ODBC Driver 17 missing/name mismatch	Check Windows ODBC Drivers and update `DRIVER`
Login failed	Windows identity lacks access	Confirm service/user identity and least-privilege grants
Database/table unavailable	Wrong instance, database or schema	Verify exact target before enabling writes
Missing columns	Schema version mismatch	Migrate schema or update the approved mapping
String/binary truncation	Text exceeds column length	Compare generated lengths with SQL definitions
Conversion/overflow error	Python value incompatible with SQL type	Inspect type, precision, scale and date/time definitions
Duplicate test data	Script rerun without idempotency	Add batch ID/unique key or delete only an approved test batch
Partial/uncertain result	Failure near commit or connectivity loss	Use explicit transaction handling and batch-level verification

Recommended production upgrades

Move server/database values into validated configuration, not source code.
Use a dedicated least-privilege account and trusted TLS certificate.
Add a unique event key or batch ID for idempotent retries.
Validate full schema metadata and allowable ranges before insertion.
Use structured logging instead of only print().
Seed randomness for repeatable tests, and label every dummy record.
Wrap execution in main() with the if __name__ == "__main__": guard for reusable imports and unit testing.

Hands-On Lab: Insert and Verify a Test Batch

Hands-on

Before you start

Use an approved test database—not a production furnace database.
Take a backup or prepare a clearly scoped batch cleanup method.
Confirm all expected SQL data types and permissions.
Estimated time: 30 minutes.

Run verification without inserting

Temporarily stop execution before generate_records(). Confirm database, table and all required columns pass.

The script identifies the exact approved target and performs no write.

Inspect generated tuples

Set a repeatable random seed, generate a small batch of five records and print each tuple before insertion.

Each tuple has 15 correctly ordered values compatible with the table schema.

Insert one controlled batch

Record the starting row count, insert the approved batch and commit once.

The batch succeeds as one transaction and the count increases by the intended quantity.

Verify values and recovery

Read the inserted batch back, verify stage/value ranges, then test a deliberately invalid row against a disposable transaction and roll it back.

Valid rows match the generated data and the failed test leaves no partial records.

Frequently asked questions

How does Python insert values into SQL Server?

Connect through pyodbc, create a cursor, prepare an INSERT statement with question-mark parameters, pass row values with execute() or executemany(), and commit the transaction after success.

Why use parameterized SQL instead of string formatting?

Parameters keep data separate from SQL syntax, reduce injection risk and allow the ODBC driver to convert compatible Python values to SQL Server types.

What does fast_executemany do?

With supported drivers, it sends batches more efficiently than executing each row separately. Test memory use and compatibility with your driver and data types before applying it to large production loads.

How do I insert a single row instead of a batch?

Use cursor.execute(insert_query, record) with one 15-value tuple, then call connection.commit(). The statement is identical — only the cursor method changes.

Why does pyodbc report “COUNT field incorrect or syntax error”?

The number of ? markers does not match the number of values in the tuple. Build the statement from your column list so both counts always come from one place.

Should I commit inside the loop or after the batch?

Commit once after the batch. A commit per row forces a transaction-log flush every time and also removes the all-or-nothing guarantee you want from a data-logging script.

How do I stop duplicate rows when the script is rerun?

Add a unique key or a batch id column, then either insert with a NOT EXISTS check or use MERGE. Without one, every run of this training script adds another 100 rows.

Can the same code write to SQL Server from a WinCC or SCADA PC?

Yes. Install the same ODBC driver on that PC, use a dedicated least-privilege SQL login instead of a personal Windows account, and run the script through Windows Task Scheduler or a service.

Does pyodbc use ? or %s placeholders?

pyodbc uses qmark paramstyle, so the marker is ?. %s belongs to pymssql and psycopg2, and :name to cx_Oracle.

Reviewed by Bhawesh Kumar SinghIndustrial Automation Trainer and Industry 4.0 Consultant · Softwell Automation · 21+ years industry experience

Python Scripting TrainingPython automation, database connectivity and industrial reporting projects.View syllabus →SQL Reporting TrainingSQL Server design, queries, plant data and automated reports.View syllabus →Industry 4.0 TrainingOPC UA, SQL, Python, dashboards and IT/OT integration.View syllabus →

Insert Data into SQL Server Using Python and pyodbc

Lab Overview

Prerequisites / What You’ll Need

Python pyodbc INSERT: Single Rows, Batches and Transactions

Related Search Topics

1. Insert Data into SQL Server: Python Workflow

Python Events → pyodbc → SQL Server INSERT

Connect

Verify Target

Generate Events

Parameterized INSERT

Batch + Transaction

Confirm

Lab Manual Index: 8 Guided Exercises

2. Requirements and Safety Conditions

3. Configure the SQL Server Connection

Lab Manual 3 — Configure and Prove the SQL Server Connection

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Checkpoint

If it fails

4. Verify the Database and Table

Lab Manual 4 — Verify the Database and Table Exist

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Checkpoint

If it fails

5. Validate All Required Columns

Lab Manual 5 — Validate All 15 Required Columns

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Checkpoint

If it fails

6. Generate Realistic Furnace Records

Lab Manual 6 — Generate and Inspect the Furnace Records

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Checkpoint

If it fails

7. Create a Parameterized pyodbc INSERT Query

Lab Manual 7 — Build the Parameterized INSERT Safely

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Checkpoint

If it fails

8. Insert Multiple Rows with executemany and Transactions

Lab Manual 8 — Batch Insert, Commit and Rollback Drill

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Checkpoint

If it fails

9. Main Execution and Post-Insert Verification

Lab Manual 9 — Run the Sequence and Verify the Batch in SQL

Objective

Procedure

What each line does

Copy-paste practice code

Expected result

Cleanup (optional)

Checkpoint

If it fails

10. Complete Python Program