Week 10: FastAPI and Pydantic, CRUD, and Databases (SQLite + SQLModel)

Month 3 | Main goal: The student builds a complete CRUD API — first in memory, then in a SQLite database with SQLModel — using Pydantic, HTTPException, Depends, and routers.

Teacher Preparation (read before the week starts)

This week is heavier than the last — make sure you have written and run the code for each session at least once.

1. Read:
   • FastAPI — "Request Body": https://fastapi.tiangolo.com/tutorial/body/
   • FastAPI — "Response Model - Return Type": https://fastapi.tiangolo.com/tutorial/response-model/
   • FastAPI — "Handling Errors": https://fastapi.tiangolo.com/tutorial/handling-errors/
   • FastAPI — "SQL (Relational) Databases": https://fastapi.tiangolo.com/tutorial/sql-databases/ (this is the basis of Session 2 — read it carefully)
   • FastAPI — "Dependencies" (the first page): https://fastapi.tiangolo.com/tutorial/dependencies/
   • FastAPI — "Bigger Applications - Multiple Files": https://fastapi.tiangolo.com/tutorial/bigger-applications/

2. Install (in the venv of this week's project):

   powershell

   Copy

   pip install "fastapi[standard]" sqlmodel
   

   SQLite doesn't need installing — it is built into Python (import sqlite3).

3. VS Code extension (recommended): "SQLite Viewer" (author: Florian Klampfer). It lets you click the database.db file and see the tables like a spreadsheet. It is very good for teaching.

4. Understand these terms before class: model, schema, request body, validation, serialization, CRUD, table, row, column, primary key, engine, session, commit, query, dependency injection, router.
5. Remember the end of Week 9: GET /bidhaa/99 returned 200 with {"kosa": ...}. Today we fix it with HTTPException — start the session by reminding the student.

Week Goals (by the end of the week the student will be able to...)

1. Write Pydantic models (BaseModel) and use them as the request body.
2. Set validation rules with Field (min_length, ge, le...) and read 422 errors.
3. Use response_model / a return type to filter the data returned (for example hiding a password).
4. Build complete CRUD: POST, GET (list & one), PUT, PATCH, DELETE, with the correct status codes (201, 204, 404).
5. Use HTTPException to return errors.
6. Explain what a database is; create a SQLite table with SQLModel; do CRUD with Session.
7. Use dependency injection (Depends, Annotated) for the session and other shared things.
8. Split a project into files: main.py, database.py, models.py, routers/.

Session 1 — Pydantic, Request Body, and In-Memory CRUD

Session flow

1. Homework review (the Library API) — the student runs it and shows /docs
2. Pydantic BaseModel outside FastAPI: validation and model_dump()
3. Request body on POST; Field validation; reading a 422
4. response_model: hiding a password
5. Complete in-memory CRUD for Tasks + HTTPException + status codes
6. In-class exercise
7. Q&A + homework

Explanation for the teacher (read this first)

Pydantic is a library (it comes with FastAPI) that works like a dataclass "with a guard". The student knows @dataclass (Week 7) — Pydantic's BaseModel is similar, but it validates and converts types when the object is created:

from pydantic import BaseModel
class Mtu(BaseModel):
    jina: str
    umri: int
Mtu(jina="Ali", umri="21")    # umri inakuwa 21 (int) — imegeuzwa
Mtu(jina="Ali", umri="mbili") # ValidationError

A dataclass wouldn't check anything.

Request body. In FastAPI, if you give an argument a Pydantic model as its type, FastAPI understands: "this data comes in the body of the request as JSON". It reads the JSON → creates the model → validates it. If that fails → 422 with details of every error.

The rule FastAPI uses to decide where an argument comes from:

• The name is in the path ({task_id}) → path parameter
• The type is simple (int, str, bool...) → query parameter
• The type is a Pydantic model → body

Field(...) adds rules: min_length, max_length (for str), gt, ge, lt, le (for numbers: greater than, greater-or-equal, less than, less-or-equal), default.

response_model (or the return type annotation -> Model) tells FastAPI: "the outgoing data should have this shape". FastAPI filters out fields that aren't in it. This is security: even if you return an object with a password, if the response_model has no password, it won't go out.

CRUD = Create, Read, Update, Delete — the four basic operations on any data:

HTTPException: you throw it (raise) inside an endpoint; FastAPI catches it and returns the status and {"detail": "..."}. It is raise, not return, because it stops the function on the spot (like any exception — Week 5).

PATCH and exclude_unset=True: For PATCH we use a model in which all fields are optional (None). But there is a problem: how do we tell apart "the user didn't send description" from "the user deliberately sent description: null"? model_dump(exclude_unset=True) returns only the fields that were sent. Then model_copy(update=...) creates a changed copy.

"In-memory" means the data lives in a Python dict — if you stop the server (or it restarts after you save a file!), the data is lost. That is the reason for the database in Session 2.

Say this to the student

"A Pydantic model is like a bank registration form. The form says: name (required), age (a number), phone (optional). The clerk (Pydantic) checks it before it goes inside. If you write 'age: ishirini' (twenty, as a word), the clerk hands the form back and tells you where the mistake is — that is the 422."

"response_model is like the bank window: inside there is all your information, even your PIN, but through the window you are only given your balance. The PIN doesn't come out."

"CRUD is an attendance book: writing a new name (Create), reading the list (Read), correcting it (Update), crossing it out (Delete). Almost every app in the world — WhatsApp, M-Pesa, Instagram — is CRUD over different things."

"raise HTTPException(404) is like the waiter saying 'we don't have it' and walking away — they don't carry on with the order."

Type this code live (share screen)

Step 1 — New folder:

cd $HOME\Documents
mkdir wiki10
cd wiki10
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install "fastapi[standard]" sqlmodel
code .

Step 2 — Pydantic on its own. Create pydantic_demo.py:

from pydantic import BaseModel, Field, ValidationError


class Mwanafunzi(BaseModel):
    jina: str = Field(min_length=2)
    umri: int = Field(ge=5, le=100)
    simu: str | None = None


m = Mwanafunzi(jina="Amina", umri="21")
print(m)
print(type(m.umri))
print(m.model_dump())
print(m.model_dump_json())

try:
    Mwanafunzi(jina="A", umri=200)
except ValidationError as e:
    print(e)

Output:

jina='Amina' umri=21 simu=None
<class 'int'>
{'jina': 'Amina', 'umri': 21, 'simu': None}
{"jina":"Amina","umri":21,"simu":null}
2 validation errors for Mwanafunzi
jina
  String should have at least 2 characters [type=string_too_short, input_value='A', input_type=str]
    For further information visit https://errors.pydantic.dev/2.x/v/string_too_short
umri
  Input should be less than or equal to 100 [type=less_than_equal, input_value=200, input_type=int]
    For further information visit https://errors.pydantic.dev/2.x/v/less_than_equal

(The version number in the link may differ — that isn't a problem.)

Emphasise: "21" was converted to 21. ALL the errors are reported together, not just the first.

Step 3 — Request body and response_model. Create main.py:

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="Tasks API (memory)")


class UserIn(BaseModel):
    username: str = Field(min_length=3)
    password: str = Field(min_length=6)
    email: str | None = None


class UserOut(BaseModel):
    username: str
    email: str | None = None


@app.post("/users", response_model=UserOut)
def sajili(user: UserIn):
    # Tunarudisha object YENYE password...
    return user

fastapi dev main.py → /docs → POST /users → Try it out → body:

{"username": "juma", "password": "siri123", "email": "juma@mail.co.tz"}

Response:

{"username": "juma", "email": "juma@mail.co.tz"}

Say: "We returned a user with a password, but response_model=UserOut filtered it out." Then try the body {"username": "ju", "password": "123"} → 422 with two string_too_short errors.

Also show Schemas at the bottom of /docs — UserIn and UserOut appeared on their own.

Step 4 — Complete in-memory CRUD. Change main.py to this (delete the users code):

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(title="Tasks API (memory)")


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    description: str | None = None
    priority: int = Field(default=3, ge=1, le=5)
    done: bool = False


class Task(TaskCreate):
    id: int


class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    done: bool | None = None


tasks: dict[int, Task] = {}
next_id = 1


def tafuta_au_404(task_id: int) -> Task:
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail=f"Task {task_id} haipo")
    return tasks[task_id]


