FastAPI Project Structure & Database CRUD

How a real FastAPI project is split into files, connects to a database with SQLAlchemy, and exposes full CRUD endpoints.

What is it?

A single main.py is fine for a two-route demo, but a real FastAPI project splits responsibilities across files, much like a layered Node.js app: database.py sets up the database connection, models.py defines the actual database tables as Python classes, schemas.py defines the Pydantic models describing what the API accepts and returns, and one or more router files group related endpoints, all wired together in main.py.

SQLAlchemy is Python's most common tool for talking to a relational database. Used here as an ORM, it maps each database row to a Python object — the same idea as Prisma or Sequelize in Node, just in Python.

Explain like I'm 10

It's the same layered kitchen from backend project structure, just relabeled for a Python kitchen: routers are the host seating guests, path functions are the server taking the order, and SQLAlchemy models are the pantry holding the actual ingredients.

Examples

Project layout and the database connection

myapp/
├── main.py           # creates the app, includes routers
├── database.py        # engine + session setup
├── models.py           # SQLAlchemy table definitions
├── schemas.py          # Pydantic request/response shapes
└── routers/
    └── items.py         # /items endpoints

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql://user:password@localhost/mydb")
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

get_db is a FastAPI dependency: because it uses yield, FastAPI runs everything before the yield, hands the route the session, then runs everything after the yield (closing it) once the request finishes — even if the route raised an error.

Full CRUD endpoints using that session

# routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from .. import models, schemas
from ..database import get_db

router = APIRouter()

