Build a Python SQL Server Pipeline

Validate CSV orders and load them reliably into SQL Server with Python.

Introduction

30 Second Summary

Order files often look fine until one broken date or impossible quantity slips into the totals. A dependable process catches those records before they distort a report.

In this project, you will build a local orders-ingestion data pipeline with Python and Microsoft SQL Server. Each run turns a messy CSV feed into trusted rows.

What You'll Build

You run one Python command and watch four accepted orders reach SQL Server while two malformed records land in a rejection file with clear reasons.

By the end of this project, you'll have:

  • A one-command batch run that reports four valid rows, two rejected rows, four staged rows, and four loaded rows in the terminal.
  • A trusted data trail you can inspect in dbo.Orders, output/rejected_orders.csv, and audit.PipelineRun.
  • A published GitHub repository that proves idempotency by keeping target_row_count at 4 after a second run.
  • Secret Mission: Parameterize the pipeline so a correction batch updates order 1002, adds order 1007, and rejects one malformed row.

Are there any prerequisites?

The project assumes your Windows setup already includes Visual Studio Code, Git, a GitHub account, and access to a writable SQL Server instance where you can create DataEngineeringLab or use an existing development database.

You'll need basic SQL and data warehousing knowledge plus Python 3.10 or later.

Before We Start

Before the hands-on work begins, commit to building a local orders-ingestion ETL pipeline that turns an imperfect orders CSV into trusted SQL Server data. Your pipeline will preserve invalid records in a rejection CSV. It will idempotently replace matching orders so reruns never duplicate trusted data.

Prepare the Windows Workspace

A reliable ETL pipeline needs a repeatable workspace before it can process any orders. Each dependency needs its own proof before data enters the picture.

You will verify the Python runtime inside an isolated environment. You will also install the driver for SQL Server before testing the editor connection separately.

In this step, get ready to:
  • Create the project workspace with its directories and configuration files.
  • Prepare an isolated Python environment with the pinned database driver.
  • Prove the VS Code extension can connect to your local SQL Server instance.
Create the project workspace

A Visual Studio Code workspace keeps every project file under one named folder. Opening that folder also gives the integrated terminal a predictable starting location.

  • Press the Windows key to open Windows search.
  • Type Visual Studio Code and press Enter to open it.
  • Select File from the menu bar.
  • Select Open Folder....
  • Select your Desktop in the folder dialog.
  • Select New Folder.
  • Enter sql-server-batch-pipeline as the folder name.
  • Select Select Folder to open the workspace.

You should see sql-server-batch-pipeline at the top of the Explorer. That confirms VS Code is working inside the intended project folder.

How will the folders divide the pipeline?

  • The data directory holds incoming CSV files.
  • The src directory holds the Python pipeline.
  • The sql directory holds database setup and verification scripts.
  • The output directory holds generated rejection files.
  • Select sql-server-batch-pipeline in the Explorer.
  • Use New Folder to create the data, src, sql, and output directories.
  • Use New File... to create requirements.txt inside sql-server-batch-pipeline.
  • Add the pinned database driver to requirements.txt by pasting:
mssql-python==1.14.0

Why pin the driver version?

The exact mssql-python==1.14.0 pin gives every installation the same driver release. That removes version drift from future pipeline runs.

  • Save requirements.txt.
  • Confirm the Explorer shows requirements.txt beside the four project directories.
  • Use New File... to create .gitignore inside sql-server-batch-pipeline.
  • Add the generated-file exclusions to .gitignore by pasting:
tutorial-env/
__pycache__/
*.pyc
output/

What will Git ignore?

  • The tutorial-env/ entry excludes the local virtual environment.
  • The __pycache__/ entry excludes Python cache directories.
  • The *.pyc entry excludes compiled Python files.
  • The output/ entry excludes generated rejection output.
  • Save .gitignore.
  • Confirm the Explorer shows .gitignore beside requirements.txt.

Files missing from the Explorer?

Confirm that sql-server-batch-pipeline is the folder shown at the top of the Explorer. Recreate any missing item inside that folder.

If a file was created in the wrong location, help me check my VS Code workspace structure.

✔️ Awesome, I've got everything!

  • Save both configuration files before continuing.
  • Keep the sql-server-batch-pipeline workspace open in VS Code.

ⓧ I'd like to double check the full code

Compare each file with these complete versions.

mssql-python==1.14.0

What should requirements.txt contain?

The file contains one pinned dependency. Match the package name and version exactly.

tutorial-env/
__pycache__/
*.pyc
output/

What should .gitignore contain?

The file contains four exclusions. Each pattern occupies its own line.

Verify Python and install the driver

The driver requires Python 3.10 or later. Checking the active version now prevents an incompatible environment from carrying into the pipeline.

  • Select View from the VS Code menu bar.
  • Select Terminal to open the integrated terminal.
  • Use the terminal dropdown to select Windows PowerShell.
  • Check the active Python version by running:
python --version

What does this command check?

The command prints the Python version that the current PowerShell session uses. This is the runtime that creates your virtual environment.

✔️ I see Python 3.10 or later

Your Python runtime meets the driver requirement. You can create the project environment below.

ⓧ I see an older Python version

The current python command points to an older installation. Install a supported version before creating the environment.

  • Open the official Python Windows guide for installation guidance.
  • Download a Windows installer for Python 3.10 or later.
  • Run the downloaded installer.
  • Select Add Python to PATH on the first installer page.
  • Select Install Now.
  • Restart Visual Studio Code when installation finishes.
  • Return to the sql-server-batch-pipeline workspace.
  • Check the newly active version by running:
python --version

What should the new check prove?

The output should now report Python 3.10 or later. The newer runtime is ready to create the project environment.

Still seeing the older version?

Run the newer Python installer again. Select Modify before enabling Add Python to PATH.

If the older version still takes priority, help me make my newer Python installation active in Windows PowerShell.

ⓧ Command not found

PowerShell cannot currently reach a Python installation. Install Python before creating the environment.

  • Open the official Python Windows guide for installation guidance.
  • Download a Windows installer for Python 3.10 or later.
  • Run the downloaded installer.
  • Select Add Python to PATH on the first installer page.
  • Select Install Now.
  • Restart Visual Studio Code when installation finishes.
  • Return to the sql-server-batch-pipeline workspace.
  • Confirm that PowerShell can now reach Python by running:
python --version

What should this check prove?

The command should now print Python 3.10 or later. PowerShell can use that runtime for the project.

Python still unavailable?

Run the installer again. Select Modify before enabling Add Python to PATH.

If the command remains unavailable, help me make Python available in Windows PowerShell.

A virtual environment gives this project its own package installation. Other Python projects on your computer remain separate.

  • Create tutorial-env before activating it by running these commands:
python -m venv tutorial-env
tutorial-env\Scripts\activate

What do these commands do?

  • The first command creates an isolated Python environment in tutorial-env.
  • The second command makes that environment active in the current PowerShell session.
  • Check the start of the PowerShell prompt for the environment name.
  • Keep this terminal open so the environment remains active.

Environment name missing from the prompt?

Confirm that the terminal is running from the sql-server-batch-pipeline workspace. Run the activation command again from that location.

If PowerShell blocks activation, help me activate tutorial-env safely in Windows PowerShell.

The isolated environment is now ready for mssql-python. The package pin in requirements.txt controls the version that pip installs.

  • Install the pinned dependency into tutorial-env by running:
python -m pip install -r requirements.txt

What does this installation command do?

Python asks pip to read requirements.txt inside the active environment. Pip installs the exact driver release recorded in that file.

Why mssql-python instead of pyodbc?

Microsoft provides mssql-python as its first-party Python driver for SQL Server. Its API also includes the bulk-copy feature used later in this project.

On Windows, normal pip installation pulls the required companion package automatically. A separate Microsoft ODBC Driver installation is unnecessary for this project.

  • Check the installed packages by running:
python -m pip list

What does the package list confirm?

The list reports every package installed in the active environment. Look for mssql-python with version 1.14.0.

Good progress. The project now has an isolated Python runtime with the required SQL Server driver.

Driver missing from the package list?

Check that the PowerShell prompt still shows the active environment. Confirm that requirements.txt contains the exact package pin.

If installation still fails, help me diagnose the mssql-python installation.

Install the SQL Server extension