@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate):
    global next_id
    task = Task(id=next_id, **task_in.model_dump())
    tasks[task.id] = task
    next_id += 1
    return task


@app.get("/tasks", response_model=list[Task])
def list_tasks(done: bool | None = None):
    matokeo = list(tasks.values())
    if done is not None:
        matokeo = [t for t in matokeo if t.done == done]
    return matokeo


@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: int):
    return tafuta_au_404(task_id)


@app.put("/tasks/{task_id}", response_model=Task)
def replace_task(task_id: int, task_in: TaskCreate):
    tafuta_au_404(task_id)
    task = Task(id=task_id, **task_in.model_dump())
    tasks[task_id] = task
    return task


@app.patch("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, task_in: TaskUpdate):
    zamani = tafuta_au_404(task_id)
    mabadiliko = task_in.model_dump(exclude_unset=True)
    mpya = zamani.model_copy(update=mabadiliko)
    tasks[task_id] = mpya
    return mpya


@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int):
    tafuta_au_404(task_id)
    del tasks[task_id]

Try them in this order in /docs (show the status code each time):

1. POST /tasks body {"title": "Kusoma Pydantic", "priority": 1} → 201, {"title":"Kusoma Pydantic","description":null,"priority":1,"done":false,"id":1}
2. POST /tasks body {"title": "Kununua vocha"} → 201, id 2, priority 3 (default)
3. GET /tasks → a list of 2
4. GET /tasks/5 → 404 {"detail":"Task 5 haipo"}
5. PATCH /tasks/1 body {"done": true} → title unchanged, done: true
6. GET /tasks?done=true → task 1 only
7. PUT /tasks/2 body {"title": "Kununua vocha ya 5000"} → priority back to 3, description null (PUT replaces the whole thing)
8. DELETE /tasks/2 → 204, no body
9. GET /tasks/2 → 404
10. POST /tasks body {"title": "", "priority": 9} → 422

Then save the file (Ctrl+S) without changing anything, and do GET /tasks → []! Say: "The server restarted and the memory was wiped. That is why we need a database — Session 2."

Step 5 — Understanding exclude_unset. Temporarily add the line print(task_in, "|", mabadiliko) to update_task then PATCH {"done": true}. Terminal:

title=None description=None priority=None done=True | {'done': True}

Say: "Without exclude_unset, we would write None over the title — we would destroy data."

Questions to ask the student

