Nested Models and Create / Update / Read Schemas

This lesson brings everything together and is the closest to a real project. It follows 06_nested_and_dto_schemas.py and has two parts:

• Part A, nested models (no database): a model can contain other models, like an order that contains a customer, who has an address.
• Part B, Create / Update / Read schemas (with SQL Server): the professional way to design Pydantic models for one database table, used with real insert, select, update and delete.

Part A: Nested models

A1. A model inside a model

Real data is rarely flat. An online order has a customer, the customer has an address, and the order has a list of items. In Pydantic, you simply use one model as the type of a field in another model:

from decimal import Decimal
from pydantic import BaseModel, Field, ValidationError, computed_field

class Address(BaseModel):
    street: str
    city: str
    pin_code: str = Field(pattern=r"^\d{6}$")


class OrderItem(BaseModel):
    product: str
    quantity: int = Field(ge=1)
    price: Decimal = Field(gt=0)

    @computed_field
    @property
    def line_total(self) -> Decimal:
        return self.quantity * self.price


class Customer(BaseModel):
    name: str
    address: Address                     # one nested model
    other_addresses: list[Address] = []  # a list of nested models


class Order(BaseModel):
    order_no: str
    customer: Customer
    items: list[OrderItem] = Field(min_length=1)

    @computed_field
    @property
    def total(self) -> Decimal:
        return sum((item.line_total for item in self.items), Decimal(0))

Nested data usually arrives as nested dictionaries or JSON. Pydantic builds every inner model for you:

order = Order.model_validate({
    "order_no": "ORD-1001",
    "customer": {"name": "Asha", "address": {"street": "MG Road", "city": "Pune", "pin_code": "411001"}},
    "items": [
        {"product": "Pen", "quantity": 10, "price": "12.50"},
        {"product": "Notebook", "quantity": 2, "price": "60"},
    ],
})
print("City     :", order.customer.address.city)        # reach inner values with dots
print("Items    :", [item.product for item in order.items])
print("Total    :", order.total)
print(order.model_dump_json(indent=2))

City     : Pune
Items    : ['Pen', 'Notebook']
Total    : 245.00
{
  "order_no": "ORD-1001",
  "customer": {
    "name": "Asha",
    "address": {
      "street": "MG Road",
      "city": "Pune",
      "pin_code": "411001"
    },
    "other_addresses": []
  },
  "items": [
    {
      "product": "Pen",
      "quantity": 10,
      "price": "12.50",
      "line_total": "125.00"
    },
    {
      "product": "Notebook",
      "quantity": 2,
      "price": "60",
      "line_total": "120"
    }
  ],
  "total": "245.00"
}

• Reach inner values with dots: order.customer.address.city.
• Every level is validated with its own rules: the PIN code pattern, the quantity range, the price check.
• Computed fields work at every level: each item has line_total, and the order adds them up into total (10 × 12.50 + 2 × 60 = 245.00).
• sum(..., Decimal(0)) starts the sum at a Decimal zero, so the result stays an exact Decimal.

A2. Errors inside nested models show the full path

try:
    Order.model_validate({
        "order_no": "ORD-1002",
        "customer": {"name": "Ravi", "address": {"street": "FC Road", "city": "Pune", "pin_code": "41"}},
        "items": [{"product": "Pen", "quantity": 1, "price": 5}, {"product": "Bag", "quantity": 0, "price": 500}],
    })
except ValidationError as error:
    for item in error.errors():
        print(f"   {item['loc']}: {item['msg']}")

   ('customer', 'address', 'pin_code'): String should match pattern '^\d{6}$'
   ('items', 1, 'quantity'): Input should be greater than or equal to 1

The error location is a path. Read ('items', 1, 'quantity') as "in items, at index 1 (the second item, the Bag), the field quantity". In a big JSON document, this tells the user exactly where to look.

Part B: Create / Update / Read schemas

Why one table needs three models

Think about the Employees table (from database.py, see lesson 05). The data looks different depending on what you are doing:

These models are often called schemas or DTOs (Data Transfer Objects): they describe the data that travels in and out of your program, separate from the database table class. FastAPI projects are organised exactly like this.

The schemas

from datetime import date
from pydantic import ConfigDict, EmailStr