The SQL Server (mssql) extension for Visual Studio Code provides the database connection dialog inside the editor. Testing it now proves that your local SQL Server instance is reachable before the pipeline uses Python.

  • Press Ctrl+Shift+X to open Extensions.
  • Search for mssql.
  • Select SQL Server (mssql).
  • Select Install.

You should see the MSSQL icon in the Activity Bar. The editor can now manage SQL Server connections.

MSSQL icon missing?

Confirm that the extension page shows SQL Server (mssql) as installed. Restart Visual Studio Code if the Activity Bar has not refreshed.

If the icon remains missing, help me verify the SQL Server extension installation.

Keep Credentials Out of Files

Connection credentials stay inside the Connection Dialog. Do not add a SQL Login password to requirements.txt or .gitignore.

  • Select the MSSQL icon in the Activity Bar.
  • Select Add Connection.
  • Select MS SQL: Add Connection if the Command Palette asks you to choose a command.
  • Enter the server name from the local SQL Server connection you already use.
  • Record that server name here: your local SQL Server name.

Use the authentication path that matches your existing local SQL Server connection.

Windows Authentication

  • Choose Windows Authentication in the Connection Dialog.
  • Select Test connection.
  • Select Connect after the test succeeds.

SQL Login

  • Choose the same SQL Login authentication method used by your existing local connection.
  • Enter your existing local development user name.
  • Enter the password from your secure storage.
  • Select Test connection.
  • Select Connect after the test succeeds.

You should see your local SQL Server connection inside the MSSQL view. This proves the editor can reach the instance that hosts the pipeline database.

Connection test unsuccessful?

Confirm that the server name matches the connection you already use. Check that the local SQL Server instance is running.

If the authentication test still fails, help me diagnose my local SQL Server connection in VS Code.

Before you run the final check, which Python version and package entry do you expect to see?

  • Verify the active runtime and installed packages by running:
python --version
python -m pip list

What do these final checks prove?

The first command confirms the active Python runtime. The second command confirms the packages available to that runtime.

You should see Python 3.10 or later. You should also see mssql-python at version 1.14.0.

  • Confirm the MSSQL icon remains visible in the Activity Bar.
  • Select the MSSQL icon.

You should see the tested local SQL Server connection in the MSSQL view. Your workspace can now support the Python and database sides of the pipeline.

That foundation is in place. The project files, isolated environment, driver, extension, and database connection are ready to work together.

Your Windows workspace is ready. Next, you will feed it a messy orders CSV and separate trusted rows from rejected records.

Validate the Orders Feed

The workspace you prepared can now run Python inside tutorial-env. Its database tools can also reach SQL Server.

Raw CSV files can contain values that cannot safely enter typed database columns. This step uses schema validation to separate trustworthy orders from malformed records.

A reliable batch also needs proof that accepted data survives the process. You will see what happens to those rows after this first version exits.

In this step, get ready to:
  • Create a six-record orders feed with two deliberate data problems.
  • Validate the six records against typed fields.
  • Persist rejected records with their reasons.
Create the messy source feed

A useful validator needs both successful rows and realistic failures. This feed includes a malformed date plus an order with zero quantity.

  • Select the data folder in the VS Code Explorer sidebar.
  • Click the new-file icon beside the folder name.
  • Enter orders.csv in the filename field.
  • Press Enter.
  • Paste these six source records into data/orders.csv:
order_id,customer_name,order_date,quantity,unit_price
1001,Ada Retail,2026-09-01,2,19.95
1002,Northwind Bikes,2026-09-02,1,250.00
1003,City Market,not-a-date,3,12.50
1004,Contoso Cafe,2026-09-03,0,8.75
1005,Fabrikam Shop,2026-09-04,4,15.00
1006,Adventure Works,2026-09-05,1,99.99

What is hidden in this feed?

  • Order 1003 contains not-a-date where an ISO date belongs.
  • Order 1004 has a quantity of 0.
  • The remaining four records contain values that can become typed Python objects.
  • Save data/orders.csv.
  • Count the source records beneath the header.

You should see six source records. Four are valid orders.

Does the feed look different?

Check that the header contains five comma-separated field names. Make sure every source record stays on one line.

Confirm that the file is named orders.csv inside the existing data folder.

Help me check the formatting of my orders CSV.

Convert each row into typed values

CSV readers return text values. The validator converts each field into the Python type expected by the future SQL Server table.

  • Select the src folder in the VS Code Explorer sidebar.
  • Click the new-file icon beside the folder name.
  • Enter pipeline.py in the filename field.
  • Press Enter.
  • Add the imports plus project paths to src/pipeline.py by pasting this code:
import csv
from datetime import date
from decimal import Decimal, InvalidOperation
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parents[1]
INPUT_FILE = BASE_DIR / "data" / "orders.csv"
REJECT_FILE = BASE_DIR / "output" / "rejected_orders.csv"
EXPECTED_FIELDS = [
    "order_id",
    "customer_name",
    "order_date",
    "quantity",
    "unit_price",
]

What does this setup do?

  • The imports provide CSV parsing plus typed dates.
  • The imports also provide exact decimal values for prices.
  • BASE_DIR anchors every file path to the sql-server-batch-pipeline workspace.
  • EXPECTED_FIELDS defines the required source schema in its expected order.
  • Save src/pipeline.py.
  • Compare the three path constants with the folders in the Explorer sidebar.

Each path should begin from BASE_DIR. The input path should end at data/orders.csv.

Do the paths point elsewhere?

Check that pipeline.py is inside src. The parents[1] expression depends on that location.

Help me check the paths in my validation script.

Type conversion catches malformed values before they reach a database. A rejected row keeps its original source values so you can diagnose the failure later.

  • Add the first version of validate_orders() below EXPECTED_FIELDS by pasting this code:
def validate_orders(input_file: Path):
    valid_rows = []
    rejected_rows = []

    with input_file.open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")

        for row_number, row in enumerate(reader, start=2):
            try:
                order_id = int(row["order_id"])
                customer_name = row["customer_name"].strip()
                order_date = date.fromisoformat(row["order_date"].strip())
                quantity = int(row["quantity"])
                unit_price = Decimal(row["unit_price"]).quantize(Decimal("0.01"))

                valid_rows.append(
                    (order_id, customer_name, order_date, quantity, unit_price)
                )
            except (ValueError, TypeError, InvalidOperation) as error:
                rejected = {
                    field: row.get(field, "") for field in EXPECTED_FIELDS
                }
                rejected["reject_reason"] = f"row {row_number}: {error}"
                rejected_rows.append(rejected)

    return valid_rows, rejected_rows

How does conversion separate rows?

  • csv.DictReader maps each source value to its header name.
  • date.fromisoformat() rejects the malformed date during conversion.
  • Decimal preserves exact monetary values at two decimal places.
  • The exception handler keeps the source fields plus a row-specific rejection reason.
  • Save src/pipeline.py.
  • Confirm that validate_orders() returns one list for accepted rows plus one list for rejected rows.

Seeing red underlines in the validator?

Check the indentation beneath with plus for plus try. Python uses indentation to identify each code block.

Confirm that every dictionary key matches a name in EXPECTED_FIELDS.

Help me find an indentation or syntax problem in validate_orders().