1. How does FastAPI know that task_in: TaskCreate comes from the body? — Because its type is a Pydantic model.
2. And done: bool | None = None in list_tasks? — It is a simple type and isn't in the path → query parameter.
3. Why do we use raise rather than return for HTTPException? — It is an exception; it stops the function on the spot and FastAPI turns it into an error response.
4. Why does DELETE return 204 with no body? — 204 = "No Content": it succeeded, and there is nothing to return.
5. What is the difference between PUT and PATCH in our code? — PUT receives TaskCreate (the whole thing; fields not sent go back to their default); PATCH receives TaskUpdate and changes only the fields sent.
6. Why do we have three models (TaskCreate, Task, TaskUpdate) instead of one? — Each has a job: the client shouldn't send an id (the server issues it), Task has the id for the response, TaskUpdate has all fields optional for PATCH.
7. What does ge=1, le=5 mean? — Greater-or-equal 1, less-or-equal 5: a value between 1 and 5.
8. Why was the data lost when we saved the file? — fastapi dev restarted; the dict is in the process's memory.

Common mistakes

In-class exercise

Exercise 1.1 — Add a field due_date: date | None = None to TaskCreate and TaskUpdate (from datetime import date). Try sending "2027-01-20" and "jana" (yesterday). What happens?

Exercise 1.2 — Add an endpoint GET /tasks/stats that returns {"jumla": n, "zilizokamilika": k, "zilizobaki": n - k} (total, completed, remaining). (Remember route order!)

Exercise 1.3 — Add a query parameter q: str | None = None to GET /tasks that filters tasks whose title contains the word q (ignoring upper/lower case).

Exercise 1.4 — In POST /tasks, prevent two tasks with the same title: if it already exists, return 409 Conflict with the detail "Task yenye title hii ipo tayari" (a task with this title already exists).

from datetime import date

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(title="Tasks API (memory)")


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    description: str | None = None
    priority: int = Field(default=3, ge=1, le=5)
    done: bool = False
    due_date: date | None = None          # 1.1


class Task(TaskCreate):
    id: int


class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    done: bool | None = None
    due_date: date | None = None          # 1.1


tasks: dict[int, Task] = {}
next_id = 1


def tafuta_au_404(task_id: int) -> Task:
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail=f"Task {task_id} haipo")
    return tasks[task_id]


@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate):
    global next_id
    # 1.4
    if any(t.title.lower() == task_in.title.lower() for t in tasks.values()):
        raise HTTPException(status_code=409, detail="Task yenye title hii ipo tayari")
    task = Task(id=next_id, **task_in.model_dump())
    tasks[task.id] = task
    next_id += 1
    return task


@app.get("/tasks", response_model=list[Task])
def list_tasks(done: bool | None = None, q: str | None = None):
    matokeo = list(tasks.values())
    if done is not None:
        matokeo = [t for t in matokeo if t.done == done]
    if q:                                   # 1.3
        matokeo = [t for t in matokeo if q.lower() in t.title.lower()]
    return matokeo


# 1.2 — LAZIMA iwe juu ya /tasks/{task_id}
@app.get("/tasks/stats")
def stats():
    n = len(tasks)
    k = sum(1 for t in tasks.values() if t.done)
    return {"jumla": n, "zilizokamilika": k, "zilizobaki": n - k}


@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: int):
    return tafuta_au_404(task_id)

# ... (PUT, PATCH, DELETE kama kwenye somo)

Session 2 — Databases: SQLite and SQLModel

Session flow

1. Homework review
2. What a database is: table, row, column, primary key; SQL in brief
3. SQLModel on its own (a script): engine, create table, insert, select, update, delete
4. Looking at database.db with SQLite Viewer
5. Connecting to FastAPI: multiple models, get_session, SessionDep, CRUD endpoints
6. In-class exercise
7. Q&A + homework

Explanation for the teacher (read this first)

A database is a special program for storing data permanently and finding it quickly. A relational database (such as SQLite, PostgreSQL, MySQL) stores data in tables:

• Table = like an Excel sheet (for example task)
• Column = a field (title, done...)
• Row = one record (one task)
• Primary key = the column that uniquely identifies a row (usually id); the database issues it itself (1, 2, 3...)

SQL is the language for talking to a database: SELECT * FROM task WHERE done = 1;. We won't write SQL by hand — SQLModel will write it for us.

SQLite is a database inside one file (database.db). It has no server and no password — it suits learning and small apps. Python comes with it.

SQLModel (written by the author of FastAPI) = Pydantic + SQLAlchemy. One class can be:

• A Pydantic model (validation, JSON), AND
• A database table — if you add table=True.

Four pieces to remember:

1. Table model: class Task(SQLModel, table=True) with id: int | None = Field(default=None, primary_key=True). id is None before saving; the database fills it in.
2. Engine: create_engine("sqlite:///database.db") — the "connection" to the database. Create ONE for the whole app.
3. SQLModel.metadata.create_all(engine) — creates all tables that don't exist yet (it doesn't delete existing ones, nor change the columns of an existing table — if you change a model while learning, delete database.db).
4. Session: with Session(engine) as session: — one "conversation" with the database. Inside it:
   • session.add(obj) — prepare to save
   • session.commit() — really save (without a commit, nothing is saved!)
   • session.refresh(obj) — re-read from the database (to get the new id)
   • session.get(Task, 5) — find by primary key (returns None if not found)
   • session.exec(select(Task).where(Task.done == True)).all() — a query
   • session.delete(obj) then commit()

Important — validation: A model with table=True does not validate when the object is created (on purpose, for SQLAlchemy reasons). That is why we use separate models for input (TaskCreate, without table=True) which do validate, and then convert them into the table Task. The arrangement recommended by the FastAPI docs:

