A Database with SQLAlchemy and SQLite
35 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)A list in memory disappears when the server stops. A database keeps the data. SQLite is a database stored in one file, which needs no setup. SQLAlchemy lets you work with its tables using Python classes, so the SQL you learned stays behind the scenes.
A class for each table
A model is a class that maps to one database table. Its attributes are the columns, and each object of the class is one row. SQLAlchemy turns your Python calls into SQL for you.
A model is a Python class that stands for one table. Each attribute is a column. create_all builds the tables in the database if they are missing, and SQLite makes the file by itself.
class Base(DeclarativeBase): pass
class Ticket(Base): __tablename__ = "tickets"
id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(100)) status: Mapped[str] = mapped_column(String(20), default="open")
engine = create_engine("sqlite:///tickets.db")Base.metadata.create_all(engine)The imports are from sqlalchemy import String, create_engine and from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column. The engine holds the connection to the file. Use sqlite:///:memory: for a throwaway database in tests.
A session adds, finds, changes and deletes
A session is a short-lived workspace for talking to the database. It remembers what you added or changed and saves everything together on commit(). Close it when the work is done.
You never touch rows directly. You open a session, make changes and call commit() to save them all together. select(...) builds a query and the session runs it.
with Session(engine) as session: session.add(Ticket(title="Login page broken")) session.commit()
query = select(Ticket).where(Ticket.status == "open") for ticket in session.scalars(query): print(ticket.title)
ticket = session.get(Ticket, 1) ticket.status = "closed" session.commit() session.delete(ticket)Example 1 · Insert and select
from sqlalchemy import String, create_engine, selectfrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
class Base(DeclarativeBase): pass
class Ticket(Base): __tablename__ = "tickets"
id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(100)) status: Mapped[str] = mapped_column(String(20), default="open")
engine = create_engine("sqlite:///:memory:")Base.metadata.create_all(engine)
with Session(engine) as session: session.add(Ticket(title="Login page broken")) session.add(Ticket(title="Add dark mode", status="closed")) session.commit()
query = select(Ticket).where(Ticket.status == "open") for ticket in session.scalars(query): print(ticket.id, ticket.title, ticket.status) print(len(session.scalars(select(Ticket)).all()))1 Login page broken open 2
What this code does
addqueues each ticket andcommitsaves them. The ids 1 and 2 come from the database, not from your code.where(Ticket.status == "open")keeps only the open ticket. The closed one is left out.session.scalars(select(Ticket)).all()reads every row into a list; here that is 2.
Example 2 · Update and delete
from sqlalchemy import String, create_engine, selectfrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
class Base(DeclarativeBase): pass
class Ticket(Base): __tablename__ = "tickets"
id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(100)) status: Mapped[str] = mapped_column(String(20), default="open")
engine = create_engine("sqlite:///:memory:")Base.metadata.create_all(engine)
with Session(engine) as session: session.add_all([Ticket(title="Login page broken"), Ticket(title="Add dark mode")]) session.commit()
ticket = session.get(Ticket, 2) ticket.status = "closed" session.commit() print(session.get(Ticket, 2).status)
session.delete(session.get(Ticket, 1)) session.commit() print(session.get(Ticket, 1)) print(session.scalars(select(Ticket.title)).all())closed None ['Add dark mode']
What this code does
- Assigning to
ticket.statusis enough. The session writes the UPDATE oncommit(). - After the delete,
session.get(Ticket, 1)returnsNone, which your route can turn into a 404. select(Ticket.title)asks for one column, so you get plain text values instead of objects.
session.get(Model, id) finds one row by its primary key and gives None when it is missing. Changing an attribute and calling commit() runs an UPDATE. Leaving the with block closes the session.
One user, many tickets
A foreign key is a column that holds the primary key of a row in another table, such as owner_id pointing at users.id. The database uses it to keep tickets linked to real users.
Real data is linked. A ticket belongs to a user, so the ticket table stores the user id as a foreign key. A relationship lets you move between the two as ordinary attributes.
class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) tickets: Mapped[list["Ticket"]] = relationship(back_populates="owner")
class Ticket(Base): __tablename__ = "tickets" id: Mapped[int] = mapped_column(primary_key=True) owner_id: Mapped[int] = mapped_column(ForeignKey("users.id")) owner: Mapped["User"] = relationship(back_populates="tickets")Example 3 · Linked tables
from sqlalchemy import ForeignKey, String, create_engine, selectfrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship
class Base(DeclarativeBase): pass
class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) tickets: Mapped[list["Ticket"]] = relationship(back_populates="owner")
class Ticket(Base): __tablename__ = "tickets" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(100)) owner_id: Mapped[int] = mapped_column(ForeignKey("users.id")) owner: Mapped["User"] = relationship(back_populates="tickets")
engine = create_engine("sqlite:///:memory:")Base.metadata.create_all(engine)
with Session(engine) as session: asha = User(name="Asha", tickets=[Ticket(title="Login page broken"), Ticket(title="Add dark mode")]) ravi = User(name="Ravi") session.add_all([asha, ravi]) session.commit()
for user in session.scalars(select(User)): print(user.name, [t.title for t in user.tickets]) print(session.get(Ticket, 2).owner.name)Asha ['Login page broken', 'Add dark mode'] Ravi [] Asha
What this code does
- Passing
tickets=[...]when you create the user also fillsowner_idfor each ticket. You never write the id yourself. user.ticketsis a list, and Ravi has none, so his list is empty.ticket.ownergoes the other way: from a ticket to its user.
user.tickets is a list of that user's tickets, and ticket.owner is the one user. back_populates keeps the two sides in step. Import ForeignKey and relationship as well.
One session for each request
A dependency is a function that FastAPI runs before your route and whose result it passes in as an argument. Depends(get_db) gives each request its own database session and closes it afterwards.
In FastAPI every request gets its own session, which is closed when the request ends. A small function with yield does this, and Depends hands the session to any route that asks for it.
SessionLocal = sessionmaker(engine)
def get_db(): with SessionLocal() as db: yield db
@app.get("/tickets/{ticket_id}")def get_ticket(ticket_id: int, db: Session = Depends(get_db)): ticket = db.get(Ticket, ticket_id) if ticket is None: raise HTTPException(status_code=404, detail="Ticket not found") return ticketExample 4 · A FastAPI API on a database
from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, ConfigDict, Fieldfrom sqlalchemy import String, create_engine, selectfrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmakerfrom sqlalchemy.pool import StaticPool
class Base(DeclarativeBase): pass
class Ticket(Base): __tablename__ = "tickets" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(100)) status: Mapped[str] = mapped_column(String(20), default="open")
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SessionLocal = sessionmaker(engine)Base.metadata.create_all(engine)
def get_db(): with SessionLocal() as db: yield db
class TicketIn(BaseModel): title: str = Field(min_length=3, max_length=100)
class TicketOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int title: str status: str
app = FastAPI()
@app.post("/tickets", response_model=TicketOut, status_code=201)def create_ticket(data: TicketIn, db: Session = Depends(get_db)): ticket = Ticket(title=data.title) db.add(ticket) db.commit() db.refresh(ticket) return ticket
@app.get("/tickets", response_model=list[TicketOut])def list_tickets(skip: int = 0, limit: int = 50, db: Session = Depends(get_db)): return db.scalars(select(Ticket).offset(skip).limit(limit)).all()
@app.get("/tickets/{ticket_id}", response_model=TicketOut)def get_ticket(ticket_id: int, db: Session = Depends(get_db)): ticket = db.get(Ticket, ticket_id) if ticket is None: raise HTTPException(status_code=404, detail="Ticket not found") return ticket
client = TestClient(app)print(client.post("/tickets", json={"title": "Login page broken"}).json())client.post("/tickets", json={"title": "Add dark mode"})print([t["title"] for t in client.get("/tickets?limit=1").json()])print(client.get("/tickets/5").status_code){'id': 1, 'title': 'Login page broken', 'status': 'open'}
['Login page broken']
404What this code does
response_model=TicketOutwithfrom_attributes=Trueturns the database object into JSON with only the listed fields.skipandlimitreach the query through.offset(skip).limit(limit).- The test engine uses
StaticPool, so every session sees the same in-memory database.
Never share one global session between requests. For lists, always accept skip and limit and pass them to .offset(skip).limit(limit), so a table with a million rows cannot flood one response.
Next step
Check your understanding
10 quick questions