Successful type conversion only proves that a value has the correct shape. Business rules also protect the pipeline from valid-looking values such as a zero quantity.

  • Find the price conversion followed by the accepted-row append in validate_orders().
                unit_price = Decimal(row["unit_price"]).quantize(Decimal("0.01"))

                valid_rows.append(

What is missing here?

The current code converts the values before accepting the row. It still needs rules for positive identifiers plus required names plus valid quantities plus non-negative prices.

  • Insert these validation checks between the price conversion and valid_rows.append():
                unit_price = Decimal(row["unit_price"]).quantize(Decimal("0.01"))

                if order_id <= 0:
                    raise ValueError("order_id must be positive")
                if not customer_name:
                    raise ValueError("customer_name is required")
                if len(customer_name) > 100:
                    raise ValueError("customer_name exceeds 100 characters")
                if quantity <= 0:
                    raise ValueError("quantity must be positive")
                if unit_price < 0:
                    raise ValueError("unit_price cannot be negative")

                valid_rows.append(

Why add business rules?

These checks enforce the values the warehouse can trust. The zero quantity now follows the same rejection path as the malformed date.

The accepted tuples contain an integer identifier plus a trimmed name plus a date plus an integer quantity plus a decimal price.

  • Save src/pipeline.py.
  • Confirm that the five validation clauses appear before valid_rows.append().

Is the zero quantity still accepted?

Check that the quantity rule uses quantity <= 0. Make sure the rule is indented inside the try block.

Help me place the validation rules correctly.

Persist the rejected rows

Rejected records need a durable output that someone can inspect after the script exits. The rejection writer preserves every original field plus the reason that validation failed.

  • Add write_rejections() below validate_orders() by pasting this code:
def write_rejections(reject_file: Path, rejected_rows):
    reject_file.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = EXPECTED_FIELDS + ["reject_reason"]

    with reject_file.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rejected_rows)

What does the rejection writer do?

  • The function confirms that the output directory exists.
  • The output header adds reject_reason after the five source fields.
  • csv.DictWriter writes the header plus every rejected source row.
  • Save src/pipeline.py.
  • Confirm that writer.writeheader() appears before writer.writerows().

Is the rejection writer underlined?

Check that write_rejections() begins at the left edge of the file. It must be separate from validate_orders().

Help me check the write_rejections function.

The entry point connects the validator to the rejection writer. It also exposes the accepted plus rejected row counts in the terminal.

  • Add the pipeline entry point below write_rejections() by pasting this code:
def run_pipeline():
    valid_rows, rejected_rows = validate_orders(INPUT_FILE)
    write_rejections(REJECT_FILE, rejected_rows)

    print(f"Valid rows: {len(valid_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")


if __name__ == "__main__":
    run_pipeline()

How does the entry point complete the flow?

  • run_pipeline() passes the source file into the validator.
  • write_rejections() persists the rejected list.
  • The two print calls expose the final validation counts.
  • The final condition runs the pipeline when you execute the file directly.

✔️ Awesome, I've got everything!

Your validation-only pipeline is assembled. Its entry point now connects the source feed to the validator plus rejection writer.

ⓧ I'd like to double check the full code

import csv
from datetime import date
from decimal import Decimal, InvalidOperation
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parents[1]
INPUT_FILE = BASE_DIR / "data" / "orders.csv"
REJECT_FILE = BASE_DIR / "output" / "rejected_orders.csv"
EXPECTED_FIELDS = [
    "order_id",
    "customer_name",
    "order_date",
    "quantity",
    "unit_price",
]


def validate_orders(input_file: Path):
    valid_rows = []
    rejected_rows = []

    with input_file.open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")

        for row_number, row in enumerate(reader, start=2):
            try:
                order_id = int(row["order_id"])
                customer_name = row["customer_name"].strip()
                order_date = date.fromisoformat(row["order_date"].strip())
                quantity = int(row["quantity"])
                unit_price = Decimal(row["unit_price"]).quantize(Decimal("0.01"))

                if order_id <= 0:
                    raise ValueError("order_id must be positive")
                if not customer_name:
                    raise ValueError("customer_name is required")
                if len(customer_name) > 100:
                    raise ValueError("customer_name exceeds 100 characters")
                if quantity <= 0:
                    raise ValueError("quantity must be positive")
                if unit_price < 0:
                    raise ValueError("unit_price cannot be negative")

                valid_rows.append(
                    (order_id, customer_name, order_date, quantity, unit_price)
                )
            except (ValueError, TypeError, InvalidOperation) as error:
                rejected = {
                    field: row.get(field, "") for field in EXPECTED_FIELDS
                }
                rejected["reject_reason"] = f"row {row_number}: {error}"
                rejected_rows.append(rejected)

    return valid_rows, rejected_rows


def write_rejections(reject_file: Path, rejected_rows):
    reject_file.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = EXPECTED_FIELDS + ["reject_reason"]

    with reject_file.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rejected_rows)


def run_pipeline():
    valid_rows, rejected_rows = validate_orders(INPUT_FILE)
    write_rejections(REJECT_FILE, rejected_rows)

    print(f"Valid rows: {len(valid_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")


if __name__ == "__main__":
    run_pipeline()

How should the full file look?

This reference combines the exact imports plus paths plus functions plus entry point used by the validation-only pipeline.

  • Save src/pipeline.py.
  • Switch back to the activated Windows PowerShell terminal from earlier.

Before you run the script, how many rows do you expect each list to contain?

  • Test the validation pipeline by running this command:
python src/pipeline.py

What should you see?

You should see Valid rows: 4 followed by Rejected rows: 2.

Those counts prove that the malformed date plus zero quantity followed the rejection path.

Are the counts different?

Compare data/orders.csv with the six-record source block above. A changed value can alter the result.

Check that every business rule appears before valid_rows.append().

Help me debug unexpected validation counts.

  • Expand the output folder in the VS Code Explorer sidebar.
  • Select rejected_orders.csv.
  • Check that order 1003 identifies its malformed date.
  • Check that order 1004 identifies the positive-quantity rule.

That check worked. Your pipeline now separates four accepted orders from two rejected orders.

Where did the accepted rows go?

The four accepted tuples lived in valid_rows only while the Python process was running. The process has ended, so those typed values are gone.

The rejection CSV persists because the script wrote it to disk. The next step gives trusted rows a durable SQL Server landing zone.

Your feed now has a clear quality boundary. Next, you will create staging plus target plus audit tables for the accepted rows.

Create the SQL Server Landing Zone

Your validator now separates four accepted orders from two rejected records. The accepted orders still vanish when the Python process ends.

This step creates a durable landing zone in Microsoft SQL Server. Separate storage areas will hold staged rows, trusted orders, plus the history of each pipeline run.

In this step, get ready to:
  • Create the database landing zone.
  • Set the SQL Server connection string for the current PowerShell session.
  • Verify that every new table starts empty.

Why does the pipeline need three zones?

A staging table receives accepted rows before publication. This gives the pipeline a place to verify each batch.

The target table stores trusted orders. The audit log records the status of each pipeline run.

Build the SQL Server landing zone

The setup script creates each database object only when it is missing. This makes the script safe to rerun while you build it in small sections.

  • Switch back to the Explorer sidebar in Visual Studio Code.
  • Use the new-file icon to create sql/setup.sql inside sql-server-batch-pipeline.
  • Select setup.sql in the Explorer sidebar.
  • Select Connect in the SQL Server (mssql) extension toolbar.
  • Choose the tested SQL Server connection from earlier.
  • Choose master as the database.
  • Create the database foundation by copying this code into setup.sql:
USE master;
GO

IF DB_ID(N'DataEngineeringLab') IS NULL
BEGIN
    CREATE DATABASE DataEngineeringLab;
END;
GO

USE DataEngineeringLab;
GO

IF SCHEMA_ID(N'stg') IS NULL
    EXECUTE(N'CREATE SCHEMA stg');
GO

IF SCHEMA_ID(N'audit') IS NULL
    EXECUTE(N'CREATE SCHEMA audit');
GO

What does this SQL create?

  • The USE master statement gives SQL Server the correct context for creating the database.
  • The DB_ID check prevents the script from recreating DataEngineeringLab.
  • The stg schema groups staging objects.
  • The audit schema groups pipeline history objects.
  • Save sql/setup.sql.
  • Select Run in the SQL editor toolbar.
  • Use the refresh icon beside your tested connection in Object Explorer.

You should see DataEngineeringLab with the stg schema plus the audit schema. Your accepted rows now have a durable destination ready for its tables.

Cannot create the database?

Your SQL Server login may lack CREATE DATABASE permission. You can use an existing writable development database.

  • Replace DataEngineeringLab in the USE DataEngineeringLab line with your writable development database.
  • Select the script from your updated USE line downward whenever you run setup.sql.
  • Use the same database name when you set the connection string later in this step.
  • Help me adapt this setup script for an existing writable SQL Server database.
  • Add the staging table by copying this code at the end of setup.sql:
IF OBJECT_ID(N'stg.Orders', N'U') IS NULL
BEGIN
    CREATE TABLE stg.Orders
    (
        run_id UNIQUEIDENTIFIER NOT NULL,
        order_id INT NOT NULL,
        customer_name NVARCHAR(100) NOT NULL,
        order_date DATE NOT NULL,
        quantity INT NOT NULL,
        unit_price DECIMAL(10, 2) NOT NULL,
        source_file NVARCHAR(260) NOT NULL
    );
END;
GO