@router.post("/items", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@router.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.put("/items/{item_id}", response_model=schemas.Item)
def update_item(item_id: int, updated: schemas.ItemCreate, db: Session = Depends(get_db)):
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    for key, value in updated.dict().items():
        setattr(item, key, value)
    db.commit()
    return item

@router.delete("/items/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db.query(models.Item).filter(models.Item.id == item_id).delete()
    db.commit()
    return {"deleted": item_id}

The same four CRUD operations from SQL — Create, Read, Update, Delete — expressed through SQLAlchemy's query API instead of raw SQL text, each one wired to its own route.

Wrapping multiple changes in a transaction

@router.post("/transfer")
def transfer(from_id: int, to_id: int, amount: int, db: Session = Depends(get_db)):
    try:
        sender = db.query(models.Account).filter(models.Account.id == from_id).one()
        receiver = db.query(models.Account).filter(models.Account.id == to_id).one()
        sender.balance -= amount
        receiver.balance += amount
        db.commit()
    except Exception:
        db.rollback()
        raise
    return {"status": "ok"}

Both balance changes need to succeed or fail together — nothing is actually written until db.commit() runs, and if anything raises before that, db.rollback() discards both pending changes so the accounts are never left half-updated.

How it works

db: Session = Depends(get_db) is FastAPI's dependency injection reusing get_db for every request — each request gets its own fresh session, used only for that request's queries, then closed. Calls like db.add, db.commit, db.query, and .delete() map to INSERT, UPDATE/COMMIT, SELECT, and DELETE statements that SQLAlchemy generates and sends over the underlying database connection, the same way the pg driver does in Node. To actually run the app, you start an ASGI server pointed at it: uvicorn main:app --reload — the Python equivalent of running node server.js, watching for changes with --reload during development.

Why does it exist?

Splitting a FastAPI project this way mirrors exactly why a Node backend gets split into routes/controllers/services/models: as endpoints and tables multiply, keeping the app's setup, data shape, and database logic each in one dedicated place keeps the project navigable, instead of tangled into a single growing file.

When to use it

Reach for this structure once a FastAPI project has more than a couple of endpoints or more than one table — the same threshold as reaching for a layered structure in a Node project.

When not to use it

For a two-endpoint script or a quick prototype, separate database.py/models.py/schemas.py/router files are more scaffolding than the project needs — one main.py is easier to follow at that size.

Common mistakes

  • Forgetting db.commit() after db.add() or a mutation, leaving the change only staged in the session instead of actually written to the database.

  • Reusing one global session across every request instead of a fresh one per request via Depends(get_db), which can leak stale data or connections between unrelated requests.

  • Returning a raw SQLAlchemy model object instead of going through a Pydantic response_model, exposing internal fields (like a password hash) that were never meant to reach the client.

Practice exercises

  1. Easy:

    Name which file — database.py, models.py, schemas.py, or a router — a new 'orders' table definition belongs in.

  2. Medium:

    Write a DELETE /items/{item_id} endpoint that returns a 404 if the item doesn't exist before deleting it.

  3. Hard:

    Explain what would go wrong if a single database session, created once at app startup, were reused across every incoming request instead of one session per request.

Interview questions

What does Depends(get_db) provide to a FastAPI route?

A fresh, request-scoped SQLAlchemy session that's automatically closed afterward — similar to Express middleware attaching something onto the request object.

What command actually runs a FastAPI app?

uvicorn main:app --reload — an ASGI server pointed at the created FastAPI instance, with --reload restarting it on code changes during development.

Why use a Pydantic response_model instead of returning the raw database object?

It controls exactly which fields are exposed to the client, preventing internal-only fields from leaking into the API response.

Why is `get_db` written as a generator function using `yield` instead of just returning a session directly?

FastAPI recognizes a dependency using yield as needing cleanup — it runs everything before yield to produce the session, hands it to the route, and once the route finishes, successfully or by raising, resumes the generator to run the code after yield, guaranteeing db.close() always executes exactly once per request.

What would happen if `db.close()` were forgotten inside get_db's finally block?

The session, and its underlying database connection, would never be released back to the pool for that request, and since get_db runs fresh per request, this leak would compound with every request — eventually exhausting the connection pool, the same way forgetting client.release() does in a raw Node driver.

In the create_item example, why is `db.refresh(db_item)` called after `db.commit()`?

After committing, the in-memory db_item object may not reflect values the database itself generated or defaulted during the insert, like an auto-incrementing id; db.refresh() re-fetches the row from the database into that same object so the returned response actually includes those generated fields.

What happens if `db.commit()` is never called after `db.add()` or after mutating an already-loaded object's attributes?

The change stays only staged in the SQLAlchemy session's in-memory state and is never actually written to the database — the request may appear to succeed since no error is raised, but the data silently isn't persisted, which can be a confusing bug to track down.

Why does update_item use `setattr(item, key, value)` in a loop instead of just assigning `item = updated`?

item is a SQLAlchemy-tracked model instance already associated with a specific row in the session; replacing it entirely with updated, a plain Pydantic object, would lose that tracking and prevent SQLAlchemy from knowing what changed — updating attributes individually on the already-tracked object lets SQLAlchemy detect exactly which fields changed and generate the right UPDATE statement.

What's the purpose of the response_model argument in a route decorator, like `@router.post("/items", response_model=schemas.Item)`?

It tells FastAPI to validate and shape whatever the route function returns against that Pydantic model before sending the response, filtering out any field not declared on it — even if the route accidentally returns a raw SQLAlchemy object with extra internal fields, only the declared fields reach the client.

Why does the example define separate schemas.Item and schemas.ItemCreate models, rather than one shared model for both input and output?

Input and output often legitimately need different shapes — creating an item might not require or allow a client-supplied id, while reading one back needs to include the database-generated id; separate schemas let each direction declare exactly the fields appropriate to it, rather than forcing one model to awkwardly cover both.

What SQL statement does `db.query(models.Item).filter(models.Item.id == item_id).first()` correspond to?

Roughly SELECT * FROM items WHERE id = :item_id LIMIT 1 — SQLAlchemy's query API builds that SQL from the chained method calls, and .first() specifically limits the result to at most one row, returning None if nothing matches rather than raising.

Why does read_item explicitly check `if not item: raise HTTPException(status_code=404, ...)` instead of just returning whatever .first() produced?

.first() returns None when no row matches rather than raising an error, so without the explicit check, a request for a nonexistent item would fall through and try to return None through the response_model, producing a confusing validation error instead of a clear, intentional 404.

What's the SQLAlchemy equivalent of the raw-SQL transaction pattern of BEGIN, commit, and ROLLBACK shown for the money-transfer example?

Multiple changes made on tracked model objects within the same session, like both sender.balance and receiver.balance, stay pending until one db.commit() call writes them together; db.rollback() inside an except block discards every pending change in that session if anything fails first, mirroring BEGIN/COMMIT/ROLLBACK without writing that SQL directly.

Why is a fresh Session created per request via Depends(get_db), rather than one shared session reused across the whole app's lifetime?

A shared, long-lived session accumulates every object it's ever loaded or changed in its identity map, can leak state between unrelated requests, and isn't safe to use concurrently from multiple requests at once — a fresh session per request keeps each request's unit of work isolated and safely scoped.

What does `uvicorn main:app --reload` mean, piece by piece?

uvicorn is the ASGI server actually running the app; main:app tells it to import the app object from the main module; --reload makes it watch source files and automatically restart the server whenever code changes, meant only for development, not production.

Why does splitting a FastAPI project into database.py, models.py, schemas.py, and router files mirror the layered structure of a typical Node backend?

Each file isolates one concern the same way a Node app's routes/controllers/services/models split does — database.py parallels a db config module, models.py parallels an ORM's model files, schemas.py plays a role similar to validation schemas, and routers parallel Express route files.

What's the risk of returning a raw SQLAlchemy model instance directly from a route with no response_model declared at all?

FastAPI would try to serialize whatever attributes exist on that ORM object to JSON, potentially exposing every column, including ones never meant to reach a client, like a password_hash or internal audit fields, since there's no schema constraining which fields are allowed out.

Why does the delete_item example return `{"deleted": item_id}` rather than a 204 No Content response?

Returning a JSON body confirming which id was deleted gives the caller explicit confirmation of what happened, which can be more convenient than a body-less 204; either is defensible — what matters is picking one convention for delete endpoints and using it consistently rather than mixing both arbitrarily.