Pipelines over real data

At Level 4 Integrator, company or lab systems and real data are in the loop. A pipeline that finishes is not necessarily a pipeline that was safe to use. The source identity, schema, quality thresholds, selected fields, output publication, and overlap policy must all be explicit and testable before anyone relies on its result.

2. The source changed and the pipeline carried on

At 07:00 the scheduled pipeline reads the live source. During the night, an instrument upgrade renamed value_c to temperature. In the Company version, a CRM migration changed account_id to organisation_id and reduced the export to one region. The job still produces a polished report because its data library silently supplies blanks or because a later model confidently interprets the wrong fields.

The dashboard is green. The result is plausible. It is also unsupported.

Connecting the reproducible project from T10-L03 to a real source changes the blast radius. The source can change without a code commit. Credentials can alter records. Sensitive columns can travel merely because they were included in a broad query. A delayed run can overlap the next schedule and publish out of order. The safe response is not a more persuasive prompt. It is a sequence of gates that stops before publication.

This book builds that sequence over a synthetic SQLite database that behaves like a real database connection: it is opened read-only, checked against an exact schema, queried through an explicit field list, tested against fixed quality thresholds, and converted into one atomic output. A deliberate corruption must halt the run and leave the last accepted output unchanged.

3. After this you can

Connect a pipeline through a read-only source identity and verify that it cannot write.
Reject an unexpected table shape before reading records into the analysis path.
Enforce row-count, uniqueness, null-rate, timestamp, and measurement-range gates.
Prove that personal and sensitive source fields never enter a model payload or generated output.
Choose and test an overlap policy for scheduled runs rather than letting two publications race.

4. Prerequisites

T10-L03 - A reproducible analysis pipeline, including immutable raw evidence, declared parameters, generated output, and deliberate-failure tests.
T12-L04 - Put a lock on it, including least privilege, secret handling, external audit, logging, and restore evidence.
Python 3.10 or newer with the standard library, a text editor, and permission to create a disposable local SQLite database.
For a later real-source adaptation: a named data owner, a source administrator who can provision read-only access, an approved secret-injection route, and a non-production connection for the first run.

Use only public, synthetic, course-provided, or explicitly approved data. Every row and identity below is synthetic. Do not copy a production database, CRM export, instrument store, participant table, customer record, free-text note, token, or connection string into the exercise. A read-only credential limits modification; it does not make every readable field appropriate for analysis or model use.

5. The idea in one page

Make acceptance a sequence

A live-data pipeline should move through gates in a fixed order:

named schedule
  -> acquire one-run lock
  -> connect read-only
  -> verify exact schema
  -> select only approved fields
  -> apply quality gates
  -> build model-safe payload, if needed
  -> atomically replace accepted output
  -> record bounded evidence and release lock

Each arrow has a stop condition. A later success cannot cancel an earlier failure. If the schema is wrong, do not calculate a null rate. If quality is outside the declared band, do not ask a model whether the data "looks reasonable." If the safe-field test fails, do not send a redacted version after the original request has already left.

Read-only has three parts

Database permission denies INSERT, UPDATE, DELETE, and schema changes. Application intent opens a read-only connection and issues only an explicit SELECT. Runtime placement keeps the credential outside code, configuration, logs, and Git. Test all three. The SQLite fixture uses a URI with mode=ro and PRAGMA query_only=ON; a production database needs an independently provisioned role with only the required database, schema, table, and column privileges.

Do not give the pipeline an owner's credential and promise to use it carefully. The database must refuse a write even if code, a dependency, or an operator attempts one.

Shape comes before values

A schema contract states the expected table, columns, types, nullability, and keys. An extra column can matter as much as a missing one: a new patient_name or customer_email field may be swept into SELECT *, logs, model payloads, or outputs. This exercise therefore requires an exact known schema and still selects only the four safe fields needed for the task.

Quality gates answer different questions:

Gate	Failure it detects	Decision fixed before the run
Row count	Empty, partial, duplicated, or unexpectedly broad extraction	Minimum and maximum accepted rows
Unique ID	Duplicate records or join expansion	No repeated primary record ID
Null rate	Missing measurements or mapping failure	Maximum proportion, including whether null is allowed
Range	Unit error, sentinel value, impossible measurement	Inclusive domain-approved minimum and maximum
Timestamp	Malformed or timezone-free event time	Parseable ISO 8601 with an offset