What does the staging table hold?

  • The run_id column links every staged row to one pipeline run.
  • The order columns preserve the typed values produced by your validator.
  • The source_file column records which CSV supplied the row.
  • Save sql/setup.sql.
  • Select Run in the SQL editor toolbar.
  • Refresh the tables in Object Explorer.

You should see stg.Orders under the database tables. It is ready to receive validated batches.

Cannot see the staging table?

Confirm the SQL editor is using DataEngineeringLab or your selected development database. Refresh the database node after the query succeeds.

Help me find why stg.Orders was not created.

  • Add the trusted target table by copying this code at the end of setup.sql:
IF OBJECT_ID(N'dbo.Orders', N'U') IS NULL
BEGIN
    CREATE TABLE dbo.Orders
    (
        order_id INT NOT NULL CONSTRAINT PK_Orders PRIMARY KEY,
        customer_name NVARCHAR(100) NOT NULL,
        order_date DATE NOT NULL,
        quantity INT NOT NULL,
        unit_price DECIMAL(10, 2) NOT NULL,
        source_file NVARCHAR(260) NOT NULL,
        last_run_id UNIQUEIDENTIFIER NOT NULL,
        loaded_at DATETIME2(7) NOT NULL
            CONSTRAINT DF_Orders_LoadedAt DEFAULT SYSUTCDATETIME()
    );
END;
GO

What does the target table protect?

  • The PK_Orders primary key gives every order one target identity.
  • The last_run_id column records which pipeline run last loaded the order.
  • The loaded_at default records the UTC load time.
  • Save sql/setup.sql.
  • Select Run in the SQL editor toolbar.
  • Refresh the tables in Object Explorer.

You should now see dbo.Orders. This table becomes the trusted destination after a staged batch passes its checks.

Target table query failed?

Check that the query editor completed the earlier database section first. The dbo schema is available in every SQL Server database.

Help me diagnose why dbo.Orders was not created.

  • Add the audit table by copying this code at the end of setup.sql:
IF OBJECT_ID(N'audit.PipelineRun', N'U') IS NULL
BEGIN
    CREATE TABLE audit.PipelineRun
    (
        run_id UNIQUEIDENTIFIER NOT NULL CONSTRAINT PK_PipelineRun PRIMARY KEY,
        source_file NVARCHAR(260) NOT NULL,
        valid_rows INT NOT NULL,
        rejected_rows INT NOT NULL,
        loaded_rows INT NULL,
        status VARCHAR(20) NOT NULL,
        started_at DATETIME2(7) NOT NULL
            CONSTRAINT DF_PipelineRun_StartedAt DEFAULT SYSUTCDATETIME(),
        completed_at DATETIME2(7) NULL,
        error_message NVARCHAR(4000) NULL,
        CONSTRAINT CK_PipelineRun_Status
            CHECK (status IN ('STARTED', 'SUCCESS', 'FAILED'))
    );
END;
GO

What does the audit table record?

  • The row counts record how much data reached each stage of a run.
  • The status constraint accepts STARTED, SUCCESS, or FAILED.
  • The timestamps show when a run started plus when it finished.
  • The error_message column preserves failure details for investigation.
  • Save sql/setup.sql.
  • Select Run in the SQL editor toolbar.
  • Refresh the tables in Object Explorer.

You should see audit.PipelineRun beside the other tables. The database can now preserve the outcome of every pipeline run.

Audit table missing?

Confirm the audit schema exists before the table statement runs. Select the query result panel to check whether SQL Server reported a permission or syntax problem.

Help me diagnose why audit.PipelineRun was not created.

  • Add the initial row-count checks by copying this code at the end of setup.sql:
SELECT COUNT(*) AS target_row_count FROM dbo.Orders;
SELECT COUNT(*) AS staged_row_count FROM stg.Orders;
SELECT COUNT(*) AS audit_row_count FROM audit.PipelineRun;
GO

What do these checks prove?

Each COUNT(*) query measures one landing-zone table. Zero rows confirm that the structures exist before the first database load.

  • Save sql/setup.sql.

Before you run the completed script, do you expect any validated orders to be stored in these tables yet?

  • Select Run in the SQL editor toolbar.

You should see target_row_count, staged_row_count, plus audit_row_count return 0. The tables are ready while the four accepted rows remain outside SQL Server.

Counts are missing or nonzero?

A missing result usually means only part of setup.sql ran. Select the full file before choosing Run.

A nonzero result means the selected database already contains pipeline data. Use a fresh development database for the expected starting state.

Help me explain the unexpected setup row counts.

✔️ Awesome, I've got everything!

Great. Double-check that you saved sql/setup.sql.

ⓧ I'd like to double check the full code

USE master;
GO

IF DB_ID(N'DataEngineeringLab') IS NULL
BEGIN
    CREATE DATABASE DataEngineeringLab;
END;
GO

USE DataEngineeringLab;
GO

IF SCHEMA_ID(N'stg') IS NULL
    EXECUTE(N'CREATE SCHEMA stg');
GO

IF SCHEMA_ID(N'audit') IS NULL
    EXECUTE(N'CREATE SCHEMA audit');
GO

IF OBJECT_ID(N'stg.Orders', N'U') IS NULL
BEGIN
    CREATE TABLE stg.Orders
    (
        run_id UNIQUEIDENTIFIER NOT NULL,
        order_id INT NOT NULL,
        customer_name NVARCHAR(100) NOT NULL,
        order_date DATE NOT NULL,
        quantity INT NOT NULL,
        unit_price DECIMAL(10, 2) NOT NULL,
        source_file NVARCHAR(260) NOT NULL
    );
END;
GO

IF OBJECT_ID(N'dbo.Orders', N'U') IS NULL
BEGIN
    CREATE TABLE dbo.Orders
    (
        order_id INT NOT NULL CONSTRAINT PK_Orders PRIMARY KEY,
        customer_name NVARCHAR(100) NOT NULL,
        order_date DATE NOT NULL,
        quantity INT NOT NULL,
        unit_price DECIMAL(10, 2) NOT NULL,
        source_file NVARCHAR(260) NOT NULL,
        last_run_id UNIQUEIDENTIFIER NOT NULL,
        loaded_at DATETIME2(7) NOT NULL
            CONSTRAINT DF_Orders_LoadedAt DEFAULT SYSUTCDATETIME()
    );
END;
GO

IF OBJECT_ID(N'audit.PipelineRun', N'U') IS NULL
BEGIN
    CREATE TABLE audit.PipelineRun
    (
        run_id UNIQUEIDENTIFIER NOT NULL CONSTRAINT PK_PipelineRun PRIMARY KEY,
        source_file NVARCHAR(260) NOT NULL,
        valid_rows INT NOT NULL,
        rejected_rows INT NOT NULL,
        loaded_rows INT NULL,
        status VARCHAR(20) NOT NULL,
        started_at DATETIME2(7) NOT NULL
            CONSTRAINT DF_PipelineRun_StartedAt DEFAULT SYSUTCDATETIME(),
        completed_at DATETIME2(7) NULL,
        error_message NVARCHAR(4000) NULL,
        CONSTRAINT CK_PipelineRun_Status
            CHECK (status IN ('STARTED', 'SUCCESS', 'FAILED'))
    );
END;
GO

SELECT COUNT(*) AS target_row_count FROM dbo.Orders;
SELECT COUNT(*) AS staged_row_count FROM stg.Orders;
SELECT COUNT(*) AS audit_row_count FROM audit.PipelineRun;
GO
Set the connection string for this session

The next pipeline step reads its database location from an environment variable. Windows PowerShell passes process-scoped variables to the Python commands launched from the same terminal.

Connection strings can contain credentials when SQL Login is used. You will verify only that the variable exists, so its value stays off-screen.

  • Switch back to the PowerShell terminal where tutorial-env is still activated.
  • Replace YOUR_SERVER_NAME with the server name from the tested VS Code connection.
  • Set SQL_CONNECTION_STRING for the current session by running this command:
$Env:SQL_CONNECTION_STRING = "Server=YOUR_SERVER_NAME;Database=DataEngineeringLab;Trusted_Connection=yes;Encrypt=yes;TrustServerCertificate=yes;"

What does this setting do?

The SQL_CONNECTION_STRING variable gives the Python process the server plus database details. Trusted_Connection=yes uses Windows Authentication.

