Python, SQLAlchemy & Pydantic in Practice: The Code and Commands That Really Matter
Python SQLAlchemy Pydantic: The Essentials in One Article — Real Code, Schemas and Concrete Steps, Excerpts from a 27-Lesson Course.
No endless theory here: open the terminal and practice. Here's the essentials of Python SQLAlchemy Pydantic, extracted directly from a complete 27-lesson course — with real code you can copy-paste right now.
- Introduction
- SQLAlchemy Core
- ORM Declarative
- Session and transactions
- Advanced queries
Factories with factory-boy
factory-boy and Faker, without duplicating code in every test.Learning objectives
- Explain why hand-written fixtures become unmanageable
- Define a
SQLAlchemyModelFactorybound to a session - Use
Faker,SequenceandSubFactoryfor realistic data - Handle relationships (one-to-many) via
RelatedFactoryList - Wire factories into the pytest fixtures from the previous module
The intuition: a factory rather than an empty form
In a test, you need a valid User object. Written by hand, it looks like User(name="Test", email="a@b.c", age=30, is_active=True, ...). Repeated across 40 tests, that's 40 times the same chore, and the day you add a NOT NULL column, all 40 tests break.
A factory is a production line: you describe once how to build a typical User, then each test simply calls UserFactory() and receives a complete, valid object. If a test needs a special case, it only overrides the relevant field: UserFactory(age=17). Everything else is filled automatically.
Declarative
We describe the model type, not each instance. Test code stays short and readable.
Realistic
Faker generates plausible names, emails and addresses instead of "aaa".
Reproducible
You can fix the seed to obtain the same data on every run.
Installation and first factory
Install factory-boy (which bundles Faker):
RelatedFactoryList with size=100 inserts 100 objects one by one. For large seed volumes, prefer create_batch or bulk insert.Wire factories into pytest
We reuse the db_session fixture from the previous module (automatic rollback). We inject this session into the factories, then expose convenient fixtures.
Complex queries and reporting
Learning objectives
- Write a multi-table aggregation (customers, orders, lines) with
GROUP BY - Rank products with the window function
rank() - Structure a readable query with a CTE
- Map a heterogeneous result (Row) to a Pydantic reporting schema
- Validate these queries with factory-data tests
The intuition: separate transactional from analytical
So far, the ERP creates orders (transactional). Reporting answers different questions: who are my best customers? which product sells best this month? These queries do not modify anything; they aggregate and rank. We isolate them in a reporting/ module to avoid mixing responsibilities.
Transactional services
Create and modify entities (place an order, decrement stock). One session, one transaction.
Reporting queries
Read-only, heavy aggregations. Return rows (Row) rather than ORM entities.
Revenue per customer
Join Customer, Order and OrderLine, then sum quantity * unit_price. We use the 2.0 select() API and func for aggregation.
ERP Project - Models
Final project • Requirements • ORM models
Specifications
- Manage customers (B2B/B2C) with addresses
- Product catalog with multi-warehouse stock
- Multi-line orders with workflow status
- Automatic invoicing with VAT
- Audited stock movements
- Reports by month / by customer / by product
Project structure
erp/ ├── alembic.ini ├── migrations/ ├── app/ │ ├── core/ │ │ ├── config.py # Settings │ │ └── database.py │ ├── models/ │ │ ├── customer.py │ │ ├── product.py │ │ ├── order.py │ │ └── invoice.py │ ├── schemas/ # Pydantic In/Out │ ├── services/ # Business logic │ ├── api/ # FastAPI routes │ └── main.py └── tests/
Customer model
import enum from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy import ForeignKey from datetime import datetime class CustomerType(str, enum.Enum): B2B = "b2b" B2C = "b2c" class Customer(Base): __tablename__ = "customers" id: Mapped[int] = mapped_column(primary_key=True) type: Mapped[CustomerType] name: Mapped[str] email: Mapped[str] = mapped_column(unique=True) tax_id: Mapped[str | None] # for B2B created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) addresses: Mapped[list["Address"]] = relationship(back_populates="customer", cascade="all") orders: Mapped[list["Order"]] = relationship(back_populates="customer") class Address(Base): __tablename__ = "addresses" id: Mapped[int] = mapped_column(primary_key=True) customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id")) street: Mapped[str] city: Mapped[str] zip_code: Mapped[str] country: Mapped[str] is_billing: Mapped[bool] = mapped_column(default=False) is_shipping: Mapped[bool] = mapped_column(default=True) customer: Mapped["Customer"] = relationship(back_populates="addresses")
Product and Stock models
from decimal import Decimal class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) sku: Mapped[str] = mapped_column(unique=True) name: Mapped[str] price: Mapped[Decimal] = mapped_column(Numeric(10, 2)) vat_rate: Mapped[Decimal] = mapped_column(Numeric(4, 2), default=Decimal("20.00")) active: Mapped[bool] = mapped_column(default=True) stocks: Mapped[list["Stock"]] = relationship(back_populates="product") class Warehouse(Base): __tablename__ = "warehouses" id: Mapped[int] = mapped_column(primary_key=True) code: Mapped[str] = mapped_column(unique=True) name: Mapped[str] class Stock(Base): __tablename__ = "stocks" id: Mapped[int] = mapped_column(primary_key=True) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) warehouse_id: Mapped[int] = mapped_column(ForeignKey("warehouses.id")) quantity: Mapped[int] = mapped_column(default=0) __table_args__ = ( UniqueConstraint("product_id", "warehouse_id"), CheckConstraint("quantity >= 0"), ) product: Mapped["Product"] = relationship(back_populates="stocks")
Order and OrderLine models
class OrderStatus(str, enum.Enum): DRAFT = "draft" CONFIRMED = "confirmed" SHIPPED = "shipped" DELIVERED = "delivered" CANCELLED = "cancelled" class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id")) reference: Mapped[str] = mapped_column(unique=True) status: Mapped[OrderStatus] = mapped_column(default=OrderStatus.DRAFT) created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) customer: Mapped["Customer"] = relationship(back_populates="orders") lines: Mapped[list["OrderLine"]] = relationship(back_populates="order", cascade="all, delete-orphan") invoice: Mapped["Invoice"] = relationship(back_populates="order", uselist=False) class OrderLine(Base): __tablename__ = "order_lines" id: Mapped[int] = mapped_column(primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("orders.id")) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) quantity: Mapped[int] = mapped_column() unit_price: Mapped[Decimal] = mapped_column(Numeric(10, 2)) order: Mapped["Order"] = relationship(back_populates="lines") product: Mapped["Product"] = relationship()
Invoice and StockMovement models
class Invoice(Base): __tablename__ = "invoices" id: Mapped[int] = mapped_column(primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), unique=True) number: Mapped[str] = mapped_column(unique=True) issued_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) subtotal: Mapped[Decimal] = mapped_column(Numeric(10, 2)) vat: Mapped[Decimal] = mapped_column(Numeric(10, 2)) total: Mapped[Decimal] = mapped_column(Numeric(10, 2)) paid: Mapped[bool] = mapped_column(default=False) order: Mapped["Order"] = relationship(back_populates="invoice") class StockMovement(Base): __tablename__ = "stock_movements" id: Mapped[int] = mapped_column(primary_key=True) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) warehouse_id: Mapped[int] = mapped_column(ForeignKey("warehouses.id")) quantity: Mapped[int] # negative = outbound reason: Mapped[str] # "order_123", "restock" created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
Summary
- Models separated by domain (customer, order, product)
- Decimal for money (never float)
- Enum for workflow statuses
- cascade for aggregations (Order -> lines)
- StockMovement = immutable audit log
This article covers the most useful excerpts — the complete Python SQLAlchemy Pydantic course (9 chapters, 27 lessons, corrected exercises and final project) takes you all the way.
./access-the-complete-course free course: Vibe CodingFAQ
How long does it take to learn Python SQLAlchemy Pydantic?
Are there any prerequisites?
Where to start concretely?
📬 Want to receive this kind of guide every week? Subscribe for free — real code, zero fluff.