Learn / Python and FastAPI Backend Engineering
Repository and Unit of Work
Own a subscription billing write flow with product-language repositories, one explicit unit-of-work transaction, rollback proof, and routes that never reach through ORM details.
Course: Python and FastAPI Backend Engineering. Level: Intermediate. Topic: Backend craft.
Stage: intermediate - Intermediate data ownership - Repositories, transactions, and unit-of-work ownership. Move persistence behind product-language boundaries and make transaction ownership reviewable.
Outcomes
- Design repositories around product questions instead of raw SQLAlchemy mechanics.
- Place commit and rollback ownership at a unit-of-work boundary for a complete write flow.
- Prove failure safety with tests that catch partial writes and lazy ORM leaks.
- Create tables and use parameterized SQL.
- Use transactions for multi-step changes.
- Hide database details behind repository functions.
- Understand when SQLite is appropriate for a FastAPI service.
- Use service functions to hold business workflows.
- Use repositories to isolate persistence.
- Use adapters for external APIs.
- Recognize when a pattern is adding value versus ceremony.
Concepts
- repository port
- unit-of-work boundary
- rollback proof
- ORM leak review
- SQLite Persistence
- SQLite is a real database with local scope
- Repository boundary
- Python + FastAPI foundations
- Guided practice
- Design Patterns
- A route should orchestrate, not own the business
- Adapters protect your core
Concept flow
Show how a route command becomes one committed business transaction, or rolls back without partial billing state.
- Route command
- Billing service
- Unit of Work
- Customer repository
- Invoice repository
- Commit or rollback
Session flow
- Name persistence by product questions (concept, 12 min) — Design repository methods around billing decisions: load a customer for billing, check idempotency, stage an invoice.
- Repository names should survive query rewrites.
- Routes should not know joins, eager-loading strategy, or transaction state.
- Commit once at the unit-of-work boundary (walkthrough, 25 min) — Implement a subscription write where idempotency, aggregate changes, invoice creation, and commit happen as one action.
- A service can coordinate repositories without owning session plumbing.
- Commit belongs after all business invariants have been checked and staged.
- Prove rollback and ORM isolation (exercise, 15 min) — Write failure tests that prove no partial invoice survives and no route depends on lazy ORM access.
- Rollback tests protect the business promise, not only database mechanics.
- DTO mapping at the boundary prevents closed-session surprises in production.
- SQLite, Repositories, and Transactions: SQLite is a real database with local scope (concept, 75 min) — SQLite stores data in a file and is excellent for development, tests, embedded apps, small internal tools, and read-heavy services. It is not a drop-in replacement for every multi-writer production workload.
- Retained source example: Create and query safely
import sqlite3
conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE notes (id INTEGER PRIMARY KEY, text TEXT NOT NULL)')
conn.execute('INSERT INTO notes (text) VALUES (?)', ('ship course',))
rows = conn.execute('SELECT text FROM notes WHERE id = ?', (1,)).fetchall()
print(rows[0][0])
Expected output: ship course
- SQLite, Repositories, and Transactions: Repository boundary (walkthrough, 75 min) — A repository function gives the rest of the application a stable API for persistence. Routes and services should not build SQL strings from user input or know every table detail.
- Retained source example: Repository function
import sqlite3
def get_note(conn: sqlite3.Connection, note_id: int) -> str | None:
row = conn.execute('SELECT text FROM notes WHERE id = ?', (note_id,)).fetchone()
return None if row is None else str(row[0])
- SQLite, Repositories, and Transactions: practices and mistakes (review, 5 min) — Retained guidance from the legacy PTesting lesson.
- Practice: Use parameterized SQL for every user-provided value.
- Practice: Keep connection lifetime explicit and testable.
- Practice: Wrap related writes in a transaction.
- Avoid: Building SQL with f-strings from request data.
- Avoid: Sharing one connection across threads without understanding driver settings.
- Avoid: Skipping migrations and manually changing schemas in production.
- SQLite, Repositories, and Transactions: references (review, 2 min) — Original references retained from the legacy library.
- sqlite3 module: https://docs.python.org/3/library/sqlite3.html
- SQLite documentation: https://www.sqlite.org/docs.html
- Service, Repository, and Adapter Patterns: A route should orchestrate, not own the business (concept, 75 min) — In a small app, a route may call one service. In a larger app, the service coordinates validation, repositories, adapters, and domain rules. This keeps HTTP details from spreading everywhere.
- Retained source example: Service shape
from dataclasses import dataclass
@dataclass(frozen=True)
class CreateTask:
title: str
def create_task(command: CreateTask) -> dict[str, str]:
title = command.title.strip()
if not title:
raise ValueError('title is required')
return {'title': title, 'status': 'open'}
- Service, Repository, and Adapter Patterns: Adapters protect your core (walkthrough, 75 min) — An adapter wraps an external system so the rest of the app depends on your small interface, not on every detail of a vendor SDK or HTTP API.
- Retained source example: Adapter interface
class EmailAdapter:
def send_welcome(self, email: str) -> None:
# Real implementation would call an email provider.
print(f'welcome queued for {email}')
- Service, Repository, and Adapter Patterns: practices and mistakes (review, 5 min) — Retained guidance from the legacy PTesting lesson.
- Practice: Introduce patterns when they reduce coupling or clarify responsibilities.
- Practice: Use plain functions and dataclasses where classes would add little value.
- Practice: Keep framework types at the application boundary when possible.
- Avoid: Creating abstract classes for every function before the app has real complexity.
- Avoid: Letting Pydantic API models become the entire domain model.
- Avoid: Putting validation, SQL, HTTP calls, and response formatting in one route.
- Service, Repository, and Adapter Patterns: references (review, 2 min) — Original references retained from the legacy library.
- FastAPI dependencies: https://fastapi.tiangolo.com/tutorial/dependencies/
- Architecture Patterns with Python: https://www.cosmicpython.com/book/preface.html
Code example
Python in app/orders/service.py.
from dataclasses import dataclass
from typing import Protocol
from uuid import UUID
@dataclass(frozen=True)
class StartSubscription:
customer_id: UUID
plan_id: UUID
idempotency_key: str
class CustomerRepository(Protocol):
async def get_for_billing(self, customer_id: UUID) -> Customer: ...
class InvoiceRepository(Protocol):
async def add(self, invoice: Invoice) -> None: ...
async def exists_for_key(self, idempotency_key: str) -> bool: ...
class UnitOfWork(Protocol):
customers: CustomerRepository
invoices: InvoiceRepository
async def __aenter__(self) -> "UnitOfWork": ...
async def __aexit__(self, exc_type, exc, tb) -> None: ...
async def commit(self) -> None: ...
async def rollback(self) -> None: ...
async def start_subscription(uow: UnitOfWork, command: StartSubscription) -> Invoice:
async with uow:
if await uow.invoices.exists_for_key(command.idempotency_key):
raise DuplicateCommand(command.idempotency_key)
customer = await uow.customers.get_for_billing(command.customer_id)
subscription = customer.start_subscription(command.plan_id)
invoice = Invoice.for_subscription(subscription, command.idempotency_key)
await uow.invoices.add(invoice)
await uow.commit()
return invoice
Walkthrough examples
- Subscription billing write pack — A team is adding paid subscriptions. A failed invoice write must not leave a customer subscribed without a billable invoice.
- File: app/orders/service.py
- File: app/orders/repositories.py
- File: app/orders/unit_of_work.py
- File: tests/test_orders_unit_of_work.py
- File: docs/python-fastapi/repository-and-unit-of-work.md
- Create repository ports for customer billing lookup and invoice idempotency instead of exposing SQLAlchemy queries.
- Implement a UnitOfWork that opens the session, exposes repositories, commits once, and rolls back on exceptions.
- Call start_subscription from the route using a command DTO and map the returned invoice into a response DTO.
- Add tests for duplicate idempotency key, successful commit, rollback after staged invoice, and no ORM relationship access in the route.
- Create and query safely — Additional example retained from the legacy lesson.
- Retained source code:
import sqlite3
conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE notes (id INTEGER PRIMARY KEY, text TEXT NOT NULL)')
conn.execute('INSERT INTO notes (text) VALUES (?)', ('ship course',))
rows = conn.execute('SELECT text FROM notes WHERE id = ?', (1,)).fetchall()
print(rows[0][0])
- Expected output: ship course
- Compare the example with the canonical PTLearn implementation.
- Repository function — Additional example retained from the legacy lesson.
- Retained source code:
import sqlite3
def get_note(conn: sqlite3.Connection, note_id: int) -> str | None:
row = conn.execute('SELECT text FROM notes WHERE id = ?', (note_id,)).fetchone()
return None if row is None else str(row[0])
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Transaction with context manager — The sqlite3 connection context commits on success and rolls back on error.
- Retained source code:
with conn:
conn.execute('INSERT INTO notes (text) VALUES (?)', ('atomic change',))
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Service shape — Additional example retained from the legacy lesson.
- Retained source code:
from dataclasses import dataclass
@dataclass(frozen=True)
class CreateTask:
title: str
def create_task(command: CreateTask) -> dict[str, str]:
title = command.title.strip()
if not title:
raise ValueError('title is required')
return {'title': title, 'status': 'open'}
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Adapter interface — Additional example retained from the legacy lesson.
- Retained source code:
class EmailAdapter:
def send_welcome(self, email: str) -> None:
# Real implementation would call an email provider.
print(f'welcome queued for {email}')
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Request path through layers — Additional example retained from the legacy lesson.
- Retained source code:
route -> input model -> service -> repository or adapter -> output model
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
Practice
- Rename one persistence method from a SQL-shaped name to a product question like get_for_billing.
- Move commit out of the route and repository so the unit of work owns the complete subscription write.
- Write one test that raises after invoice creation and proves rollback leaves no partial invoice.
- Use a parameterized query: Choose the placeholder style used by sqlite3 to avoid string interpolation for values.
- Starter code: conn.execute('SELECT * FROM notes WHERE id = ____', (note_id,))
- Hint: sqlite3 accepts question mark placeholders.
- Reference solution: ?
- Accepted answers: ?
- Place SQL in the right layer: Which layer should usually contain raw SQL: route, service, or repository?
- Hint: The layer that abstracts persistence details.
- Reference solution: repository
- Accepted answers: repository
Checklist
- Repositories are named in product language and hide SQLAlchemy query shape.
- The service owns the business decision but not the database session mechanics.
- Exactly one unit-of-work boundary commits the subscription write.
- Rollback and idempotency behavior have repeatable test evidence.
- Route responses are built from DTOs, not lazy ORM relationship traversal.
Quiz prompts
- Why should the repository method be named get_for_billing instead of select_customer_with_plan_join? — Repository APIs should communicate business intent; implementation details belong behind the boundary.
- Where should commit happen for a multi-step subscription write? — A unit of work lets the service treat related persistence changes as one business transaction.
- What does a rollback test need to prove? — Rollback evidence is about atomicity: either the whole business action is saved or none of it is.
- Which review smell suggests an ORM leak? — Lazy ORM access outside the transaction boundary creates fragile runtime behavior and unclear ownership.
- Why should user values be passed as query parameters instead of formatted into SQL strings? — The driver safely binds values when placeholders are used.
- What is the main benefit of an adapter around an external API? — Adapters make external dependencies easier to replace, fake, and test.
Flashcards
- What should a repository method name reveal? The product question it answers, such as get_for_billing, not the SQL or ORM mechanics used today.
- What does the unit of work own? The transaction boundary: session lifetime, commit, rollback, and repository coordination for one business action.
- What is rollback proof? A repeatable test showing a failed multi-step write leaves no partial persisted state.
- What is an ORM leak in a route? Route or response code depending on lazy ORM behavior, session state, or relationship traversal instead of DTOs.
- Why check idempotency inside the transaction? The duplicate check and write decision must be protected by the same business boundary to avoid double billing.
Labs
- Ship a rollback-safe subscription write — Refactor a billing endpoint so route code sends a command, service code owns the business action, and the unit of work owns commit/rollback.
- Create CustomerRepository and InvoiceRepository ports named around billing decisions.
- Implement UnitOfWork as an async context manager that rolls back on exceptions and commits only when told.
- Move subscription creation into start_subscription and return a DTO-safe Invoice result to the route.
- Write tests for successful commit, duplicate idempotency key, rollback after staged invoice, and route DTO mapping.
- Test output proves commit happens once for the successful subscription path.
- Rollback test evidence shows no partial invoice remains after an injected failure.
- Review notes include the before/after route showing ORM/session details removed from the HTTP layer.
Challenge
- Review-ready billing transaction boundary (Core) — Turn a raw ORM write flow into a product-language repository and unit-of-work slice with rollback evidence, idempotency handling, and route DTO isolation.
- Repository methods are named after billing decisions instead of SQL shape.
- Only the unit of work commits or rolls back the write transaction.
- Tests prove success, duplicate idempotency, rollback on failure, and route DTO mapping.
- The review note names one ORM leak removed from the route layer.
Canonical lesson URL