The encryption settings support this controlled local development connection. Keep shared or production connection details outside this project.

  • Verify that the variable exists without displaying its value by running:
Test-Path Env:SQL_CONNECTION_STRING

What does this check reveal?

The command checks whether SQL_CONNECTION_STRING exists in the current PowerShell session. It returns only a Boolean result, so the connection string remains hidden.

You should see True. The next Python process launched from this terminal can now read the connection setting.

Seeing False?

Run the assignment command again in the same PowerShell terminal. Confirm that the variable name is exactly SQL_CONNECTION_STRING.

If your tested connection uses SQL Login, set the variable to that existing local development connection string. Keep its password out of setup.sql plus every tracked file.

Help me set SQL_CONNECTION_STRING without exposing credentials.

Verify the empty landing zone

A successful setup needs two forms of evidence. Query results prove that every table is empty while Object Explorer confirms that every database object exists.

  • Return to sql/setup.sql from earlier.
  • Confirm that the SQL editor connection targets master.

Before you run the final check, do you expect the database to contain the four validated rows or three empty tables?

  • Execute the complete script by selecting Run.

You should see target_row_count, staged_row_count, plus audit_row_count return 0.

  • Use the refresh icon beside your tested SQL Server connection in Object Explorer.
  • Expand DataEngineeringLab or your selected writable development database.
  • Expand the database tables in Object Explorer.

You should see stg.Orders, dbo.Orders, plus audit.PipelineRun.

That is the durable foundation complete. Your pipeline now has separate places for staged batches, trusted orders, plus run history.

Next, you will connect the validator to these tables so one run can stage, promote, then audit the accepted orders.

Load, Promote, and Audit the Batch

Your validation-only script separates clean orders from malformed source rows. The four accepted orders still disappear when the Python process ends.

This step uses a staging table to land those rows in SQL Server. The mssql-python driver then promotes them inside a transaction that also records the audit result.

In this step, get ready to:
  • Connect the pipeline to the configured SQL Server database.
  • Verify a four-row bulk copy in the staging table.
  • Complete a transactional promotion with a successful audit record.
Add the connection and audit helpers

The pipeline needs its database connection from the current PowerShell environment. A generated run identifier links every staging row to one audit record.

  • In src/pipeline.py, replace the import and configuration section at the top with this code:
import csv
import os
from datetime import date
from decimal import Decimal, InvalidOperation
from pathlib import Path
from uuid import uuid4

import mssql_python

BASE_DIR = Path(__file__).resolve().parents[1]
INPUT_FILE = BASE_DIR / "data" / "orders.csv"
REJECT_FILE = BASE_DIR / "output" / "rejected_orders.csv"
EXPECTED_FIELDS = [
    "order_id",
    "customer_name",
    "order_date",
    "quantity",
    "unit_price",
]
STAGING_COLUMNS = [
    "run_id",
    "order_id",
    "customer_name",
    "order_date",
    "quantity",
    "unit_price",
    "source_file",
]

What does this configuration add?

  • The os import gives the script access to the connection string stored in the current process environment.
  • The uuid4 import creates a unique identifier for each pipeline run.
  • The STAGING_COLUMNS list maps each Python tuple value to the matching column in stg.Orders.
  • Save src/pipeline.py.
  • Confirm the existing validation flow still runs by executing this command:
python src/pipeline.py

What does this check prove?

This run proves Python can import the database driver without breaking the existing CSV validation flow.

You should still see Valid rows: 4 followed by Rejected rows: 2.

Does the driver import fail?

Confirm that tutorial-env remains active in this PowerShell session. Run the dependency installation from Step 1 again if the package is missing.

Ask for help with the active environment or package import: help me diagnose why src/pipeline.py cannot import mssql_python from tutorial-env.

  • In src/pipeline.py, add get_connection_string() below write_rejections() by copying this code:
def get_connection_string():
    connection_string = os.environ.get("SQL_CONNECTION_STRING")
    if not connection_string:
        raise RuntimeError(
            "Set SQL_CONNECTION_STRING in the current PowerShell session before running."
        )
    return connection_string

How is the connection protected?

The helper reads SQL_CONNECTION_STRING from the current PowerShell process. It stops the pipeline before database work begins when that value is missing.

  • Add mark_failed() directly below get_connection_string() by copying this code:
def mark_failed(connection, run_id, error):
    failure_cursor = connection.cursor()
    failure_cursor.execute(
        """
        UPDATE audit.PipelineRun
        SET status = 'FAILED',
            completed_at = SYSUTCDATETIME(),
            error_message = %(error_message)s
        WHERE run_id = %(run_id)s;
        """,
        {"run_id": run_id, "error_message": str(error)[:4000]},
    )
    connection.commit()
    failure_cursor.close()

What does failure auditing capture?

The helper changes the current audit record to FAILED. It stores the error text with a completion timestamp so a failed batch remains traceable.

  • Save src/pipeline.py.
  • Check both helper definitions for valid Python syntax by running the script again:
python src/pipeline.py

What does the second check prove?

The unchanged validation counts confirm that both new helper definitions load successfully. The next edit calls them during the database run.

You should again see four valid rows with two rejected rows.

Seeing a Python syntax error?

Check that both functions start at the left edge of src/pipeline.py. Confirm that the SQL text inside mark_failed() begins and ends with three double quotes.

Ask for help comparing the helper definitions: help me find the syntax difference in my connection and failure-audit helpers.

Stage and verify the accepted rows

Bulk copy places the four typed order tuples into stg.Orders through a separate internal connection. The pipeline checks the stored count before any target rows can change.

Why verify staging first?

The bulk-copy operation commits independently from the main connection. A count check prevents an incomplete staging load from reaching the trusted target.

  • In src/pipeline.py, replace the existing run_pipeline() function with the next four code blocks in order:
def run_pipeline():
    valid_rows, rejected_rows = validate_orders(INPUT_FILE)
    write_rejections(REJECT_FILE, rejected_rows)

    print(f"Valid rows: {len(valid_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")

    connection_string = get_connection_string()
    run_id = uuid4()
    connection = None
    cursor = None

What does the run setup hold?

  • The existing validation functions still produce the accepted and rejected collections first.
  • The run_id value identifies every database change made by this execution.
  • The connection variables start as None so the cleanup logic can safely check them after an error.
    try:
        connection = mssql_python.connect(connection_string)
        cursor = connection.cursor()

        cursor.execute(
            """
            INSERT INTO audit.PipelineRun
                (run_id, source_file, valid_rows, rejected_rows, status)
            VALUES
                (%(run_id)s, %(source_file)s, %(valid_rows)s,
                 %(rejected_rows)s, 'STARTED');
            """,
            {
                "run_id": run_id,
                "source_file": INPUT_FILE.name,
                "valid_rows": len(valid_rows),
                "rejected_rows": len(rejected_rows),
            },
        )
        connection.commit()

How does the audit begin?

The main connection inserts a STARTED record before loading staging. That record captures the source filename with both validation counts.

The immediate commit preserves the audit record so later failure handling can update the same run.

        staging_rows = [
            (run_id, *row, INPUT_FILE.name) for row in valid_rows
        ]
        result = {"rows_copied": 0}

        if staging_rows:
            result = cursor.bulkcopy(
                "stg.Orders",
                staging_rows,
                batch_size=5000,
                column_mappings=STAGING_COLUMNS,
                use_internal_transaction=True,
            )

How does bulk copy map the rows?

  • Each tuple starts with run_id so the staging rows belong to one execution.
  • Each tuple ends with INPUT_FILE.name so the source remains visible after loading.
  • The column_mappings argument applies the explicit STAGING_COLUMNS order.
  • The internal transaction commits the bulk-copy batches independently.
        cursor.execute(
            "SELECT COUNT(*) FROM stg.Orders WHERE run_id = %(run_id)s;",
            {"run_id": run_id},
        )
        staged_rows = cursor.fetchval()

        if staged_rows != len(valid_rows):
            raise RuntimeError(
                f"Staging validation failed: expected {len(valid_rows)}, got {staged_rows}"
            )

What does the staging guard protect?

The query counts only rows associated with the current run_id. A mismatch raises an error before target promotion begins.

Promote the batch and confirm the audit