TaskBase (fields za pamoja)
├── Task(TaskBase, table=True)  → table (ina id)
├── TaskCreate(TaskBase)        → body ya POST
├── TaskPublic(TaskBase)        → response (id: int — si None)
└── TaskUpdate(SQLModel)        → body ya PATCH (fields zote optional)

connect_args={"check_same_thread": False} — by default SQLite refuses to be used by more than one thread; FastAPI runs def endpoints on different threads. This allows it. We set it for SQLite only.

The get_session dependency: a function that uses yield (a generator — Week 8!):

def get_session():
    with Session(engine) as session:
        yield session

FastAPI calls it before the endpoint, hands the endpoint the session, and after the response it returns and finishes the with (the session is closed). SessionDep = Annotated[Session, Depends(get_session)] is just a short name so we don't have to write all that every time.

Lifespan: we want create_all to run once when the server starts. We use lifespan (the modern way; @app.on_event("startup") is old/deprecated).

Say this to the student

"A database is like an office filing cabinet. A table is one drawer (the 'tasks' drawer), each file inside is a row, and each file has its own unique number on top — that is the primary key."

"Memory (a dict) is like writing on the board — if someone wipes it (the server stops), it is gone. A database is like writing in an exercise book and locking it in the cabinet."

"A session is like going to the bank: you go in (with Session), you fill in the slip (add), but the money isn't deposited until the cashier stamps it (commit). If you leave without the stamp, nothing has happened."

"Depends(get_session) is like a hotel: you don't carry your own towel; when you reach the room the towel is ready, and when you leave they take it away. The endpoint is given a ready session, and FastAPI closes it afterwards."

Type this code live (share screen)

Step 1 — SQLModel on its own. Create db_demo.py:

from sqlmodel import Field, Session, SQLModel, create_engine, select


