CHAPTER 49 · ADVANCED4 min read

Project: An API Data Pipeline

Build a fixture-driven catalog import with validation, atomic persistence, idempotency, and acceptance checks.

Python 3.12+Free to readModule 11 of 13

YOUR NEXT SMALL WINS

By the end, you’ll be able to…

Separate source retrieval, normalization, and persistence.
Reject a bad batch before changing durable data.
Verify repeated imports and stable reporting.

01Define the import contract

Build a catalog synchronizer for a training provider. Each source page is UTF-8 JSON containing an items array. Every item requires a positive integer id, a nonblank title of at most eighty characters, and a nonnegative integer minutes value. Reject booleans where integers are required. Reject duplicate identifiers within one import rather than allowing source order to choose a winner silently.

The importer must validate the complete bounded batch before writing, update existing source IDs, retain records absent from this incremental feed, and report rows in ID order. Importing the same pages twice must leave the same database contents. These requirements describe a useful learning project with explicit limitations; they are not a claim of production readiness or a complete synchronization protocol.

02Keep the transport replaceable

Use local bytes fixtures for the core implementation. A real HTTP adapter can later provide pages with the timeout, status, media-type, and response-size checks from the HTTP lesson. The importer should not know whether bytes came from a file, a test fixture, or an approved endpoint. This seam keeps the acceptance tests independent of network availability and changing remote content.

Set a batch limit before writing and preserve page and row context in validation errors. For an unbounded feed, one giant transaction and one giant in-memory list are inappropriate; design bounded pages with durable checkpoints and a stated partial-success policy. The small project below deliberately caps accepted rows and uses a whole-batch contract to make failure behavior easy to demonstrate.

03Implement validation and an atomic upsert

Save the following program as pipeline.py. parse_pages constructs normalized tuples and tracks IDs across pages. It returns only after the whole batch is valid. import_pages then applies a parameterized upsert inside one transaction. The primary key makes repeated source records update the same local identity, while CHECK and NOT NULL constraints preserve basic invariants at the database boundary.

The demonstration creates an in-memory database, imports the same fixture twice, and attempts an invalid batch. It compares the report before and after the failure, proving that validation failure did not change prior data. A database failure during executemany would also escape the transaction block and trigger rollback. Connection lifetime remains separate from transaction lifetime and ends in finally.

pipeline.py🐍Python

Ctrl + ↵

import json
import sqlite3

MAX_ROWS = 1000
MAX_PAGES = 20

def parse_pages(pages):
    rows, seen = [], set()
    for page_number, payload in enumerate(pages, 1):
        if page_number > MAX_PAGES:
            raise ValueError('batch exceeds page limit')
        if len(payload) > 100_000:
            raise ValueError(f'page {page_number}: too large')
        data = json.loads(payload.decode('utf-8'))
        if not isinstance(data, dict) or not isinstance(data.get('items'), list):
            raise ValueError(f'page {page_number}: expected items list')
        for row_number, item in enumerate(data['items'], 1):
            context = f'page {page_number}, row {row_number}'
            if not isinstance(item, dict):
                raise ValueError(f'{context}: expected object')
            identifier = item.get('id')
            title, minutes = item.get('title'), item.get('minutes')
            if type(identifier) is not int or identifier < 1:
                raise ValueError(f'{context}: invalid id')
            if identifier in seen:
                raise ValueError(f'{context}: duplicate id')
            if not isinstance(title, str) or not 1 <= len(title.strip()) <= 80:
                raise ValueError(f'{context}: invalid title')
            if type(minutes) is not int or minutes < 0:
                raise ValueError(f'{context}: invalid minutes')
            seen.add(identifier)
            rows.append((identifier, title.strip(), minutes))
            if len(rows) > MAX_ROWS:
                raise ValueError('batch exceeds row limit')
    return rows

def initialize(connection):
    with connection:
        connection.execute('''CREATE TABLE IF NOT EXISTS courses (
            id INTEGER PRIMARY KEY,
            title TEXT NOT NULL,
            minutes INTEGER NOT NULL CHECK(minutes >= 0))''')

def import_pages(connection, pages):
    rows = parse_pages(pages)
    with connection:
        connection.executemany('''INSERT INTO courses VALUES (?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
            title = excluded.title, minutes = excluded.minutes''', rows)
    return len(rows)

def report(connection):
    return connection.execute(
        'SELECT id, title, minutes FROM courses ORDER BY id').fetchall()

def main():
    pages = [
        b'{"items":[{"id":2,"title":" SQL ","minutes":90}]}',
        b'{"items":[{"id":1,"title":"Python","minutes":120}]}',
    ]
    connection = sqlite3.connect(':memory:', autocommit=False)
    try:
        initialize(connection)
        assert import_pages(connection, pages) == 2
        before = report(connection)
        assert import_pages(connection, pages) == 2
        assert report(connection) == before
        try:
            import_pages(connection, [b'{"items":[{"id":3,"title":"Bad","minutes":true}]}'])
        except ValueError:
            print('invalid batch rejected')
        assert report(connection) == before
        print(report(connection))
    finally:
        connection.close()