The verified staging batch is ready for publication. The remaining statements replace matching target keys before inserting the current rows as one committed unit.

  • Continue run_pipeline() directly below the staging-count check by copying the next code blocks in order:
        cursor.execute(
            """
            DELETE target
            FROM dbo.Orders AS target
            INNER JOIN stg.Orders AS source
                ON source.order_id = target.order_id
            WHERE source.run_id = %(run_id)s;
            """,
            {"run_id": run_id},
        )

Why delete matching keys?

The delete touches only target orders found in the current staging batch. Replacing those keys makes the same source feed safe to process again.

        cursor.execute(
            """
            INSERT INTO dbo.Orders
                (order_id, customer_name, order_date, quantity, unit_price,
                 source_file, last_run_id)
            SELECT order_id, customer_name, order_date, quantity, unit_price,
                   source_file, run_id
            FROM stg.Orders
            WHERE run_id = %(run_id)s;
            """,
            {"run_id": run_id},
        )

How are trusted rows published?

The insert selects only the verified rows for the current run. SQL Server supplies the target table's loaded_at timestamp through its default constraint.

        cursor.execute(
            """
            UPDATE audit.PipelineRun
            SET loaded_rows = %(loaded_rows)s,
                status = 'SUCCESS',
                completed_at = SYSUTCDATETIME(),
                error_message = NULL
            WHERE run_id = %(run_id)s;
            """,
            {"run_id": run_id, "loaded_rows": staged_rows},
        )

        cursor.execute(
            "DELETE FROM stg.Orders WHERE run_id = %(run_id)s;",
            {"run_id": run_id},
        )
        connection.commit()

What completes the transaction?

The audit record changes to SUCCESS with the loaded count. The current staging rows are removed before one commit makes the promotion permanent.

        cursor.execute("SELECT COUNT(*) FROM dbo.Orders;")
        target_rows = cursor.fetchval()

        print(f"Staged rows: {result['rows_copied']}")
        print(f"Loaded to target: {staged_rows}")
        print(f"Target rows after load: {target_rows}")
        print(f"Run {run_id} succeeded")

What does the terminal report?

The final query reads the target count after the commit. The printed values make the accepted, staged, loaded, and final target totals visible in one run.

    except Exception as error:
        if connection is not None:
            connection.rollback()
            try:
                mark_failed(connection, run_id, error)
            except mssql_python.Error:
                connection.rollback()
        raise
    finally:
        if cursor is not None:
            cursor.close()
        if connection is not None:
            connection.close()

How does failure handling work?

An exception rolls back pending work on the main connection. The pipeline then attempts to mark the audit run as FAILED before raising the original error again.

The finally block closes any cursor or connection that was created.

✔️ Awesome, I've got everything!

  • Save src/pipeline.py before running the completed pipeline.

ⓧ I'd like to double check the full code

import csv
import os
from datetime import date
from decimal import Decimal, InvalidOperation
from pathlib import Path
from uuid import uuid4

import mssql_python

BASE_DIR = Path(__file__).resolve().parents[1]
INPUT_FILE = BASE_DIR / "data" / "orders.csv"
REJECT_FILE = BASE_DIR / "output" / "rejected_orders.csv"
EXPECTED_FIELDS = [
    "order_id",
    "customer_name",
    "order_date",
    "quantity",
    "unit_price",
]
STAGING_COLUMNS = [
    "run_id",
    "order_id",
    "customer_name",
    "order_date",
    "quantity",
    "unit_price",
    "source_file",
]


def validate_orders(input_file: Path):
    valid_rows = []
    rejected_rows = []

    with input_file.open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")

        for row_number, row in enumerate(reader, start=2):
            try:
                order_id = int(row["order_id"])
                customer_name = row["customer_name"].strip()
                order_date = date.fromisoformat(row["order_date"].strip())
                quantity = int(row["quantity"])
                unit_price = Decimal(row["unit_price"]).quantize(Decimal("0.01"))

                if order_id <= 0:
                    raise ValueError("order_id must be positive")
                if not customer_name:
                    raise ValueError("customer_name is required")
                if len(customer_name) > 100:
                    raise ValueError("customer_name exceeds 100 characters")
                if quantity <= 0:
                    raise ValueError("quantity must be positive")
                if unit_price < 0:
                    raise ValueError("unit_price cannot be negative")

                valid_rows.append(
                    (order_id, customer_name, order_date, quantity, unit_price)
                )
            except (ValueError, TypeError, InvalidOperation) as error:
                rejected = {
                    field: row.get(field, "") for field in EXPECTED_FIELDS
                }
                rejected["reject_reason"] = f"row {row_number}: {error}"
                rejected_rows.append(rejected)

    return valid_rows, rejected_rows


def write_rejections(reject_file: Path, rejected_rows):
    reject_file.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = EXPECTED_FIELDS + ["reject_reason"]

    with reject_file.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rejected_rows)


def get_connection_string():
    connection_string = os.environ.get("SQL_CONNECTION_STRING")
    if not connection_string:
        raise RuntimeError(
            "Set SQL_CONNECTION_STRING in the current PowerShell session before running."
        )
    return connection_string


def mark_failed(connection, run_id, error):
    failure_cursor = connection.cursor()
    failure_cursor.execute(
        """
        UPDATE audit.PipelineRun
        SET status = 'FAILED',
            completed_at = SYSUTCDATETIME(),
            error_message = %(error_message)s
        WHERE run_id = %(run_id)s;
        """,
        {"run_id": run_id, "error_message": str(error)[:4000]},
    )
    connection.commit()
    failure_cursor.close()


def run_pipeline():
    valid_rows, rejected_rows = validate_orders(INPUT_FILE)
    write_rejections(REJECT_FILE, rejected_rows)

    print(f"Valid rows: {len(valid_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")

    connection_string = get_connection_string()
    run_id = uuid4()
    connection = None
    cursor = None

    try:
        connection = mssql_python.connect(connection_string)
        cursor = connection.cursor()

        cursor.execute(
            """
            INSERT INTO audit.PipelineRun
                (run_id, source_file, valid_rows, rejected_rows, status)
            VALUES
                (%(run_id)s, %(source_file)s, %(valid_rows)s,
                 %(rejected_rows)s, 'STARTED');
            """,
            {
                "run_id": run_id,
                "source_file": INPUT_FILE.name,
                "valid_rows": len(valid_rows),
                "rejected_rows": len(rejected_rows),
            },
        )
        connection.commit()

        staging_rows = [
            (run_id, *row, INPUT_FILE.name) for row in valid_rows
        ]
        result = {"rows_copied": 0}

        if staging_rows:
            result = cursor.bulkcopy(
                "stg.Orders",
                staging_rows,
                batch_size=5000,
                column_mappings=STAGING_COLUMNS,
                use_internal_transaction=True,
            )

        cursor.execute(
            "SELECT COUNT(*) FROM stg.Orders WHERE run_id = %(run_id)s;",
            {"run_id": run_id},
        )
        staged_rows = cursor.fetchval()

        if staged_rows != len(valid_rows):
            raise RuntimeError(
                f"Staging validation failed: expected {len(valid_rows)}, got {staged_rows}"
            )

        cursor.execute(
            """
            DELETE target
            FROM dbo.Orders AS target
            INNER JOIN stg.Orders AS source
                ON source.order_id = target.order_id
            WHERE source.run_id = %(run_id)s;
            """,
            {"run_id": run_id},
        )

        cursor.execute(
            """
            INSERT INTO dbo.Orders
                (order_id, customer_name, order_date, quantity, unit_price,
                 source_file, last_run_id)
            SELECT order_id, customer_name, order_date, quantity, unit_price,
                   source_file, run_id
            FROM stg.Orders
            WHERE run_id = %(run_id)s;
            """,
            {"run_id": run_id},
        )

        cursor.execute(
            """
            UPDATE audit.PipelineRun
            SET loaded_rows = %(loaded_rows)s,
                status = 'SUCCESS',
                completed_at = SYSUTCDATETIME(),
                error_message = NULL
            WHERE run_id = %(run_id)s;
            """,
            {"run_id": run_id, "loaded_rows": staged_rows},
        )

        cursor.execute(
            "DELETE FROM stg.Orders WHERE run_id = %(run_id)s;",
            {"run_id": run_id},
        )
        connection.commit()

        cursor.execute("SELECT COUNT(*) FROM dbo.Orders;")
        target_rows = cursor.fetchval()

        print(f"Staged rows: {result['rows_copied']}")
        print(f"Loaded to target: {staged_rows}")
        print(f"Target rows after load: {target_rows}")
        print(f"Run {run_id} succeeded")
    except Exception as error:
        if connection is not None:
            connection.rollback()
            try:
                mark_failed(connection, run_id, error)
            except mssql_python.Error:
                connection.rollback()
        raise
    finally:
        if cursor is not None:
            cursor.close()
        if connection is not None:
            connection.close()