Thresholds are not universal truths. A lab's temperature range needs a domain owner; a CRM's expected activity count needs a process owner. Record who approved each value and when. A threshold set after seeing a failure is an explanation, not an independent gate.

Select before the model boundary

Minimise at the query, not after prompting. The Lab source deliberately contains operator_email and notes; neither is selected. The model-safe payload contains only measurement_id, batch_code, measured_at, and value_c. The test searches the complete output for canary values as well as checking every key. This does not certify anonymisation or approve a real model endpoint. It proves one narrow field boundary for the fixture.

One schedule needs one overlap rule

Choose skip, queue, or replace, then test it. This book uses skip: if the previous run still owns the lock, the next invocation exits non-zero without reading or publishing. Queue can be appropriate when every period must run in order. Replace is dangerous when a newer process can interrupt an older process during publication. Whatever the choice, identify the missed-run alert and the operator who decides whether to rerun.

6. The worked example: a database pipeline that refuses bad input

The executable Lab fixture uses a measurement database. The Company framing applies the same controls to CRM activity. Build and test the Lab fixture first; do not point this code at a live source during the exercise.

Create the bounded project

Create this structure in a disposable folder:

real-data-pipeline/
|-- config.json
|-- source/
|   `-- measurements.db       # generated synthetic fixture
|-- state/
|-- output/
`-- steps/
    |-- make_fixture.py
    |-- pipeline.py
    `-- test_pipeline.py

Save this as config.json:

{
  "database": "source/measurements.db",
  "table": "measurements",
  "expected_schema": [
    ["measurement_id", "TEXT", 1, 1],
    ["batch_code", "TEXT", 1, 0],
    ["measured_at", "TEXT", 1, 0],
    ["value_c", "REAL", 0, 0],
    ["unit", "TEXT", 1, 0],
    ["operator_email", "TEXT", 1, 0],
    ["notes", "TEXT", 1, 0]
  ],
  "model_fields": ["measurement_id", "batch_code", "measured_at", "value_c"],
  "expected_unit": "C",
  "min_rows": 4,
  "max_rows": 6,
  "max_null_rate": 0.0,
  "min_value": 0.0,
  "max_value": 50.0,
  "output": "output/accepted-run.json",
  "lock": "state/pipeline.lock"
}

The expected schema includes the sensitive columns because their unexpected rename or addition must be visible. Inclusion in the schema contract is not permission to retrieve their values. The query below never selects them.

Save this fixture generator as steps/make_fixture.py:

import sqlite3
from contextlib import closing
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DATABASE = ROOT / "source" / "measurements.db"

DATABASE.parent.mkdir(parents=True, exist_ok=True)
DATABASE.unlink(missing_ok=True)
with closing(sqlite3.connect(DATABASE)) as connection:
    connection.execute("""
        CREATE TABLE measurements (
            measurement_id TEXT PRIMARY KEY NOT NULL,
            batch_code TEXT NOT NULL,
            measured_at TEXT NOT NULL,
            value_c REAL,
            unit TEXT NOT NULL,
            operator_email TEXT NOT NULL,
            notes TEXT NOT NULL
        )
    """)
    connection.executemany(
        "INSERT INTO measurements VALUES (?, ?, ?, ?, ?, ?, ?)",
        [
            ("M-101", "CEDAR-A", "2026-09-01T08:00:00+00:00", 18.2, "C", "canary-one@example.invalid", "PRIVATE-CANARY-ONE"),
            ("M-102", "CEDAR-A", "2026-09-01T08:05:00+00:00", 19.1, "C", "canary-two@example.invalid", "PRIVATE-CANARY-TWO"),
            ("M-103", "CEDAR-B", "2026-09-01T08:10:00+00:00", 20.0, "C", "canary-three@example.invalid", "PRIVATE-CANARY-THREE"),
            ("M-104", "CEDAR-B", "2026-09-01T08:15:00+00:00", 21.4, "C", "canary-four@example.invalid", "PRIVATE-CANARY-FOUR")
        ]
    )
    connection.commit()
print(f"FIXTURE_READY rows=4 database={DATABASE}")

Save the gated pipeline as steps/pipeline.py:

import argparse
import json
import os
import sqlite3
from contextlib import closing
from datetime import datetime
from pathlib import Path

SAFE_FIELDS = ["measurement_id", "batch_code", "measured_at", "value_c"]

def inside(root, relative, expected_parent):
    value = Path(relative)
    if value.is_absolute():
        raise ValueError(f"Path must be relative: {relative}")
    resolved = (root / value).resolve()
    boundary = (root / expected_parent).resolve()
    try:
        resolved.relative_to(boundary)
    except ValueError as error:
        raise ValueError(f"Path must stay inside {expected_parent}/: {relative}") from error
    return resolved

def load_config(path):
    root = path.resolve().parent
    config = json.loads(path.read_text(encoding="utf-8"))
    if config.get("table") != "measurements":
        raise ValueError("Only the reviewed measurements table is allowed")
    if config.get("model_fields") != SAFE_FIELDS:
        raise ValueError(f"model_fields must be exactly {SAFE_FIELDS}")
    if not 0 <= config["max_null_rate"] <= 1:
        raise ValueError("max_null_rate must be between 0 and 1")
    if not 0 < config["min_rows"] <= config["max_rows"]:
        raise ValueError("row thresholds are invalid")
    if config["min_value"] > config["max_value"]:
        raise ValueError("measurement range is invalid")
    return root, config

def acquire_lock(path):
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError as error:
        raise RuntimeError(f"overlap blocked by existing lock: {path}") from error
    os.write(descriptor, f"pid={os.getpid()}\n".encode())
    return descriptor

def read_and_gate(database, config):
    uri = f"{database.as_uri()}?mode=ro"
    with closing(sqlite3.connect(uri, uri=True)) as connection:
        connection.execute("PRAGMA query_only = ON")
        connection.execute("BEGIN")
        actual = [
            [row[1], row[2].upper(), row[3], row[5]]
            for row in connection.execute("PRAGMA table_info(measurements)")
        ]
        if actual != config["expected_schema"]:
            raise ValueError(f"schema gate failed: expected {config['expected_schema']}, received {actual}")

        rows = connection.execute("""
            SELECT measurement_id, batch_code, measured_at, value_c
            FROM measurements
            WHERE unit = ?
            ORDER BY measurement_id
        """, (config["expected_unit"],)).fetchall()

        count = len(rows)
        if not config["min_rows"] <= count <= config["max_rows"]:
            raise ValueError(f"row-count gate failed: {count}")
        identifiers = [row[0] for row in rows]
        if len(identifiers) != len(set(identifiers)):
            raise ValueError("uniqueness gate failed")

        nulls = sum(row[3] is None for row in rows)
        null_rate = nulls / count
        if null_rate > config["max_null_rate"]:
            raise ValueError(f"null-rate gate failed: {null_rate:.3f}")

        values = []
        payloads = []
        for row in rows:
            timestamp = datetime.fromisoformat(row[2])
            if timestamp.tzinfo is None:
                raise ValueError(f"timestamp gate failed: {row[0]} has no UTC offset")
            if row[3] is None or not config["min_value"] <= row[3] <= config["max_value"]:
                raise ValueError(f"range gate failed: {row[0]} value={row[3]}")
            values.append(row[3])
            payloads.append(dict(zip(SAFE_FIELDS, row, strict=True)))
        return payloads, null_rate, min(values), max(values)

def publish(path, document):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.replace(path)

def run(config_path):
    root, config = load_config(config_path)
    database = inside(root, config["database"], "source")
    output = inside(root, config["output"], "output")
    lock = inside(root, config["lock"], "state")
    descriptor = acquire_lock(lock)
    try:
        payloads, null_rate, minimum, maximum = read_and_gate(database, config)
        document = {
            "gate_status": "accepted",
            "model_fields": SAFE_FIELDS,
            "row_count": len(payloads),
            "null_rate": null_rate,
            "minimum": minimum,
            "maximum": maximum,
            "model_payloads": payloads
        }
        publish(output, document)
        print(
            f"PASS schema={len(config['expected_schema'])} columns "
            f"rows={len(payloads)} null_rate={null_rate:.3f} "
            f"range={minimum:.1f}..{maximum:.1f} output={output}"
        )
    finally:
        os.close(descriptor)
        lock.unlink(missing_ok=True)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", type=Path, required=True)
    arguments = parser.parse_args()
    try:
        run(arguments.config)
    except (KeyError, OSError, RuntimeError, TypeError, ValueError, json.JSONDecodeError, sqlite3.Error) as error:
        raise SystemExit(f"FAIL: {error}") from error

Opening by URI with mode=ro makes an accidental write fail at SQLite. query_only adds connection-level intent. The exact SELECT is the field boundary. The transaction gives the schema and rows one database snapshot. Every gate completes before publish; Path.replace then replaces one complete JSON artifact on the same filesystem. Do not put the temporary file on a different mount.

Test success, corruption, privacy, and overlap

Save this as steps/test_pipeline.py:

import hashlib
import json
import sqlite3
import subprocess
import sys
import unittest
from contextlib import closing
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
CONFIG = ROOT / "config.json"
DATABASE = ROOT / "source" / "measurements.db"
OUTPUT = ROOT / "output" / "accepted-run.json"
LOCK = ROOT / "state" / "pipeline.lock"

def command():
    return subprocess.run(
        [sys.executable, str(ROOT / "steps" / "pipeline.py"), "--config", str(CONFIG)],
        cwd=ROOT,
        capture_output=True,
        text=True
    )

def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()

class LiveDataPipelineTest(unittest.TestCase):
    def setUp(self):
        subprocess.run([sys.executable, str(ROOT / "steps" / "make_fixture.py")], check=True)
        OUTPUT.unlink(missing_ok=True)
        LOCK.unlink(missing_ok=True)

    def test_accepted_run_contains_only_safe_fields(self):
        result = command()
        self.assertEqual(result.returncode, 0, result.stderr)
        document = json.loads(OUTPUT.read_text(encoding="utf-8"))
        self.assertEqual(document["gate_status"], "accepted")
        self.assertEqual(document["row_count"], 4)
        self.assertTrue(all(set(row) == set(document["model_fields"]) for row in document["model_payloads"]))
        rendered = OUTPUT.read_text(encoding="utf-8")
        self.assertNotIn("@example.invalid", rendered)
        self.assertNotIn("PRIVATE-CANARY", rendered)

    def test_read_only_connection_refuses_write(self):
        uri = f"{DATABASE.resolve().as_uri()}?mode=ro"
        with closing(sqlite3.connect(uri, uri=True)) as connection:
            with self.assertRaisesRegex(sqlite3.OperationalError, "readonly"):
                connection.execute("DELETE FROM measurements")

    def test_corrupt_value_halts_without_replacing_output(self):
        first = command()
        self.assertEqual(first.returncode, 0, first.stderr)
        before = digest(OUTPUT)
        with closing(sqlite3.connect(DATABASE)) as connection:
            connection.execute("UPDATE measurements SET value_c = 999 WHERE measurement_id = 'M-103'")
            connection.commit()
        failed = command()
        self.assertNotEqual(failed.returncode, 0)
        self.assertIn("range gate failed", failed.stderr)
        self.assertEqual(before, digest(OUTPUT))

    def test_existing_lock_blocks_overlap(self):
        LOCK.parent.mkdir(parents=True, exist_ok=True)
        LOCK.write_text("synthetic existing run\n", encoding="utf-8")
        result = command()
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("overlap blocked", result.stderr)
        self.assertFalse(OUTPUT.exists())

if __name__ == "__main__":
    unittest.main()

From the project root, run:

py -3 steps/make_fixture.py
py -3 steps/pipeline.py --config config.json
py -3 -m unittest steps/test_pipeline.py -v

On macOS or Linux, replace py -3 with python3. The direct run should end with:

PASS schema=7 columns rows=4 null_rate=0.000 range=18.2..21.4 output=...accepted-run.json

The test runner must report four tests and OK. The deliberate corruption test succeeds only when the nested pipeline process fails with range gate failed and the accepted output hash remains unchanged. A red outer test caused by the deliberate bad row is not a pass.

Inspect output/accepted-run.json. It must contain four payloads and no operator_email, notes, @example.invalid, or PRIVATE-CANARY. Do not send this exercise payload to a model; the artifact proves preparation at the boundary without introducing a provider.

Lab framing: an instrument-store contract

The fixture represents an instrument store where the domain owner accepts Celsius values from 0.0 through 50.0, requires four through six rows in this bounded run, accepts no absent measurement, and requires timezone-aware event times. In a real integration, replace SQLite only after the source owner supplies a documented read replica, view, or account with column-level SELECT permission.

Ask the database administrator to prove the role from the database side. For PostgreSQL, for example, the reviewed grant should target a view or explicit safe columns, not the entire schema:

GRANT CONNECT ON DATABASE measurements TO pipeline_reader;
GRANT USAGE ON SCHEMA reporting TO pipeline_reader;
GRANT SELECT (measurement_id, batch_code, measured_at, value_c, unit)
ON reporting.measurements TO pipeline_reader;

The administrator, not the pipeline process, executes grants. The runtime receives the resulting connection secret through the approved secret store. A negative test in staging attempts an UPDATE and must receive a permission denial. Do not run that test against a source unless the owner has approved the exact harmless canary row and transaction plan.

Company framing: a reproducible CRM row-count gate

Build this alternative in a separate disposable copy of the project; use it instead of the Lab fixture. Keep pipeline.py from the Lab example as the production pattern, but use the compact Company harness below to prove the same four controls with CRM-shaped data. Save this configuration as crm-config.json:

{
  "database": "source/crm.db",
  "table": "activities",
  "period": "2026-W36",
  "expected_schema": [
    ["activity_id", "TEXT", 1, 1],
    ["account_segment", "TEXT", 1, 0],
    ["occurred_at", "TEXT", 1, 0],
    ["activity_type", "TEXT", 1, 0],
    ["period", "TEXT", 1, 0],
    ["contact_email", "TEXT", 1, 0],
    ["owner_notes", "TEXT", 1, 0]
  ],
  "model_fields": ["activity_id", "account_segment", "occurred_at", "activity_type"],
  "min_rows": 4,
  "max_rows": 5,
  "output": "output/crm-accepted.json",
  "lock": "state/crm.lock"
}

Save this complete synthetic harness as steps/crm_example.py. It creates four fictional activities, opens the database read-only, checks the exact schema, selects only approved fields, applies row-count, uniqueness, and timezone gates, and publishes atomically:

import hashlib, json, os, sqlite3, subprocess, sys, unittest
from contextlib import closing
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
C = json.loads((ROOT / "crm-config.json").read_text(encoding="utf-8"))
DB, OUT, LOCK = (ROOT / C["database"], ROOT / C["output"], ROOT / C["lock"])
SAFE = C["model_fields"]
ROWS = [
    ("A-101", "Small", "2026-09-01T09:00:00+00:00", "Meeting", "2026-W36", "private1@example.invalid", "PRIVATE-CANARY-1"),
    ("A-102", "Mid", "2026-09-01T10:00:00+00:00", "Call", "2026-W36", "private2@example.invalid", "PRIVATE-CANARY-2"),
    ("A-103", "Small", "2026-09-02T11:00:00+00:00", "Demo", "2026-W36", "private3@example.invalid", "PRIVATE-CANARY-3"),
    ("A-104", "Enterprise", "2026-09-03T12:00:00+00:00", "Meeting", "2026-W36", "private4@example.invalid", "PRIVATE-CANARY-4")
]

def fixture():
    DB.parent.mkdir(parents=True, exist_ok=True); DB.unlink(missing_ok=True)
    with closing(sqlite3.connect(DB)) as con:
        con.execute("CREATE TABLE activities (activity_id TEXT PRIMARY KEY NOT NULL, account_segment TEXT NOT NULL, occurred_at TEXT NOT NULL, activity_type TEXT NOT NULL, period TEXT NOT NULL, contact_email TEXT NOT NULL, owner_notes TEXT NOT NULL)")
        con.executemany("INSERT INTO activities VALUES (?,?,?,?,?,?,?)", ROWS)
        con.commit()

def run():
    LOCK.parent.mkdir(parents=True, exist_ok=True)
    try: fd = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError as error: raise RuntimeError("overlap blocked") from error
    try:
        uri = f"{DB.resolve().as_uri()}?mode=ro"
        with closing(sqlite3.connect(uri, uri=True)) as con:
            con.execute("PRAGMA query_only=ON")
            actual = [[r[1], r[2].upper(), r[3], r[5]] for r in con.execute("PRAGMA table_info(activities)")]
            if actual != C["expected_schema"]: raise ValueError("schema gate failed")
            rows = con.execute("SELECT activity_id, account_segment, occurred_at, activity_type FROM activities WHERE period=? ORDER BY activity_id", (C["period"],)).fetchall()
        if not C["min_rows"] <= len(rows) <= C["max_rows"]: raise ValueError(f"row-count gate failed: {len(rows)}")
        if len({r[0] for r in rows}) != len(rows): raise ValueError("uniqueness gate failed")
        if any(datetime.fromisoformat(r[2]).tzinfo is None for r in rows): raise ValueError("timestamp gate failed")
        document = {"gate_status": "accepted", "row_count": len(rows), "model_fields": SAFE,
                    "model_payloads": [dict(zip(SAFE, row, strict=True)) for row in rows]}
        OUT.parent.mkdir(parents=True, exist_ok=True); tmp = OUT.with_suffix(".tmp")
        tmp.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8"); tmp.replace(OUT)
        print(f"PASS rows={len(rows)} safe_fields={len(SAFE)} output={OUT}")
    finally:
        os.close(fd); LOCK.unlink(missing_ok=True)

def command():
    return subprocess.run([sys.executable, __file__, "--run"], cwd=ROOT, capture_output=True, text=True)

class CRMTests(unittest.TestCase):
    def setUp(self): fixture(); OUT.unlink(missing_ok=True); LOCK.unlink(missing_ok=True)
    def test_safe_accepted_run(self):
        self.assertEqual(command().returncode, 0); text = OUT.read_text(encoding="utf-8")
        self.assertNotIn("@example.invalid", text); self.assertNotIn("PRIVATE-CANARY", text)
        document = json.loads(text); self.assertEqual(document["row_count"], 4)
        self.assertTrue(all(set(row) == set(SAFE) for row in document["model_payloads"]))
    def test_read_only_refuses_write(self):
        with closing(sqlite3.connect(f"{DB.resolve().as_uri()}?mode=ro", uri=True)) as con:
            with self.assertRaisesRegex(sqlite3.OperationalError, "readonly"): con.execute("DELETE FROM activities")
    def test_low_count_preserves_output(self):
        self.assertEqual(command().returncode, 0); before = hashlib.sha256(OUT.read_bytes()).hexdigest()
        with closing(sqlite3.connect(DB)) as con:
            con.execute("DELETE FROM activities WHERE activity_id IN ('A-103','A-104')"); con.commit()
        failed = command(); self.assertNotEqual(failed.returncode, 0); self.assertIn("row-count gate failed: 2", failed.stderr)
        self.assertEqual(before, hashlib.sha256(OUT.read_bytes()).hexdigest())
    def test_overlap_is_blocked(self):
        LOCK.parent.mkdir(parents=True, exist_ok=True); LOCK.write_text("synthetic run\n", encoding="utf-8")
        failed = command(); self.assertNotEqual(failed.returncode, 0); self.assertIn("overlap blocked", failed.stderr)

if __name__ == "__main__":
    if "--run" in sys.argv:
        try: run()
        except Exception as error: raise SystemExit(f"FAIL: {error}") from error
    else: unittest.main(argv=[sys.argv[0]])

Run py -3 steps/crm_example.py (python3 on macOS or Linux). The result must be four tests and OK. This is equivalent to the Lab gate set: accepted safe output, database-enforced read-only access, deliberate quality corruption without replacement, and overlap rejection. A production reporting view should omit contact and note fields entirely; their synthetic canaries exist here only to make leakage detectable. Scale 4-5 to an owner-approved historical band such as 800-1,200 before staging, never after seeing a failed run.

Annotated visual evidence (synthetic Company run):

[A] PASS rows=4 safe_fields=4 output=.../crm-accepted.json
[B] FAIL: row-count gate failed: 2
[C] accepted hash before corruption == accepted hash after corruption
[D] test_safe... ok | test_read_only... ok | test_low_count... ok | test_overlap... ok
    Ran 4 tests ... OK

Figure 1. Annotated terminal capture layout. Text alternative: A marks the accepted four-row baseline; B marks the deliberate two-row failure; C states that the before and after hashes match; D lists four passing controls. This course-authored text rendering uses only the synthetic fixture and contains no UI, identity, secret, or volatile product state. Reproduce it from the commands rather than treating it as evidence of your run. Provenance: T10-L04 fixture, reviewed 2026-09-04; privacy review: synthetic canaries excluded; staleness review: standard-library CLI, recheck if commands or expected messages change.

Schedule without racing

The application lock already rejects overlap. On a Linux host, a systemd timer can provide the schedule while the script remains the final overlap control. Use absolute paths, a dedicated unprivileged service identity, and a secret-free configuration:

# /etc/systemd/system/t10-l04-pipeline.service
[Unit]
Description=Gated read-only data pipeline

[Service]
Type=oneshot
User=pipeline-reader
WorkingDirectory=/opt/real-data-pipeline
ExecStart=/usr/bin/python3 /opt/real-data-pipeline/steps/pipeline.py --config /opt/real-data-pipeline/config.json
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/opt/real-data-pipeline/output /opt/real-data-pipeline/state

# /etc/systemd/system/t10-l04-pipeline.timer
[Unit]
Description=Run the gated pipeline every hour

[Timer]
OnCalendar=hourly
Persistent=true
RandomizedDelaySec=60

[Install]
WantedBy=timers.target

Have the platform owner review paths and hardening for the actual host before enabling it. Persistent=true can trigger a missed run after downtime; decide whether that is acceptable. Verify the installed units with systemd-analyze verify, run the service once against the synthetic source, and inspect systemctl status plus journalctl -u t10-l04-pipeline.service. Do not log source rows or connection strings.

7. What goes wrong

The credential can write

Symptom: the pipeline's identity owns the table or an UPDATE succeeds in staging.

Fix: stop the integration. Ask the source administrator for a dedicated role with only the required connection, schema, view, table, and column reads. Rotate the overbroad credential and rerun a harmless negative permission test.

There is no schema gate

Symptom: a renamed column becomes null, an extra sensitive field appears in output, or a positional import assigns values to the wrong meaning.

Fix: compare the source's actual metadata with an exact reviewed contract before selecting records. Treat missing, renamed, reordered where relevant, retyped, or unexpected fields as a failed run. Update the contract only through source-owner review and fixture tests.

Quality thresholds never fire

Symptom: minimum is zero, maximum is effectively infinite, or a range was widened after every failure until the dashboard stayed green.

Fix: derive narrow thresholds from domain and process knowledge, record owner and date, and inject one below, one above, one null, one duplicate, and one malformed timestamp. A gate is real only when its failure stops publication.

Personal fields reach the model

Symptom: the query uses SELECT *, then relies on a prompt saying not to use email or notes.

Fix: expose a safe database view or explicit column grant, select an allow-list, construct a new payload with exact keys, and test both key names and canary values. Treat logs, traces, eval exports, and dead-letter queues as part of the same boundary.

Two runs overlap

Symptom: two processes read different snapshots and the older one publishes last, or both write the same temporary path.

Fix: select skip or queue deliberately, acquire one lock before source access, use run-specific temporary state, and alert on skipped schedules. A stale lock needs an operator procedure that first proves no process still owns the run; never delete it automatically merely because it is inconvenient.

A failed run leaves plausible old output

Symptom: consumers see yesterday's accepted file and assume today's red run produced it.

Fix: include run identity and source snapshot metadata under the approved retention policy, alert on failure, and make consumers check freshness. Preserve the last accepted output for recovery, but never label its presence as evidence of a current success.

A real-source test changes real data

Symptom: a negative permission test writes a production row before proving the credential is read-only.

Fix: perform the first permission test against an approved synthetic staging source. Have the database administrator inspect grants independently. Do not use a production rollback as the planned safety control.

8. Do it yourself: corrupt an input and prove the stop in 120 minutes

Minutes 0-15: choose the Lab or Company framing and write the source contract before connecting anything: owner, table or view, safe fields, prohibited fields, read-only identity, expected schema, row band, null ceiling, range or category rules, timezone rule, schedule, overlap decision, and publication owner. Use the supplied synthetic fixture for execution.

Minutes 15-35: create the files for your chosen lane. Lab: create config.json, steps/make_fixture.py, and source/measurements.db. Company: in a separate disposable copy, create crm-config.json and steps/crm_example.py; its fixture() function creates source/crm.db. Inspect the chosen seven-column schema. Confirm the sensitive canary fields are fake and that the safe field list contains only four task-required fields.

Minutes 35-60: complete the chosen executable path: add steps/pipeline.py for Lab, or use run() in steps/crm_example.py for Company. Trace the lock, read-only URI, query-only setting, schema comparison, explicit query, quality gates, safe payload construction, temporary write, atomic replacement, and lock release. Also trace the Lab transaction if you chose Lab. Run the accepted fixture once and inspect every output key and value.

Minutes 60-80: run the four tests for your lane: add and run steps/test_pipeline.py for Lab, or run steps/crm_example.py for Company. Confirm the database refuses a write, accepted output contains no sensitive key or canary, a pre-existing lock blocks the run, and the known-good path produces exactly four payloads.

Minutes 80-95: perform the required corruption for your lane. Lab: change only M-103 from value_c=20.0 to value_c=999; require a non-zero result containing range gate failed: M-103 value=999.0. Company: delete only A-103 and A-104, exactly as test_low_count_preserves_output does; the selected period then has two rows, so require a non-zero result containing row-count gate failed: 2. In either lane, prove the accepted output's SHA-256 hash did not change, recreate the fixture, and require a green run again.

Minutes 95-108: adapt the written contract, not the live connection, to your real source. Ask its owner to confirm every field and threshold. Record how the runtime secret will be injected, how database-side read-only permission will be tested, and where failure alerts go. Do not paste the secret into this record.

Minutes 108-116: document the schedule and overlap policy. Trigger the lock test again and record whether a missed run is skipped or queued, who receives the alert, how freshness is represented, and how a stale lock is investigated without starting a second process.

Minutes 116-120: assemble one passing-test record with lane, command, timestamp, fixture revision, accepted-output hash before corruption, the lane-specific corruption details, non-zero exit, exact gate message, accepted-output hash after corruption, privacy assertions, overlap assertion, reviewer, and decision. Remove the disposable database and outputs when their retention period ends.

9. Exit check

Deliver exactly one artifact: one passing test record for either the Lab lane or the Company lane, showing that its quality gate halted the pipeline on deliberately corrupted synthetic input.

Every passing record must identify the chosen lane and the tested code and configuration revision; show an accepted four-row baseline; prove the accepted output hash is unchanged after corruption; report all four automated tests passing; confirm the database write was refused, sensitive canaries were absent, and overlap was blocked; name the reviewer and date; and end with Decision: PASS. Its lane-specific evidence must be one of these:

Lab: record M-103 value_c=999 as the sole corruption and capture the non-zero result containing range gate failed: M-103 value=999.0.
Company: record deletion of only A-103 and A-104, state that two rows remain for 2026-W36, and capture the non-zero result containing row-count gate failed: 2.

Choose one lane; do not submit one record per lane. The generated database, scripts, configuration, and JSON support this single passing-test artifact; they are not additional exit artifacts. A screenshot of a green scheduler, a model answer, or a failed run that overwrote output does not pass.

10. Rule to remember

Fail loudly on a shape you did not expect.

11. Further reading & tools

Taught: T10-L03 - A reproducible analysis pipeline - supplies the raw/steps/output discipline, deterministic rerun, and failure-before-publication baseline extended here.
Taught: T12-L04 - Put a lock on it - supplies least privilege, secret, network, logging, and external-audit controls for the real connection.
Taught: Python sqlite3 documentation (opens in a new tab) - primary API reference for URI connections, transactions, queries, and exceptions used by the fixture.
Taught: SQLite URI filenames (opens in a new tab) - primary reference for mode=ro in the exercise connection.
Catalogued: PostgreSQL GRANT (opens in a new tab) - primary reference for database-side privileges; an administrator must tailor and review the actual grants.
Catalogued: systemd timer units (opens in a new tab) - primary scheduling reference; verify host-specific service hardening separately.
Catalogued: KNIME - a visual workflow alternative that still needs exact schema, quality, field, publication, and overlap gates.
Catalogued: T06-L04 - Real data behind it - applies related source, migration, and access boundaries to a colleague-facing application.
Catalogued: T04-L04 - Permission-aware retrieval over real systems - extends real-data quality and access decisions into shared retrieval.
Catalogued: T10-L05 - Operating data pipelines - continues with versioned inputs and outputs, provenance, drift, retention, and publication authority.
Catalogued: Tools index - compare approved databases and orchestration tools only after the contracts and gates are defined.