if __name__ == '__main__':
    main()

Output

Click Run or press Ctrl+Enter

Output will appear here after you run the code.

04Acceptance checks and extension path

Accept the project only when it imports valid pages, normalizes surrounding title whitespace, updates an existing ID, rejects duplicate IDs and boolean minutes, handles an empty items array, and leaves existing rows unchanged after a bad batch. Verify durable persistence with a temporary database file in addition to the in-memory checks. Test malformed JSON separately from a valid document with an invalid schema.

A useful next version adds an argparse interface, structured summary output, schema versioning, and an injected transport with bounded retries. A larger feed needs checkpoint semantics, source version information, and an operational recovery plan. Keep each extension tied to a requirement: concurrency is unnecessary until measurement identifies retrieval latency as a bottleneck, and multiple writers complicate the current simple transaction ownership model.

LET IT CLICK

The ideas worth keeping.

Replaceable inputs keep core tests deterministic.
Validate first and persist one bounded business change atomically.
Acceptance checks must cover idempotency and failed imports.

TURN UNDERSTANDING INTO SOMETHING YOU CAN DO

Make it yours.

Type and test your code right here in the interactive workspace. Run your solution, check the output, and compare with the worked guide when ready.

Specify duplicate rejection

Implement a small validate_ids function that rejects duplicate positive IDs before persistence. Test both valid and duplicate batches.

your_turn_1.py🐍Python

Ctrl + ↵

Output

Click Run or press Ctrl+Enter

Output will appear here after you run the code.

Explore a worked solution

one_possible_solution.pyPython

def validate_ids(values):    seen = set()    for value in values:        if type(value) is not int or value < 1 or value in seen:            raise ValueError('invalid or duplicate id')        seen.add(value)    return seenassert validate_ids([2, 1]) == {1, 2}try:    validate_ids([1, 1])except ValueError:    passelse:    raise AssertionError('duplicate accepted')

Rejecting duplicates gives deterministic failure instead of silently selecting the final record. The main pipeline extends the same rule across page boundaries.

Define a stable summary

Write summarize(rows), where rows are (id, title, minutes) tuples. Return count and total_minutes without depending on row order.

your_turn_2.py🐍Python

Ctrl + ↵

Output

Click Run or press Ctrl+Enter

Output will appear here after you run the code.

Explore a worked solution

one_possible_solution.pyPython

def summarize(rows):    count = total_minutes = 0    for _, _, minutes in rows:        count += 1        total_minutes += minutes    return {'count': count, 'total_minutes': total_minutes}assert summarize([(2, 'SQL', 90), (1, 'Python', 120)]) == {    'count': 2, 'total_minutes': 210}assert summarize([]) == {'count': 0, 'total_minutes': 0}

A single pass works with a list or iterator. Presentation ordering belongs to the report query, while this aggregate has an order-independent contract.

A SMALL QUESTION. A USEFUL NEXT STEP.

What stayed with you?

Check the ideas while they’re fresh. Your answers stay in this page and feedback appears as you go.

A LITTLE ACTIVE RECALL0 / 3 answered

01Why validate the complete bounded batch before writing?

ATo guarantee there can never be a database failure.BTo reject source-data errors before changing durable state.CTo avoid using transactions.

02What makes repeated imports preserve one row per source course?

APrinting the rows twice.BA stable primary key plus the defined upsert policy.CSorting titles alphabetically.

03Which extension requires a new failure contract?

AChanging a lesson title.BReplacing a bounded all-or-nothing batch with checkpointed streaming.CAdding a pure summary function.

⚡CODING CHALLENGE

🔒Required to Mark Lesson Complete

Specify duplicate rejection

Implement a small validate_ids function that rejects duplicate positive IDs before persistence. Test both valid and duplicate batches.

EXPECTED OUTPUT

[Process exited with code 0 - No output]

challenge.pyPython 3

Ctrl + ↵

Execution Console

Click "Run & Verify" (or press Ctrl + Enter) to test your code against the challenge.

Keep exploring.

The official references behind this chapter, when you want a second angle or the full details.

MODULE 11 · PRACTICAL SELF-CHECK

Pause. Put the pieces together.

You’ve reached the end of Data and Service Boundaries. Before moving on, use this task to check what you can do independently.

Import the same records twice, reject a bad batch, and prove existing database rows remain unchanged.

Try it in your own environment and write down the result. This is a self-check; reading completion does not evaluate your code.

Review this module

One chapter closer.

Reading is a start. Trying it yourself makes the connection.

Solve the hands-on coding challenge in the knowledge section to verify your code and unlock chapter completion.