if __name__ == "__main__":
    run_pipeline()

How to use this reference

Compare this file with your saved src/pipeline.py. Check the indentation around try, except, and finally before running it.

Before you run the completed pipeline, do you expect all six source records to reach SQL Server?

  • Run the completed pipeline from the activated PowerShell session with this command:
python src/pipeline.py

What does this run test?

This command validates the source feed before opening the database connection. It stages only accepted rows before completing the target promotion and audit update.

You should see Valid rows: 4, Rejected rows: 2, Staged rows: 4, Loaded to target: 4, and Target rows after load: 4 on separate lines. The final line contains the generated run identifier followed by succeeded.

Did the database load fail?

Confirm that SQL_CONNECTION_STRING is still set in the same PowerShell session. Confirm that it points to the writable database containing stg.Orders, dbo.Orders, and audit.PipelineRun.

Ask for help with the failed stage or promotion: help me diagnose my SQL Server pipeline error without exposing my connection string or password.

  • Return to the SQL Server editor from earlier.
  • Select DataEngineeringLab in Object Explorer.
  • Select New Query.
  • Paste these count queries into the editor:
SELECT COUNT(*) AS target_row_count FROM dbo.Orders;
SELECT COUNT(*) AS staged_row_count FROM stg.Orders;
SELECT COUNT(*) AS audit_row_count FROM audit.PipelineRun;

What do these counts measure?

  • The target count shows how many trusted orders were published.
  • The staging count confirms whether the completed run cleaned up its temporary rows.
  • The audit count shows how many pipeline attempts have been recorded.

Before you execute the query, what should each count reveal about the completed transaction?

  • Select Run in the query editor.

You should see target_row_count equal to 4. You should see staged_row_count equal to 0 and audit_row_count equal to 1.

That is the risky part complete. Your accepted orders now survive the Python process with a clean staging area and a traceable audit result.

Your pipeline now turns validated rows into trusted SQL Server data. Next, you will rerun the same feed to prove that the target stays duplicate-free before publishing the project.

Prove Repeatability and Publish the Project

Your Python pipeline now validates a messy CSV feed before loading trusted rows into SQL Server.

A one-time load can hide duplicate-key bugs. A second run reveals whether the target stays reliable.

You will rerun the same feed. A verification query will prove that the target remains stable while the audit history grows.

You will document the working pipeline. You will then publish the project with GitHub.

In this step, get ready to:
  • Rerun the default feed to prove that the target does not gain duplicate orders.
  • Verify the target rows, empty staging table, and two successful audit records.
  • Document the project before committing it locally and publishing it to GitHub.
Prove the second run is idempotent

Idempotency means the same input can run repeatedly without multiplying the target data. This pipeline achieves that by replacing only the target keys found in the current staging batch.

Before you run the pipeline again, consider which count should change: the target count or the audit count.

  • Switch back to the PowerShell terminal from earlier.
  • Run the default pipeline for a second time with this command:
python src/pipeline.py

What does this rerun prove?

The rerun creates a fresh audit identifier. It processes the same six source records again.

The terminal reports four valid rows. It also reports two rejected rows.

Four rows pass through staging. Four rows are promoted to the target.

The successful run identifier confirms that the second batch completed. The database query comes next to prove that the target still contains only four orders.

Did the second run fail?

Confirm that the terminal prompt still shows the activated tutorial-env environment.

Confirm that SQL_CONNECTION_STRING remains set in this PowerShell session.

Ask for help with the failed rerun.

  • Create sql/verify.sql in the sql folder through the Visual Studio Code Explorer sidebar by pasting this query:
USE DataEngineeringLab;
GO

SELECT order_id,
       customer_name,
       order_date,
       quantity,
       unit_price,
       source_file,
       last_run_id,
       loaded_at
FROM dbo.Orders
ORDER BY order_id;

SELECT run_id,
       source_file,
       valid_rows,
       rejected_rows,
       loaded_rows,
       status,
       started_at,
       completed_at,
       error_message
FROM audit.PipelineRun
ORDER BY started_at DESC;

SELECT COUNT(*) AS target_row_count FROM dbo.Orders;
SELECT COUNT(*) AS staged_row_count FROM stg.Orders;
SELECT COUNT(*) AS audit_row_count FROM audit.PipelineRun;
GO

What does this query verify?

  • The first query lists the trusted orders in key order.
  • The second query lists the newest audit record first.
  • The final queries count the target rows, staging rows, and audit rows.
  • Save sql/verify.sql.

Before you execute the query, consider whether staging should still contain rows after a successful promotion.

  • Return to the database connection from earlier in the SQL Server (mssql) extension.
  • Select Run in the query toolbar.

You will see orders 1001, 1002, 1005, and 1006. The two newest audit records show successful runs for orders.csv.

The final results show a target count of 4. The staging count is 0. The audit count is 2.

Why do these counts matter?

A target count of 4 proves that rerunning the feed did not create duplicates. An audit count of 2 proves that each attempt remains traceable.

An empty staging table confirms that both successful batches completed their cleanup.

Do the counts look different?

Confirm that sql/verify.sql is connected to DataEngineeringLab.

Rerun the pipeline once if the audit count is still 1.

Ask for help interpreting the verification results.

✔️ Awesome, I've got everything!

Your verification query is saved. The results prove that the second run preserved four target orders and added one audit record.

ⓧ I'd like to double check the full code

USE DataEngineeringLab;
GO

SELECT order_id,
       customer_name,
       order_date,
       quantity,
       unit_price,
       source_file,
       last_run_id,
       loaded_at
FROM dbo.Orders
ORDER BY order_id;

SELECT run_id,
       source_file,
       valid_rows,
       rejected_rows,
       loaded_rows,
       status,
       started_at,
       completed_at,
       error_message
FROM audit.PipelineRun
ORDER BY started_at DESC;

SELECT COUNT(*) AS target_row_count FROM dbo.Orders;
SELECT COUNT(*) AS staged_row_count FROM stg.Orders;
SELECT COUNT(*) AS audit_row_count FROM audit.PipelineRun;
GO

How should I compare this file?

Compare the query order and selected fields with your saved file. Confirm that all three count queries appear before the final GO.

Document and commit the pipeline

A README turns the working files into a project another person can understand. It records the pipeline architecture before listing the Windows run process.

  • Create README.md inside the sql-server-batch-pipeline folder through the Explorer sidebar by pasting this introduction:
# SQL Server Batch Pipeline

A local Python data engineering project that validates a CSV feed, writes rejected records, bulk-loads accepted rows into SQL Server staging, transactionally promotes them to a target table, and records an audit trail.

## Architecture

Raw CSV -> Python validation -> rejected CSV or SQL Server staging -> transactional target load -> audit history

What does this introduction show?

The opening sentence states the pipeline's purpose. The architecture line traces each record from the raw feed to its accepted or rejected destination.

  • Save README.md.

The editor now shows the project title above a one-line architecture path.

Does the README formatting look wrong?

Confirm that the title begins with one hash character. Confirm that the Architecture heading begins with two hash characters.

Ask for help checking the Markdown structure.

  • Append the following Windows instructions after one blank line below the architecture path:
## Run on Windows

1. Create and activate the environment:

   ```powershell
   python -m venv tutorial-env
   tutorial-env\Scripts\activate
   python -m pip install -r requirements.txt
   ```

2. Install the SQL Server (mssql) extension in Visual Studio Code and run `sql/setup.sql` while connected to `master`.

3. Set the connection string for the current PowerShell session. Replace the server placeholder:

   ```powershell
   $Env:SQL_CONNECTION_STRING = "Server=YOUR_SERVER_NAME;Database=DataEngineeringLab;Trusted_Connection=yes;Encrypt=yes;TrustServerCertificate=yes;"
   ```

4. Run the pipeline:

   ```powershell
   python src/pipeline.py
   ```

5. Run `sql/verify.sql` in Visual Studio Code.