class Task(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str = Field(index=True)
    priority: int = 3
    done: bool = False


engine = create_engine("sqlite:///demo.db", echo=True)


def main() -> None:
    SQLModel.metadata.create_all(engine)

    # CREATE
    with Session(engine) as session:
        t1 = Task(title="Kufua nguo", priority=2)
        t2 = Task(title="Kusoma SQLModel", priority=1)
        print("Kabla ya commit, id =", t1.id)
        session.add(t1)
        session.add(t2)
        session.commit()
        session.refresh(t1)
        print("Baada ya commit, id =", t1.id)

    # READ
    with Session(engine) as session:
        zote = session.exec(select(Task)).all()
        print("Zote:", zote)
        muhimu = session.exec(select(Task).where(Task.priority == 1)).all()
        print("Priority 1:", [t.title for t in muhimu])
        moja = session.get(Task, 2)
        print("id 2:", moja)

    # UPDATE
    with Session(engine) as session:
        t = session.get(Task, 1)
        t.done = True
        session.add(t)
        session.commit()

    # DELETE
    with Session(engine) as session:
        t = session.get(Task, 2)
        session.delete(t)
        session.commit()
        print("Zilizobaki:", session.exec(select(Task)).all())


if __name__ == "__main__":
    main()

Run python db_demo.py. Because of echo=True, you will see the actual SQL in the terminal, for example:

CREATE TABLE task (
        id INTEGER NOT NULL,
        title VARCHAR NOT NULL,
        priority INTEGER NOT NULL,
        done BOOLEAN NOT NULL,
        PRIMARY KEY (id)
)
...
INSERT INTO task (title, priority, done) VALUES (?, ?, ?)
...
Kabla ya commit, id = None
Baada ya commit, id = 1
...
SELECT task.id, task.title, task.priority, task.done FROM task
Zote: [Task(priority=2, title='Kufua nguo', id=1, done=False), Task(priority=1, title='Kusoma SQLModel', id=2, done=False)]
...
Priority 1: ['Kusoma SQLModel']
id 2: priority=1 title='Kusoma SQLModel' id=2 done=False
...
Zilizobaki: [Task(priority=2, title='Kufua nguo', id=1, done=True)]

(The order of the fields inside the repr may look "mixed up" and differ between versions — that isn't a problem; what matters is the data.)

Say: "We didn't write SQL, but SQL was written. echo=True shows us what is going on behind the scenes." After that change it to echo=False — it is very noisy.

Run the script a second time — you will get new ids 3 and 4, and task 1 is still there. Say: "The data stayed! The demo.db file is the cabinet."

Step 2 — SQLite Viewer. Click demo.db in the VS Code Explorer → show the task table.

Step 3 — An API with a database. Create main_db.py:

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query, status
from sqlmodel import Field, Session, SQLModel, create_engine, select


# ---------- Models ----------
class TaskBase(SQLModel):
    title: str = Field(min_length=1, max_length=100, index=True)
    description: str | None = None
    priority: int = Field(default=3, ge=1, le=5)
    done: bool = False


class Task(TaskBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class TaskCreate(TaskBase):
    pass


class TaskPublic(TaskBase):
    id: int


class TaskUpdate(SQLModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    done: bool | None = None


# ---------- Database ----------
sqlite_url = "sqlite:///database.db"
engine = create_engine(sqlite_url, connect_args={"check_same_thread": False})


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()   # inaendeshwa server ikiwaka
    yield                    # server inafanya kazi hapa
    # (code baada ya yield inaendeshwa server ikizima)


app = FastAPI(title="Tasks API (SQLite)", lifespan=lifespan)


# ---------- Endpoints ----------
@app.post("/tasks", response_model=TaskPublic, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate, session: SessionDep):
    task = Task.model_validate(task_in)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@app.get("/tasks", response_model=list[TaskPublic])
def list_tasks(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    done: bool | None = None,
):
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    return session.exec(query.offset(offset).limit(limit)).all()


@app.get("/tasks/{task_id}", response_model=TaskPublic)
def read_task(task_id: int, session: SessionDep):
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task haipo")
    return task


@app.patch("/tasks/{task_id}", response_model=TaskPublic)
def update_task(task_id: int, task_in: TaskUpdate, session: SessionDep):
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task haipo")
    task.sqlmodel_update(task_in.model_dump(exclude_unset=True))
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, session: SessionDep):
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task haipo")
    session.delete(task)
    session.commit()

Run: fastapi dev main_db.py. In /docs:

1. POST 3 tasks.
2. GET /tasks?done=false&limit=2 → the first 2.
3. PATCH one with {"done": true}.
4. Stop the server (Ctrl+C) and start it again → GET /tasks → the data is still there! This is the big moment of the session.
5. Open database.db with SQLite Viewer.

Explain the new lines:

• Task.model_validate(task_in) — converts the (validated) TaskCreate into the table Task.
• task.sqlmodel_update(dict) — changes fields from a dict.
• Query(ge=1, le=100) — validation for a query parameter (like Field for the body). It stops someone asking for limit=1000000.

Questions to ask the student

1. Why id: int | None = Field(default=None, ...)? — Before saving we have no id (None); the database issues it at commit.
2. What happens if you forget session.commit()? — Nothing is saved; the changes are lost when the session closes.
3. What does session.refresh(task) do? — Re-reads from the database to get new values (such as id).
4. Why does TaskPublic have id: int (not int | None)? — In a response, the task has already been saved, so the id always exists; the docs become more accurate.
5. Why don't we use Task (the table) directly as the POST body? — Table models don't validate, and the client could send an id themselves.
6. What does session.get(Task, 99) return if it doesn't exist? — None.
7. If create_all runs a second time, does it delete data? — No; it only creates tables that don't exist.
8. get_session uses yield — what is that similar to from what we have learned? — Generators (Week 8) and context managers (with).

Common mistakes

In-class exercise

Exercise 2.1 — Add an endpoint PUT /tasks/{task_id} (replace the whole thing) to main_db.py using TaskCreate as the body.

Exercise 2.2 — Add a query parameter q: str | None = None to GET /tasks that searches within the title. Hint: Task.title.contains(q).

Exercise 2.3 — Add a query parameter panga: str = "id" (sort) that accepts "id" or "priority", and sorts the results by that column. For any other value, return 400. Hint: query.order_by(Task.priority).

Exercise 2.4 — Write a script ripoti.py (report; not an API) that reads database.db and prints the number of completed and uncompleted tasks.

@app.put("/tasks/{task_id}", response_model=TaskPublic)
def replace_task(task_id: int, task_in: TaskCreate, session: SessionDep):
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task haipo")
    task.sqlmodel_update(task_in.model_dump())   # fields zote
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

@app.get("/tasks", response_model=list[TaskPublic])
def list_tasks(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    done: bool | None = None,
    q: str | None = None,
    panga: str = "id",
):
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    if q:
        query = query.where(Task.title.contains(q))
    if panga == "id":
        query = query.order_by(Task.id)
    elif panga == "priority":
        query = query.order_by(Task.priority)
    else:
        raise HTTPException(status_code=400, detail="panga iwe 'id' au 'priority'")
    return session.exec(query.offset(offset).limit(limit)).all()

from sqlmodel import Session, create_engine, select

from main_db import Task   # tunatumia model ileile

engine = create_engine("sqlite:///database.db")

with Session(engine) as session:
    tasks = session.exec(select(Task)).all()
    zimekamilika = sum(1 for t in tasks if t.done)
    print(f"Jumla: {len(tasks)}")
    print(f"Zimekamilika: {zimekamilika}")
    print(f"Hazijakamilika: {len(tasks) - zimekamilika}")

Session 3 — Dependency Injection and Project Structure (Routers)

Session flow

1. Homework review
2. Depends in detail: a simple dependency (pagination), a dependency that uses another dependency (get_task_or_404)
3. Why split files; the folder structure
4. Moving main_db.py into app/ with database.py, models.py, routers/tasks.py
5. requirements.txt, .gitignore, push to GitHub
6. In-class exercise
7. Q&A + homework

Explanation for the teacher (read this first)

Dependency Injection (DI) sounds hard but the idea is simple: an endpoint says "I need this thing", and FastAPI "brings it" (injects it) by calling another function (the dependency) first.

def pagination(offset: int = 0, limit: int = 10) -> dict:
    return {"offset": offset, "limit": limit}

@app.get("/items")
def items(p: Annotated[dict, Depends(pagination)]):
    ...

FastAPI sees Depends(pagination) → reads the parameters of pagination (offset, limit — query parameters!) → calls it → gives the endpoint the result as p.

Benefits:

• Avoiding repeated code — pagination, getting a session, checking the logged-in user (Week 11!) — write once, use everywhere.
• Testing — in tests we can swap a dependency (for example, a test database session) using app.dependency_overrides (Week 11).
• Dependencies can depend on other dependencies — get_task_or_404 depends on SessionDep.

A dependency with a path parameter: def get_task_or_404(task_id: int, session: SessionDep) -> Task — task_id is taken from the path /tasks/{task_id} because the name matches. The endpoint then doesn't need to repeat the 404 code.

Within one request, FastAPI calls each dependency only once (cache) — so get_task_or_404 and the endpoint get the same session. This is important: if we change a task and commit on that session, it works.

Project structure. A single 300-line file is hard to read. We split by job:

tasks_api/
├── .venv/                (haiendi GitHub)
├── .gitignore
├── requirements.txt
├── database.db           (haiendi GitHub)
└── app/
    ├── __init__.py       (tupu — inafanya `app` kuwa package)
    ├── main.py           (FastAPI app, lifespan, include_router)
    ├── database.py       (engine, get_session, SessionDep, create_db_and_tables)
    ├── models.py         (SQLModel models)
    └── routers/
        ├── __init__.py   (tupu)
        └── tasks.py      (APIRouter ya /tasks)

APIRouter is like a "small FastAPI": you write @router.get(...) instead of @app.get(...), then main.py does app.include_router(tasks.router). prefix="/tasks" is added to the start of every path in the router; tags=["tasks"] organises /docs into groups.

Running: from the tasks_api folder (not inside app/):

fastapi dev app/main.py

The FastAPI CLI sees app/__init__.py, understands the module is app.main, and adds tasks_api/ to sys.path. So in the code we import with from app.database import SessionDep.

Trailing slash: a router with prefix="/tasks" and @router.get("/") produces the path /tasks/. If you request /tasks (without the /), FastAPI returns 307 Temporary Redirect to /tasks/ — browsers, /docs, requests and TestClient follow the redirect by themselves, so you normally won't see a problem. But it is good to know when you see a 307 in the logs.

Say this to the student

"Dependency injection is like a craftsman arriving at work: he doesn't say 'I'll make a hammer first'. He says 'give me a hammer', and the supervisor (FastAPI) brings him one. The craftsman focuses only on his work."

"get_task_or_404 is like the guard at the door: before you enter the office (the endpoint), the guard checks whether the task exists. If it doesn't, you are sent back to the gate (404) — the office doesn't even know you came."

"Splitting files is like a house: the kitchen (database.py), the store room (models.py), the living room (main.py), the bedrooms (routers). If you put everything in the living room, you will end up cooking in bed."

Type this code live (share screen)

Step 1 — A simple dependency. In main_db.py from the previous session, add above the endpoints:

class Pagination:
    def __init__(self, offset: int = 0, limit: Annotated[int, Query(ge=1, le=100)] = 20):
        self.offset = offset
        self.limit = limit


PaginationDep = Annotated[Pagination, Depends()]

And change list_tasks:

@app.get("/tasks", response_model=list[TaskPublic])
def list_tasks(session: SessionDep, p: PaginationDep, done: bool | None = None):
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    return session.exec(query.offset(p.offset).limit(p.limit)).all()

Explain: "A class can also be a dependency — FastAPI calls it like a function (Pagination(offset=..., limit=...)). Depends() with nothing inside means 'use the type itself (Pagination)'." Open /docs — offset and limit still appear as query parameters.

Step 2 — A dependency that uses a dependency. Add:

def get_task_or_404(task_id: int, session: SessionDep) -> Task:
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task haipo")
    return task


TaskDep = Annotated[Task, Depends(get_task_or_404)]

Now the endpoints become short:

@app.get("/tasks/{task_id}", response_model=TaskPublic)
def read_task(task: TaskDep):
    return task


@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task: TaskDep, session: SessionDep):
    session.delete(task)
    session.commit()

Try /tasks/999 → the 404 still works. Emphasise: "The endpoint's session and the one in get_task_or_404 are the SAME (FastAPI calls get_session only once per request)."

Step 3 — Project structure. In the terminal:

cd $HOME\Documents
mkdir tasks_api
cd tasks_api
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install "fastapi[standard]" sqlmodel
mkdir app, app\routers
New-Item app\__init__.py, app\routers\__init__.py
code .

app/database.py:

from typing import Annotated

from fastapi import Depends
from sqlmodel import Session, SQLModel, create_engine

sqlite_url = "sqlite:///database.db"
engine = create_engine(sqlite_url, connect_args={"check_same_thread": False})


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

app/models.py:

from sqlmodel import Field, SQLModel


class TaskBase(SQLModel):
    title: str = Field(min_length=1, max_length=100, index=True)
    description: str | None = None
    priority: int = Field(default=3, ge=1, le=5)
    done: bool = False


class Task(TaskBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class TaskCreate(TaskBase):
    pass


class TaskPublic(TaskBase):
    id: int


class TaskUpdate(SQLModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    done: bool | None = None

app/routers/tasks.py:

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlmodel import select

from app.database import SessionDep
from app.models import Task, TaskCreate, TaskPublic, TaskUpdate

router = APIRouter(prefix="/tasks", tags=["tasks"])


def get_task_or_404(task_id: int, session: SessionDep) -> Task:
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task haipo")
    return task


TaskDep = Annotated[Task, Depends(get_task_or_404)]


@router.post("/", response_model=TaskPublic, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate, session: SessionDep):
    task = Task.model_validate(task_in)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@router.get("/", response_model=list[TaskPublic])
def list_tasks(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    done: bool | None = None,
):
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    return session.exec(query.offset(offset).limit(limit)).all()


@router.get("/{task_id}", response_model=TaskPublic)
def read_task(task: TaskDep):
    return task


@router.patch("/{task_id}", response_model=TaskPublic)
def update_task(task: TaskDep, task_in: TaskUpdate, session: SessionDep):
    task.sqlmodel_update(task_in.model_dump(exclude_unset=True))
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task: TaskDep, session: SessionDep):
    session.delete(task)
    session.commit()

app/main.py:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import tasks


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Tasks API", version="0.1.0", lifespan=lifespan)
app.include_router(tasks.router)


@app.get("/", tags=["mwanzo"])
def mwanzo():
    return {"ujumbe": "Tasks API inafanya kazi. Nenda /docs"}

Run from tasks_api/:

fastapi dev app/main.py

Open /docs — show the tasks and mwanzo (home) groups (from tags). Do POST, GET, PATCH, DELETE again.

A question to ask here: "How does create_db_and_tables know about Task when database.py doesn't import models.py?" — Answer: main.py imports routers/tasks.py, which imports models.py. Importing a class with table=True registers it on SQLModel.metadata. So by the time lifespan runs, the metadata knows about Task.

Step 4 — Project files.

pip freeze > requirements.txt

.gitignore:

.venv/
__pycache__/
*.db

git init
git add .
git commit -m "Tasks API: FastAPI + SQLModel + routers"

(Then create a GitHub repo and git remote add origin ... + git push -u origin main, as in Week 8.)

Questions to ask the student

1. What is a dependency, in your own words? — A function (or class) that FastAPI calls before the endpoint to give the endpoint something it needs.
2. Where does the task_id in get_task_or_404 come from? — From the path /{task_id}, because the name matches.
3. Why does delete_task receive session again when get_task_or_404 already has a session? — The endpoint itself needs a session to do delete/commit; FastAPI gives it the same one.
4. prefix="/tasks" + @router.get("/{task_id}") = which path? — /tasks/{task_id}.
5. Why is app/__init__.py needed? — It makes app a package so that from app.models import ... works and fastapi dev app/main.py recognises the module app.main.
6. Why is *.db in .gitignore? — It is local data; each person builds their own database; it may also hold personal data.
7. How does requirements.txt help? — Someone else can run pip install -r requirements.txt to get the same libraries.

Common mistakes

In-class exercise

Exercise 3.1 — Add a second router app/routers/categories.py with a model Category (id, name — unique, min_length 2) and endpoints POST /categories/ and GET /categories/. Connect it in main.py.

Exercise 3.2 — Add category_id: int | None = Field(default=None, foreign_key="category.id") to TaskBase. In create_task, if category_id was sent but the category doesn't exist, return 400 "Category haipo" (category doesn't exist). (Delete database.db first because the model has changed.)

Exercise 3.3 — Add an endpoint GET /categories/{category_id}/tasks that returns all tasks in that category (404 if the category doesn't exist).

class CategoryBase(SQLModel):
    name: str = Field(min_length=2, max_length=50, unique=True, index=True)


class Category(CategoryBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class CategoryCreate(CategoryBase):
    pass


class CategoryPublic(CategoryBase):
    id: int


# na kwenye TaskBase ongeza:
#     category_id: int | None = Field(default=None, foreign_key="category.id")
# na kwenye TaskUpdate ongeza:
#     category_id: int | None = None

from fastapi import APIRouter, HTTPException, status
from sqlmodel import select

from app.database import SessionDep
from app.models import Category, CategoryCreate, CategoryPublic, Task, TaskPublic

router = APIRouter(prefix="/categories", tags=["categories"])


@router.post("/", response_model=CategoryPublic, status_code=status.HTTP_201_CREATED)
def create_category(cat_in: CategoryCreate, session: SessionDep):
    ipo = session.exec(select(Category).where(Category.name == cat_in.name)).first()
    if ipo:
        raise HTTPException(status_code=409, detail="Category hii ipo tayari")
    cat = Category.model_validate(cat_in)
    session.add(cat)
    session.commit()
    session.refresh(cat)
    return cat


@router.get("/", response_model=list[CategoryPublic])
def list_categories(session: SessionDep):
    return session.exec(select(Category)).all()


@router.get("/{category_id}/tasks", response_model=list[TaskPublic])   # 3.3
def tasks_za_category(category_id: int, session: SessionDep):
    if session.get(Category, category_id) is None:
        raise HTTPException(status_code=404, detail="Category haipo")
    return session.exec(select(Task).where(Task.category_id == category_id)).all()

from app.routers import categories, tasks
...
app.include_router(tasks.router)
app.include_router(categories.router)

from app.models import Category, Task, TaskCreate, TaskPublic, TaskUpdate

@router.post("/", response_model=TaskPublic, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate, session: SessionDep):
    if task_in.category_id is not None and session.get(Category, task_in.category_id) is None:
        raise HTTPException(status_code=400, detail="Category haipo")
    task = Task.model_validate(task_in)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

Homework

To submit: next session.

"Contacts API" — a contacts book. Build a new API with today's structure (app/main.py, app/database.py, app/models.py, app/routers/contacts.py).

The Contact model:

• id (primary key)
• jina (name): str, min_length 2, max_length 60

• simu (phone): str, must start with +255 or 0, followed by 9 digits. Hint (SQLModel's Field doesn't check a pattern, use this one from Pydantic):

  python

  Copy

  from typing import Annotated
  from pydantic import StringConstraints
  Simu = Annotated[str, StringConstraints(pattern=r"^(\+255|0)\d{9}$")]
  

• email: str | None

• mji (city): str = "Dar es Salaam"
• favorite: bool = False

Endpoints:

1. POST /contacts/ → 201; if the simu already exists → 409
2. GET /contacts/ → pagination (offset, limit ≤ 50), filter mji, filter favorite
3. GET /contacts/{id} → 404 if not found (use a get_contact_or_404 dependency)
4. PATCH /contacts/{id}
5. DELETE /contacts/{id} → 204
6. GET /contacts/tafuta?jina=... → contacts whose name contains that word

Also: requirements.txt, .gitignore, and a short README.md with the commands to run it.

from typing import Annotated

from pydantic import StringConstraints
from sqlmodel import Field, SQLModel

Simu = Annotated[str, StringConstraints(pattern=r"^(\+255|0)\d{9}$")]


class ContactBase(SQLModel):
    jina: str = Field(min_length=2, max_length=60, index=True)
    simu: Simu = Field(unique=True, index=True)
    email: str | None = None
    mji: str = "Dar es Salaam"
    favorite: bool = False


class Contact(ContactBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class ContactCreate(ContactBase):
    pass


class ContactPublic(ContactBase):
    id: int


class ContactUpdate(SQLModel):
    jina: str | None = Field(default=None, min_length=2, max_length=60)
    simu: Simu | None = None
    email: str | None = None
    mji: str | None = None
    favorite: bool | None = None

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlmodel import select

from app.database import SessionDep
from app.models import Contact, ContactCreate, ContactPublic, ContactUpdate

router = APIRouter(prefix="/contacts", tags=["contacts"])


def get_contact_or_404(contact_id: int, session: SessionDep) -> Contact:
    contact = session.get(Contact, contact_id)
    if contact is None:
        raise HTTPException(status_code=404, detail="Contact haipo")
    return contact


ContactDep = Annotated[Contact, Depends(get_contact_or_404)]


def simu_ipo(session, simu: str, isipokuwa_id: int | None = None) -> bool:
    c = session.exec(select(Contact).where(Contact.simu == simu)).first()
    return c is not None and c.id != isipokuwa_id


@router.post("/", response_model=ContactPublic, status_code=status.HTTP_201_CREATED)
def create_contact(c_in: ContactCreate, session: SessionDep):
    if simu_ipo(session, c_in.simu):
        raise HTTPException(status_code=409, detail="Namba hii ya simu ipo tayari")
    contact = Contact.model_validate(c_in)
    session.add(contact)
    session.commit()
    session.refresh(contact)
    return contact


@router.get("/", response_model=list[ContactPublic])
def list_contacts(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(ge=1, le=50)] = 20,
    mji: str | None = None,
    favorite: bool | None = None,
):
    q = select(Contact)
    if mji is not None:
        q = q.where(Contact.mji == mji)
    if favorite is not None:
        q = q.where(Contact.favorite == favorite)
    return session.exec(q.order_by(Contact.jina).offset(offset).limit(limit)).all()


# LAZIMA iwe juu ya /{contact_id}
@router.get("/tafuta", response_model=list[ContactPublic])
def tafuta(jina: str, session: SessionDep):
    return session.exec(select(Contact).where(Contact.jina.contains(jina))).all()


@router.get("/{contact_id}", response_model=ContactPublic)
def read_contact(contact: ContactDep):
    return contact


@router.patch("/{contact_id}", response_model=ContactPublic)
def update_contact(contact: ContactDep, c_in: ContactUpdate, session: SessionDep):
    data = c_in.model_dump(exclude_unset=True)
    if "simu" in data and simu_ipo(session, data["simu"], isipokuwa_id=contact.id):
        raise HTTPException(status_code=409, detail="Namba hii ya simu ipo tayari")
    contact.sqlmodel_update(data)
    session.add(contact)
    session.commit()
    session.refresh(contact)
    return contact


@router.delete("/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_contact(contact: ContactDep, session: SessionDep):
    session.delete(contact)
    session.commit()

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import contacts


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Contacts API", lifespan=lifespan)
app.include_router(contacts.router)

# Contacts API
```powershell
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
fastapi dev app/main.py
```
Fungua http://127.0.0.1:8000/docs

If you are asked a question you don't know

Say: "I don't know for sure — let's check the docs together", then search in front of the student. Good search terms: "fastapi sqlmodel one to many", "sqlmodel relationship", "pydantic field validator", "fastapi dependency yield". Official sources: FastAPI SQL Databases https://fastapi.tiangolo.com/tutorial/sql-databases/ · SQLModel docs https://sqlmodel.tiangolo.com/ · Pydantic docs https://docs.pydantic.dev/latest/ · FastAPI Dependencies https://fastapi.tiangolo.com/tutorial/dependencies/

End-of-week checklist

• Writes a Pydantic model with Field validation and understands the resulting 422
• Knows how FastAPI decides between path / query / body
• Uses response_model to hide fields (for example a password)
• Builds complete CRUD with correct 201, 204, 404
• Uses raise HTTPException(...)
• Understands PATCH + model_dump(exclude_unset=True)
• Explains table, row, column, primary key
• Creates an engine and a table, and uses Session (add, commit, refresh, get, exec(select(...)), delete)
• Understands why we have TaskBase / Task / TaskCreate / TaskPublic / TaskUpdate
• Uses SessionDep and writes their own dependency (get_..._or_404)
• Splits a project with app/ + routers/ and runs fastapi dev app/main.py
• The project is on GitHub with a correct requirements.txt and .gitignore