Insert Data into SQL Server Using Python and pyodbc

• 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.

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.

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.

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.

3. Configure the SQL Server Connection

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.

4. Verify the Database and Table

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.

5. Validate All Required 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",
]

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.

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.

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.

Create a Parameterized pyodbc 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
(
    ?, ?, ?, ?, ?, ?,
    ?, ?, ?, ?, ?, ?,
    ?, ?, ?
)

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.

Insert Multiple Rows with executemany and Transactions

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.

9. Main Execution and Post-Insert Verification

1. Check for SQF_DB through master.
2. Connect to the target database.
3. Verify dbo.tblEvent.
4. Compare actual column names with the required list.
5. Generate 100 timestamped test rows.
6. Insert and commit the batch.
7. Run SELECT COUNT(*) and display the new table total.
8. 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.

10. Complete Python Program

"""
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.")

11. Troubleshooting and Production Improvements

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.