class DepartmentSummary(BaseModel):
    """Small department info, used inside EmployeeRead."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str


class EmployeeBase(BaseModel):
    """Fields and rules shared by Create and Read."""
    first_name: str = Field(min_length=2, max_length=50)
    last_name: str = Field(min_length=1, max_length=50)
    email: EmailStr
    salary: Decimal = Field(gt=0, max_digits=10, decimal_places=2)
    hire_date: date


class EmployeeCreate(EmployeeBase):
    """What the user sends to ADD an employee. No id: SQL Server creates it."""
    department_id: int = Field(gt=0)


class EmployeeUpdate(BaseModel):
    """What the user sends to CHANGE an employee. Send only the fields to change."""
    first_name: str | None = Field(default=None, min_length=2, max_length=50)
    last_name: str | None = Field(default=None, min_length=1, max_length=50)
    email: EmailStr | None = None
    salary: Decimal | None = Field(default=None, gt=0, max_digits=10, decimal_places=2)
    is_active: bool | None = None
    department_id: int | None = Field(default=None, gt=0)


class EmployeeRead(EmployeeBase):
    """What we SEND BACK. Built from the ORM object, includes the nested department."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    is_active: bool
    department: DepartmentSummary            # nested model filled from employee.department


class EmployeeSummary(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    first_name: str
    last_name: str


class DepartmentRead(BaseModel):
    """A department with the list of its employees (nested list)."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    location: str | None
    employees: list[EmployeeSummary]          # filled from department.employees

• EmployeeBase holds the shared fields and rules. EmployeeCreate and EmployeeRead inherit from it, so each rule (name length, email check, salary limits) is written only once.
• EmployeeUpdate repeats the fields, but each one is | None with a default of None, so every field is optional. The rules still apply when a value is sent.
• EmployeeRead has a nested department: DepartmentSummary. With from_attributes=True, Pydantic reads orm_employee.department (the SQLAlchemy relationship) and builds the nested model too.
• DepartmentRead goes the other way: one department with a list of its employees, filled from the department.employees relationship.

The CRUD functions

Each function takes a schema in and gives a schema out. The ORM objects stay inside these functions; the rest of the program only ever sees Pydantic models.

from sqlalchemy import select
from database import Department, Employee

def create_employee(session, data: EmployeeCreate) -> EmployeeRead:
    if session.get(Department, data.department_id) is None:
        raise ValueError(f"Department {data.department_id} does not exist")

    employee = Employee(**data.model_dump())      # schema -> ORM object
    session.add(employee)
    session.commit()                              # INSERT; SQL Server gives the new id
    return EmployeeRead.model_validate(employee)  # ORM object -> schema


def get_employee(session, employee_id: int) -> EmployeeRead | None:
    employee = session.get(Employee, employee_id)
    return EmployeeRead.model_validate(employee) if employee else None


def list_employees(session) -> list[EmployeeRead]:
    rows = session.scalars(select(Employee).order_by(Employee.id)).all()
    return [EmployeeRead.model_validate(row) for row in rows]


def update_employee(session, employee_id: int, changes: EmployeeUpdate) -> EmployeeRead | None:
    employee = session.get(Employee, employee_id)
    if employee is None:
        return None

    # exclude_unset=True -> only the fields the user really sent.
    # Without it every missing field would be None and overwrite the data!
    for field_name, value in changes.model_dump(exclude_unset=True).items():
        setattr(employee, field_name, value)

    session.commit()                              # UPDATE
    session.refresh(employee)
    return EmployeeRead.model_validate(employee)


def delete_employee(session, employee_id: int) -> bool:
    employee = session.get(Employee, employee_id)
    if employee is None:
        return False
    session.delete(employee)
    session.commit()                              # DELETE
    return True


def get_department(session, department_id: int) -> DepartmentRead | None:
    department = session.get(Department, department_id)
    return DepartmentRead.model_validate(department) if department else None

• Employee(**data.model_dump()): the Create schema's fields match the table columns, so one line creates the ORM object.
• session.get(Employee, id) finds a row by primary key, or returns None.
• create_employee first checks that the department exists, so the user gets a clear message instead of a foreign-key error from SQL Server.
• update_employee uses setattr(object, name, value) to set each changed field by name. The key part is exclude_unset=True; see B4 below.

Using the functions

The lesson first prepares the database and removes the employee it adds (in case an earlier run stopped half-way), then runs every CRUD step inside one session:

from sqlalchemy import delete
from database import SessionLocal, add_sample_data, create_database_and_tables

create_database_and_tables()
add_sample_data()

NEW_EMAIL = "sneha.kulkarni@example.com"

with SessionLocal() as session:
    session.execute(delete(Employee).where(Employee.email == NEW_EMAIL))
    session.commit()
    ...   # steps B1 to B6 below

The outputs below use the sample data from database.py. The id of the new employee (6 here) will be higher if you have run the lesson before, because SQL Server never reuses identity values.

B1. READ: every employee with a nested department

for e in list_employees(session):
    print(f"   {e.id:>3}  {e.first_name:<8} {e.department.name:<8} active={e.is_active}")

     1  Asha     IT       active=True
     2  Ravi     IT       active=True
     3  Neha     HR       active=True
     4  Kiran    Finance  active=True
     5  Meena    HR       active=False

e.department.name comes from the nested DepartmentSummary, which Pydantic filled from the SQLAlchemy relationship.

B2. CREATE: JSON from a user, then EmployeeCreate, then SQL Server

request_json = f"""{{
    "first_name": "Sneha", "last_name": "Kulkarni", "email": "{NEW_EMAIL}",
    "salary": "88000", "hire_date": "2026-10-01", "department_id": 1
}}"""
new_data = EmployeeCreate.model_validate_json(request_json)
created = create_employee(session, new_data)
print(created.model_dump_json(indent=2))

{
  "first_name": "Sneha",
  "last_name": "Kulkarni",
  "email": "sneha.kulkarni@example.com",
  "salary": "88000",
  "hire_date": "2026-10-01",
  "id": 6,
  "is_active": true,
  "department": {
    "id": 1,
    "name": "IT"
  }
}

This is the full journey of real API data: JSON text was validated into EmployeeCreate, turned into an ORM object, inserted into SQL Server (which created the id), and returned as EmployeeRead with the nested department. (The {{ and }} in the f-string produce single braces in the JSON text.)

B3. CREATE with bad data: stopped before it reaches the database

try:
    EmployeeCreate.model_validate({"first_name": "S", "last_name": "", "email": "not-an-email",
                                   "salary": -5, "hire_date": "yesterday"})
except ValidationError as error:
    for item in error.errors():
        print(f"   {item['loc']}: {item['msg']}")

   ('first_name',): String should have at least 2 characters
   ('last_name',): String should have at least 1 character
   ('email',): value is not a valid email address: An email address must have an @-sign.
   ('salary',): Input should be greater than 0
   ('hire_date',): Input should be a valid date or datetime, input is too short
   ('department_id',): Field required

Six problems found, and not a single SQL statement was sent. Validating at the entrance protects your database from bad data.

B4. UPDATE: send only what changes

changes = EmployeeUpdate(salary="92000", department_id=2)
print("Fields sent      :", changes.model_dump(exclude_unset=True))
print("Without exclude  :", changes.model_dump())             # see all the None values
updated = update_employee(session, created.id, changes)
print("After update     :", updated.salary, updated.department.name)

Fields sent      : {'salary': Decimal('92000'), 'department_id': 2}
Without exclude  : {'first_name': None, 'last_name': None, 'email': None, 'salary': Decimal('92000'), 'is_active': None, 'department_id': 2}
After update     : 92000.00 HR

This is the most important lesson of the whole series. Look at the "Without exclude" line: if update_employee used plain model_dump(), it would set first_name, email and the others to None, wiping out real data (or failing because the columns don't allow NULL). With exclude_unset=True, only salary and department_id are updated, exactly what the user asked for.

B5. READ one department with its employees (nested list)

print(get_department(session, 1).model_dump_json(indent=2))

{
  "id": 1,
  "name": "IT",
  "location": "Pune",
  "employees": [
    {
      "id": 1,
      "first_name": "Asha",
      "last_name": "Rao"
    },
    {
      "id": 2,
      "first_name": "Ravi",
      "last_name": "Kumar"
    }
  ]
}

Sneha is not in IT any more, because B4 moved her to department 2 (HR). The nested list was built from the department.employees relationship, using the small EmployeeSummary model, so no salaries are shown.

B6. DELETE

print("Deleted:", delete_employee(session, created.id))
print("Find it again:", get_employee(session, created.id))

Deleted: True
Find it again: None

How this maps to FastAPI

The functions above are exactly what FastAPI endpoints call. In FastAPI the same schemas are used like this:

@app.post("/employees", response_model=EmployeeRead)
def add(data: EmployeeCreate): ...            # FastAPI validates the request body with EmployeeCreate

@app.patch("/employees/{employee_id}", response_model=EmployeeRead)
def change(employee_id: int, changes: EmployeeUpdate): ...

See it in a full project in Basic CRUD in FastAPI.

Exercises

From the end of 06_nested_and_dto_schemas.py:

1. Add DepartmentCreate(name, location) and a create_department() function.
2. Add a DepartmentUpdate schema and an update_department() function.
3. Make EmployeeRead hide salary for a "public" list: create EmployeePublicRead.
4. Add a computed_field full_name to EmployeeRead.
5. Make EmployeeUpdate refuse an empty update (a model_validator that checks that at least one field was sent; hint: self.model_fields_set).

Summary

• Use a model as a field type to nest models; lists of models work too. Pydantic builds and validates every level.
• Nested error locations are paths like ('items', 1, 'quantity').
• One table usually needs three schemas: Create (no id), Update (all optional) and Read (with id and nested data).
• A shared base class keeps each rule in one place.
• CRUD functions take a schema in and give a schema out; ORM objects stay inside.
• Always use model_dump(exclude_unset=True) for updates, so fields the user didn't send are not overwritten.

Next: project settings with pydantic-settings and a .env file.