`TrustServerCertificate=yes` is used only for this controlled local development project. Do not commit credentials or reuse a local-development connection string for shared or production systems.

What do these instructions capture?

The runbook records the environment creation process. It also covers dependency installation and database setup.

The final warning keeps the local certificate setting within its intended development context. It also keeps credentials out of version control.

  • Save README.md.

The file now documents the full Windows run process from environment creation through SQL verification.

Is part of the runbook missing?

Confirm that each numbered step appears once. Confirm that every command fence has a matching closing fence.

Ask for help comparing the README sections.

✔️ Awesome, I've got everything!

Your README explains what the pipeline does. It also gives another Windows user the commands needed to run it.

ⓧ I'd like to double check the full code

# SQL Server Batch Pipeline

A local Python data engineering project that validates a CSV feed, writes rejected records, bulk-loads accepted rows into SQL Server staging, transactionally promotes them to a target table, and records an audit trail.

## Architecture

Raw CSV -> Python validation -> rejected CSV or SQL Server staging -> transactional target load -> audit history

## Run on Windows

1. Create and activate the environment:

   ```powershell
   python -m venv tutorial-env
   tutorial-env\Scripts\activate
   python -m pip install -r requirements.txt
   ```

2. Install the SQL Server (mssql) extension in Visual Studio Code and run `sql/setup.sql` while connected to `master`.

3. Set the connection string for the current PowerShell session. Replace the server placeholder:

   ```powershell
   $Env:SQL_CONNECTION_STRING = "Server=YOUR_SERVER_NAME;Database=DataEngineeringLab;Trusted_Connection=yes;Encrypt=yes;TrustServerCertificate=yes;"
   ```

4. Run the pipeline:

   ```powershell
   python src/pipeline.py
   ```

5. Run `sql/verify.sql` in Visual Studio Code.

`TrustServerCertificate=yes` is used only for this controlled local development project. Do not commit credentials or reuse a local-development connection string for shared or production systems.

How should I compare this file?

Check each heading and numbered instruction against your saved README. Pay close attention to the connection-string warning at the bottom.

  • Switch to .gitignore in the Explorer sidebar.
  • Confirm that it lists tutorial-env/.
  • Confirm that it lists __pycache__/.
  • Confirm that it lists *.pyc.
  • Confirm that it lists output/.

Git records the project files that explain or reproduce the pipeline. The ignore rules keep the virtual environment and generated rejection output outside the commit.

  • Return to the PowerShell terminal from earlier.
  • Initialize the repository on the main branch by running this command:
git init -b main

What does this command do?

The command creates the local Git repository. It sets main as the first branch.

The terminal confirms that a Git repository was initialized inside the project folder.

Did repository initialization fail?

Confirm that the terminal is still inside the sql-server-batch-pipeline folder.

Ask for help with the initialization error.

  • Stage the project files by running this command:
git add .

What does this command do?

The command stages files from the project folder. Git applies the existing ignore rules while it builds that staged snapshot.

  • Select the source control icon in the Visual Studio Code Activity Bar.
  • Inspect the staged file list.

The tracked list includes the project code and documentation. It excludes tutorial-env and output.

  • Create the first local commit by running this command:
git commit -m "First commit"

What does this command do?

The command stores the staged snapshot in local history. The commit message identifies it as the project's first saved version.

Your local project now has a committed baseline. The terminal reports the new commit with its changed-file summary.

Could Git not create the commit?

Git may need your author identity before it can create a first commit. Follow the terminal guidance for your local Git installation.

Ask for help completing the first commit.

Publish the repository to GitHub

The local commit preserves your work on this computer. A remote repository makes the pipeline reviewable from another machine.

  • Use GitHub's repository menu to start a new repository.
  • Enter sql-server-batch-pipeline as the repository name.
  • Leave the README initialization option unselected.
  • Leave the license initialization option unselected.
  • Leave the gitignore initialization option unselected.
  • Create the empty repository.

GitHub now shows instructions for pushing existing local code. The empty repository avoids a competing initial commit.

  • Copy the remote URL from the repository page.
  • Record the copied value here: your GitHub repository URL.
  • Connect the local repository to GitHub by replacing REMOTE-URL with your GitHub repository URL in this command:
git remote add origin REMOTE-URL

What does this command do?

The command saves the GitHub repository as the local remote named origin.

  • Verify the saved remote by running this command:
git remote -v

What does this command show?

The command lists the saved remote destinations. The output shows origin beside the GitHub repository URL.

Authentication can feel like a detour. Git only needs to confirm your GitHub account before the first push.

Before you push, consider which tracked project files should appear on GitHub after the command succeeds.

  • Push the local main branch to GitHub by running this command:
git push -u origin main

What does this command do?

The command uploads the local commit to origin/main. It also links the local branch to that remote branch.

The terminal confirms that the push completed. Your local main branch now tracks origin/main.

Did the push fail?

Complete any GitHub authentication request before retrying the push. Confirm that git remote -v shows the intended repository URL.

Ask for help with the failed push.

  • Refresh the GitHub repository page.
  • Inspect the published file list.

You will see README.md and requirements.txt alongside the source, data, and SQL folders. You will also see .gitignore.

The tutorial-env folder is absent. The generated output folder is also absent.

That completes the evidence trail: two audited runs preserve four trusted orders, and the reproducible project is now published.

Secret mission

Make the Pipeline Accept Any Input File

Your pipeline can process its default feed reliably. This challenge adds a command-line input for later feeds. A correction batch updates an existing order. It also inserts a new order while isolating one malformed row.

Clean Up Your Resources

Clean Up Your Resources

This project adds no cloud resources or incremental usage charges. Choose how much of your local Microsoft SQL Server workspace you want to keep for later.

Resources you used:

  • The local DataEngineeringLab database with five target orders. It also contains three successful audit records.
  • The local sql-server-batch-pipeline folder with your Python code. This folder also contains tutorial-env plus the generated rejection files.
  • The GitHub repository connected to origin.

Keep everything running

No action needed. Choose this if you plan to continue building the SQL Server data engineering roadmap.

  • Leave the DataEngineeringLab database in place for future pipeline work.
  • Keep the sql-server-batch-pipeline folder with its virtual environment and generated outputs.
  • Retain the GitHub repository as the remote copy of your project.

Pause - I'll come back to this later

This option ends your active sessions while preserving every saved resource.

  • Close the current Windows PowerShell session.
  • Disconnect the database connection in Visual Studio Code.
  • Keep the DataEngineeringLab database for your next session.

Closing PowerShell deactivates tutorial-env and clears the process-scoped SQL_CONNECTION_STRING. No pipeline process continues in the background.

Delete - I don't want to use this again

Removing the database and local folder is permanent. Your GitHub repository remains a recovery copy until you delete it too.

  • Disconnect every query tab connected to DataEngineeringLab.
  • Switch the current connection in the SQL Server (mssql) extension to master.
  • Permanently remove the database by running this query:
DROP DATABASE DataEngineeringLab;

What Does This Delete?

This removes DataEngineeringLab with its staging table, target table, and audit history. The project files and GitHub repository remain untouched.

  • Refresh Object Explorer.

You should no longer see DataEngineeringLab in the database list.

  • Close the current Windows PowerShell session.

This clears the process-scoped SQL_CONNECTION_STRING and deactivates tutorial-env.

  • Use File Explorer to delete the sql-server-batch-pipeline folder from the location where you created it.

This removes the local repository, virtual environment, source files, and generated rejection output.

  • Delete the GitHub repository connected to origin through GitHub's repository settings if you do not want a remote copy.

Nice Work!

Nice Work!

You did it! Your Python batch pipeline now turns messy order feeds into trusted Microsoft SQL Server data.

You can validate source records, preserve rejected rows, promote accepted data safely, and prove every run through an audit trail.

You learned how to:

  • Build a typed CSV validation flow that parses dates, quantities, and decimal prices. Malformed records land in a rejection file with clear reasons.
  • Create a staging-to-target load that bulk-copies accepted rows before promoting them inside a transaction. Each run leaves a timestamped audit record.
  • Prove idempotent reruns by processing the same feed twice without duplicating target rows. You also published the complete project to GitHub.
  • Secret Mission: Made the pipeline accept any input file through --input. The correction batch updated order 1002, inserted order 1007, and rejected malformed order 1008.

Ready to quiz yourself?