Wiki 10: FastAPI na Pydantic, CRUD, na Database (SQLite + SQLModel)

Mwezi 3 | Lengo kuu: Mwanafunzi ajenge API kamili ya CRUD — kwanza kwenye memory, kisha kwenye database ya SQLite kwa SQLModel — akitumia Pydantic, HTTPException, Depends, na routers.

Maandalizi ya Mwalimu (soma kabla ya wiki kuanza)

Wiki hii ni nzito kuliko iliyopita — hakikisha umeandika na kuendesha code ya kila kipindi angalau mara moja.

1. Soma:
   • 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/ (hii ndiyo msingi wa Kipindi 2 — isome kwa makini)
   • FastAPI — "Dependencies" (kurasa ya kwanza): https://fastapi.tiangolo.com/tutorial/dependencies/
   • FastAPI — "Bigger Applications - Multiple Files": https://fastapi.tiangolo.com/tutorial/bigger-applications/

2. Sakinisha (kwenye venv ya mradi wa wiki hii):

   powershell

   Copy

   pip install "fastapi[standard]" sqlmodel
   

   SQLite haihitaji kusakinishwa — iko ndani ya Python (import sqlite3).

3. VS Code extension (inapendekezwa): "SQLite Viewer" (mwandishi: Florian Klampfer). Inakuruhusu kubofya faili database.db na kuona tables kama spreadsheet. Ni nzuri sana kwa kufundishia.

4. Elewa maneno haya kabla ya darasa: model, schema, request body, validation, serialization, CRUD, table, row, column, primary key, engine, session, commit, query, dependency injection, router.
5. Kumbuka mwisho wa Wiki 9: GET /bidhaa/99 ilirudisha 200 ikiwa na {"kosa": ...}. Leo tunairekebisha kwa HTTPException — anza kipindi kwa kumkumbusha.

Malengo ya Wiki (mwisho wa wiki mwanafunzi ataweza...)

1. Kuandika Pydantic models (BaseModel) na kuzitumia kama request body.
2. Kuweka sheria za validation kwa Field (min_length, ge, le...) na kusoma 422 errors.
3. Kutumia response_model / return type kuchuja data inayorudishwa (mfano kuficha password).
4. Kujenga CRUD kamili: POST, GET (list & one), PUT, PATCH, DELETE, pamoja na status codes sahihi (201, 204, 404).
5. Kutumia HTTPException kurudisha makosa.
6. Kueleza database ni nini; kuunda table ya SQLite kwa SQLModel; kufanya CRUD kwa Session.
7. Kutumia dependency injection (Depends, Annotated) kwa session na vitu vingine vya pamoja.
8. Kugawa mradi katika faili: main.py, database.py, models.py, routers/.

Kipindi 1 — Pydantic, Request Body, na CRUD ya Memory

Mtiririko wa kipindi

1. Marudio ya kazi ya nyumbani (API ya Maktaba) — mwanafunzi aiendeshe na kuonyesha /docs
2. Pydantic BaseModel nje ya FastAPI: validation na model_dump()
3. Request body kwenye POST; Field validation; kusoma 422
4. response_model: kuficha password
5. CRUD kamili ya Tasks kwenye memory + HTTPException + status codes
6. Zoezi darasani
7. Q&A + kazi ya nyumbani

Elezo kwa mwalimu (soma kwanza wewe)

Pydantic ni library (inakuja na FastAPI) inayofanya kazi kama dataclass "yenye mlinzi". Mwanafunzi anajua @dataclass (Wiki 7) — Pydantic BaseModel inafanana, lakini inathibitisha na kugeuza types wakati wa kuunda object:

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

dataclass haingekagua chochote.

Request body. Kwenye FastAPI, ukiweka argument yenye type ya Pydantic model, FastAPI inaelewa: "data hii inakuja kwenye body ya request kama JSON". Inasoma JSON → inaunda model → inaithibitisha. Ikishindikana → 422 yenye maelezo ya kila kosa.

Kanuni FastAPI inayotumia kuamua argument inatoka wapi:

• Jina liko kwenye path ({task_id}) → path parameter
• Type ni rahisi (int, str, bool...) → query parameter
• Type ni Pydantic model → body

Field(...) inaongeza sheria: min_length, max_length (kwa str), gt, ge, lt, le (kwa namba: greater than, greater-or-equal, less than, less-or-equal), default.

response_model (au return type annotation -> Model) inaiambia FastAPI: "data inayotoka iwe na umbo hili". FastAPI inachuja fields zisizokuwemo. Hii ni usalama: hata ukirudisha object yenye password, ikiwa response_model haina password, haitatoka.

CRUD = Create, Read, Update, Delete — operesheni nne za msingi kwenye data yoyote:

HTTPException: unaiyarusha (raise) ndani ya endpoint; FastAPI inaikamata na kurudisha status na {"detail": "..."}. Ni raise, si return, kwa sababu inasimamisha function papo hapo (kama exception yoyote — Wiki 5).

PATCH na exclude_unset=True: Kwa PATCH tunatumia model ambayo fields zote ni optional (None). Lakini kuna tatizo: tunatofautishaje "mtumiaji hakutuma description" na "mtumiaji ametuma description: null kwa makusudi"? model_dump(exclude_unset=True) inarudisha fields zilizotumwa tu. Kisha model_copy(update=...) inaunda nakala iliyobadilishwa.

"In-memory" maana yake data iko kwenye dict ya Python — ukizima server (au ikijiwasha upya baada ya kuhifadhi faili!), data inapotea. Hiyo ndiyo sababu ya database Kipindi 2.

Sema hivi kwa mwanafunzi

"Pydantic model ni kama fomu ya kujiandikisha benki. Fomu inasema: jina (lazima), umri (namba), simu (si lazima). Mhudumu (Pydantic) anaikagua kabla haijaingia ndani. Ukiandika 'umri: ishirini', anakurudishia fomu na kukuambia kosa liko wapi — hiyo ndiyo 422."

"response_model ni kama dirisha la benki: ndani kuna taarifa zako zote, hata PIN, lakini kupitia dirisha unapewa salio tu. PIN haitoki nje."

"CRUD ni kitabu cha mahudhurio: kuandika jina jipya (Create), kusoma orodha (Read), kurekebisha (Update), kufuta (Delete). Karibu kila app duniani — WhatsApp, M-Pesa, Instagram — ni CRUD juu ya vitu tofauti."

"raise HTTPException(404) ni kama mhudumu kusema 'hakuna' na kuondoka — haendelei na order."

Andika code hii live (share screen)

Hatua 1 — Folda mpya:

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

Hatua 2 — Pydantic peke yake. Unda 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

(Namba ya version kwenye link inaweza kutofautiana — si tatizo.)

Sisitiza: "21" imegeuzwa kuwa 21. Makosa YOTE yameripotiwa pamoja, si la kwanza tu.

Hatua 3 — Request body na response_model. Unda 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"}

Sema: "Tulirudisha user mwenye password, lakini response_model=UserOut imeichuja." Kisha jaribu body {"username": "ju", "password": "123"} → 422 yenye makosa mawili string_too_short.

Onyesha pia Schemas chini ya /docs — UserIn na UserOut zimeonekana zenyewe.

Hatua 4 — CRUD kamili ya memory. Badilisha main.py kuwa hivi (futa ya users):

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]

Jaribu kwa mpangilio huu kwenye /docs (onyesha status code kila mara):

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 → list ya 2
4. GET /tasks/5 → 404 {"detail":"Task 5 haipo"}
5. PATCH /tasks/1 body {"done": true} → title haijabadilika, done: true
6. GET /tasks?done=true → task 1 tu
7. PUT /tasks/2 body {"title": "Kununua vocha ya 5000"} → priority imerudi 3, description null (PUT inabadilisha kizima)
8. DELETE /tasks/2 → 204, hakuna body
9. GET /tasks/2 → 404
10. POST /tasks body {"title": "", "priority": 9} → 422

Kisha hifadhi faili (Ctrl+S) bila kubadilisha kitu, na fanya GET /tasks → []! Sema: "Server imejiwasha upya, memory imefutika. Ndiyo maana tunahitaji database — Kipindi 2."

Hatua 5 — Kuelewa exclude_unset. Ongeza kwa muda kwenye update_task mstari print(task_in, "|", mabadiliko) kisha PATCH {"done": true}. Terminal:

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

Sema: "Bila exclude_unset, tungeandika None juu ya title — tungeharibu data."

Maswali ya kumuuliza mwanafunzi

1. FastAPI inajuaje kwamba task_in: TaskCreate inatoka kwenye body? — Kwa sababu type yake ni Pydantic model.
2. Na done: bool | None = None kwenye list_tasks? — Ni type rahisi, haiko kwenye path → query parameter.
3. Kwa nini tunatumia raise badala ya return kwa HTTPException? — Ni exception; inasimamisha function papo hapo na FastAPI inaigeuza kuwa response ya error.
4. Kwa nini DELETE inarudisha 204 bila body? — 204 = "No Content": imefanikiwa, hakuna cha kurudisha.
5. Tofauti ya PUT na PATCH kwenye code yetu? — PUT inapokea TaskCreate (kitu kizima, fields zisizotumwa zinarudi default); PATCH inapokea TaskUpdate na kubadilisha fields zilizotumwa tu.
6. Kwa nini tuna models tatu (TaskCreate, Task, TaskUpdate) badala ya moja? — Kila moja ina kazi: mteja hapaswi kutuma id (server ndiyo inaitoa), Task ina id kwa response, TaskUpdate ina fields zote optional kwa PATCH.
7. ge=1, le=5 maana yake? — Greater-or-equal 1, less-or-equal 5: thamani kati ya 1 na 5.
8. Kwa nini data ilipotea tulipohifadhi faili? — fastapi dev ilijiwasha upya; dict iko kwenye memory ya process.

Makosa ya kawaida

Zoezi darasani

Zoezi 1.1 — Ongeza field due_date: date | None = None kwenye TaskCreate na TaskUpdate (from datetime import date). Jaribu kutuma "2027-01-20" na "jana". Nini kinatokea?

Zoezi 1.2 — Ongeza endpoint GET /tasks/stats inayorudisha {"jumla": n, "zilizokamilika": k, "zilizobaki": n - k}. (Kumbuka route order!)

Zoezi 1.3 — Ongeza query parameter q: str | None = None kwenye GET /tasks inayochuja tasks ambazo title ina neno q (bila kujali herufi kubwa/ndogo).

Zoezi 1.4 — Kwenye POST /tasks, zuia tasks mbili zenye title ileile: ikiwa ipo tayari, rudisha 409 Conflict yenye detail "Task yenye title hii ipo tayari".

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)

Kipindi 2 — Database: SQLite na SQLModel

Mtiririko wa kipindi

1. Marudio ya kazi ya nyumbani
2. Database ni nini: table, row, column, primary key; SQL kwa ufupi
3. SQLModel peke yake (script): engine, create table, insert, select, update, delete
4. Kuangalia database.db kwa SQLite Viewer
5. Kuunganisha na FastAPI: models nyingi, get_session, SessionDep, CRUD endpoints
6. Zoezi darasani
7. Q&A + kazi ya nyumbani

Elezo kwa mwalimu (soma kwanza wewe)

Database ni programu maalum ya kuhifadhi data kwa kudumu na kuitafuta haraka. Relational database (kama SQLite, PostgreSQL, MySQL) inahifadhi data kwenye tables:

• Table = kama sheet ya Excel (mfano task)
• Column = field (title, done...)
• Row = rekodi moja (task moja)
• Primary key = column inayotambulisha row kipekee (kawaida id); database inaitoa yenyewe (1, 2, 3...)

SQL ni lugha ya kuongea na database: SELECT * FROM task WHERE done = 1;. Hatutaandika SQL kwa mkono — SQLModel itaiandika kwa ajili yetu.

SQLite ni database ndani ya faili moja (database.db). Haina server, haina password — inafaa kujifunzia na apps ndogo. Python inaikuja nayo.

SQLModel (imeandikwa na mwandishi wa FastAPI) = Pydantic + SQLAlchemy. Class moja inaweza kuwa:

• Pydantic model (validation, JSON), NA
• Table ya database — ukiongeza table=True.

Vipande vinne vya kukumbuka:

1. Model ya table: class Task(SQLModel, table=True) yenye id: int | None = Field(default=None, primary_key=True). id ni None kabla ya kuhifadhiwa; database inaijaza.
2. Engine: create_engine("sqlite:///database.db") — "muunganiko" na database. Unda MOJA kwa app nzima.
3. SQLModel.metadata.create_all(engine) — inaunda tables zote ambazo bado hazipo (haifuti zilizopo, wala haibadilishi columns za table iliyopo — ukibadilisha model, futa database.db wakati wa kujifunza).
4. Session: with Session(engine) as session: — "mazungumzo" moja na database. Ndani yake:
   • session.add(obj) — andaa kuhifadhi
   • session.commit() — hifadhi kweli (bila commit, hakuna kinachohifadhiwa!)
   • session.refresh(obj) — soma upya kutoka database (kupata id mpya)
   • session.get(Task, 5) — tafuta kwa primary key (inarudisha None kama haipo)
   • session.exec(select(Task).where(Task.done == True)).all() — query
   • session.delete(obj) kisha commit()

Muhimu — validation: Model yenye table=True haifanyi validation wakati wa kuunda object (kwa makusudi, kwa sababu za SQLAlchemy). Ndiyo maana tunatumia models tofauti kwa input (TaskCreate, bila table=True) ambazo zinathibitisha, kisha tunazigeuza kuwa Task ya table. Mpangilio unaopendekezwa na 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} — SQLite kwa default inakataa kutumika na thread zaidi ya moja; FastAPI inaendesha def endpoints kwenye threads tofauti. Hii inaruhusu. Tunaweka kwa SQLite tu.

Dependency get_session: function inayotumia yield (generator — Wiki 8!):

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

FastAPI inaiita kabla ya endpoint, inampa endpoint session, na baada ya response inarudi na kufunga with (session inafungwa). SessionDep = Annotated[Session, Depends(get_session)] ni jina fupi tu ili tusiandike hilo kila mara.

Lifespan: tunataka create_all iendeshwe mara moja server inapowaka. Tunatumia lifespan (njia ya kisasa; @app.on_event("startup") ni ya zamani/deprecated).

Sema hivi kwa mwanafunzi

"Database ni kama kabati la mafaili ofisini. Table ni droo moja (droo ya 'tasks'), kila faili ndani ni row, na kila faili lina namba yake ya kipekee juu — hiyo ni primary key."

"Memory (dict) ni kama kuandika ubaoni — mtu akifuta (server ikizima), imekwisha. Database ni kama kuandika kwenye daftari na kulifungia kabatini."

"Session ni kama kwenda benki: unaingia (with Session), unajaza vocha (add), lakini pesa haijawekwa mpaka keshia agonge muhuri (commit). Ukiondoka bila muhuri, hakuna kilichotokea."

"Depends(get_session) ni kama hoteli: hukubebi taulo lako; ukifika chumbani taulo liko tayari, na ukiondoka wanalichukua. Endpoint inapewa session tayari, na FastAPI inaifunga baadaye."

Andika code hii live (share screen)

Hatua 1 — SQLModel peke yake. Unda 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()

Endesha python db_demo.py. Kwa sababu ya echo=True, utaona SQL halisi kwenye terminal, mfano:

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)]

(Mpangilio wa fields ndani ya repr unaweza kuonekana "umechanganyika" na kutofautiana kati ya versions — si tatizo; kilicho muhimu ni data.)

Sema: "Hatukuandika SQL, lakini SQL imeandikwa. echo=True inatuonyesha kinachoendelea nyuma ya pazia." Baada ya hapo badilisha kuwa echo=False — kelele nyingi.

Endesha script mara ya pili — utapata id 3 na 4 mpya, na task 1 bado ipo. Sema: "Data imebaki! Faili demo.db ndiyo kabati."

Hatua 2 — SQLite Viewer. Bofya demo.db kwenye Explorer ya VS Code → onyesha table task.

Hatua 3 — API yenye database. Unda 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()

Endesha: fastapi dev main_db.py. Kwenye /docs:

1. POST tasks 3.
2. GET /tasks?done=false&limit=2 → 2 za kwanza.
3. PATCH moja {"done": true}.
4. Simamisha server (Ctrl+C) na uiwashe tena → GET /tasks → data bado ipo! Hili ndilo tukio kubwa la kipindi.
5. Fungua database.db kwa SQLite Viewer.

Eleza mistari mipya:

• Task.model_validate(task_in) — geuza TaskCreate (iliyothibitishwa) kuwa Task ya table.
• task.sqlmodel_update(dict) — badilisha fields kutoka dict.
• Query(ge=1, le=100) — validation kwa query parameter (kama Field kwa body). Inazuia mtu kuomba limit=1000000.

Maswali ya kumuuliza mwanafunzi

1. Kwa nini id: int | None = Field(default=None, ...)? — Kabla ya kuhifadhi hatuna id (None); database inaitoa wakati wa commit.
2. Nini kinatokea ukisahau session.commit()? — Hakuna kinachohifadhiwa; mabadiliko yanapotea session ikifungwa.
3. session.refresh(task) ina kazi gani? — Kusoma upya kutoka database ili kupata thamani mpya (kama id).
4. Kwa nini TaskPublic ina id: int (si int | None)? — Kwenye response, task tayari imehifadhiwa, hivyo id ipo kila mara; docs zinakuwa sahihi zaidi.
5. Kwa nini hatutumii Task (table) moja kwa moja kama body ya POST? — Table models hazifanyi validation, na mteja angeweza kutuma id mwenyewe.
6. session.get(Task, 99) inarudisha nini ikiwa haipo? — None.
7. create_all ikiendeshwa mara ya pili inafuta data? — Hapana; inaunda tables zisizokuwepo tu.
8. get_session inatumia yield — inafanana na nini tulichojifunza? — Generators (Wiki 8) na context managers (with).

Makosa ya kawaida

Zoezi darasani

Zoezi 2.1 — Ongeza endpoint PUT /tasks/{task_id} (badilisha kizima) kwenye main_db.py ukitumia TaskCreate kama body.

Zoezi 2.2 — Ongeza query parameter q: str | None = None kwenye GET /tasks inayotafuta ndani ya title. Dokezo: Task.title.contains(q).

Zoezi 2.3 — Ongeza query parameter panga: str = "id" inayokubali "id" au "priority", na kupanga matokeo kwa column hiyo. Ikiwa thamani nyingine, rudisha 400. Dokezo: query.order_by(Task.priority).

Zoezi 2.4 — Andika script ripoti.py (si API) inayosoma database.db na kuchapisha idadi ya tasks zilizokamilika na zisizokamilika.

@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}")

Kipindi 3 — Dependency Injection na Muundo wa Mradi (Routers)

Mtiririko wa kipindi

1. Marudio ya kazi ya nyumbani
2. Depends kwa undani: dependency rahisi (pagination), dependency inayotumia dependency nyingine (get_task_or_404)
3. Kwa nini kugawa faili; muundo wa folda
4. Kuhamisha main_db.py kwenda app/ yenye database.py, models.py, routers/tasks.py
5. requirements.txt, .gitignore, push GitHub
6. Zoezi darasani
7. Q&A + kazi ya nyumbani

Elezo kwa mwalimu (soma kwanza wewe)

Dependency Injection (DI) inasikika ngumu lakini wazo ni rahisi: endpoint inasema "nahitaji kitu hiki", na FastAPI "inakileta" (inject) kwa kuita function nyingine (dependency) kwanza.

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 inaona Depends(pagination) → inasoma parameters za pagination (offset, limit — query parameters!) → inaiita → inampa endpoint matokeo kama p.

Faida:

• Kuepuka kurudia code — pagination, kupata session, kuangalia user aliyeingia (Wiki 11!) — andika mara moja, tumia kila mahali.
• Testing — kwenye tests tunaweza kubadilisha dependency (mfano session ya database ya test) kwa app.dependency_overrides (Wiki 11).
• Dependencies zinaweza kutegemea dependencies nyingine — get_task_or_404 inategemea SessionDep.

Dependency yenye path parameter: def get_task_or_404(task_id: int, session: SessionDep) -> Task — task_id itachukuliwa kutoka path /tasks/{task_id} kwa sababu jina linafanana. Endpoint basi haihitaji kurudia code ya 404.

Ndani ya request moja, FastAPI inaita kila dependency mara moja tu (cache) — hivyo get_task_or_404 na endpoint zinapata session ileile. Hili ni muhimu: tukibadilisha task na kufanya commit kwenye session hiyo, inafanya kazi.

Muundo wa mradi. Faili moja la mistari 300 ni gumu kusoma. Tunagawa kwa kazi:

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 ni kama "FastAPI ndogo": unaandika @router.get(...) badala ya @app.get(...), kisha main.py inafanya app.include_router(tasks.router). prefix="/tasks" inaongezwa mwanzoni mwa kila path ya router; tags=["tasks"] inapanga /docs kwa makundi.

Kuendesha: kutoka folda ya tasks_api (si ndani ya app/):

fastapi dev app/main.py

FastAPI CLI inaona app/__init__.py, inaelewa module ni app.main, na inaongeza tasks_api/ kwenye sys.path. Kwa hiyo ndani ya code tunaimport kwa from app.database import SessionDep.

Trailing slash: router yenye prefix="/tasks" na @router.get("/") inatoa path /tasks/. Ukiomba /tasks (bila /), FastAPI inarudisha 307 Temporary Redirect kwenda /tasks/ — browser, /docs, requests na TestClient zinafuata redirect zenyewe, hivyo kwa kawaida hutaona tatizo. Lakini ni vizuri kujua ukiona 307 kwenye logs.

Sema hivi kwa mwanafunzi

"Dependency injection ni kama fundi anayefika kazini: hasemi 'nitatengeneza nyundo kwanza'. Anasema 'nipe nyundo', na msimamizi (FastAPI) anamletea. Fundi anazingatia kazi yake tu."

"get_task_or_404 ni kama mlinzi wa mlangoni: kabla hujaingia ofisini (endpoint), anakagua kama task ipo. Kama haipo, unarudishwa getini (404) — ofisi haijui hata ulikuja."

"Kugawa faili ni kama nyumba: jikoni (database.py), stoo (models.py), sebuleni (main.py), vyumba (routers). Ukiweka kila kitu sebuleni, utapika kitandani."

Andika code hii live (share screen)

Hatua 1 — Dependency rahisi. Kwenye main_db.py ya jana, ongeza juu ya 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()]

Na ubadilishe 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()

Eleza: "Class pia inaweza kuwa dependency — FastAPI inaiita kama function (Pagination(offset=..., limit=...)). Depends() bila kitu ndani maana yake 'tumia type yenyewe (Pagination)'." Fungua /docs — offset na limit bado zinaonekana kama query parameters.

Hatua 2 — Dependency inayotumia dependency. Ongeza:

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)]

Sasa endpoints zinakuwa fupi:

@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()

Jaribu /tasks/999 → 404 bado inafanya kazi. Sisitiza: "session ya endpoint na ya get_task_or_404 ni ILEILE (FastAPI inaita get_session mara moja tu kwa request)."

Hatua 3 — Muundo wa mradi. Kwenye 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"}

Endesha kutoka tasks_api/:

fastapi dev app/main.py

Fungua /docs — onyesha kundi la tasks na mwanzo (kutoka tags). Fanya POST, GET, PATCH, DELETE tena.

Swali la kuuliza hapa: "create_db_and_tables inajuaje kuhusu Task wakati database.py haiimport models.py?" — Jibu: main.py inaimport routers/tasks.py, ambayo inaimport models.py. Kuimport class yenye table=True kunaisajili kwenye SQLModel.metadata. Hivyo kufikia wakati lifespan inaendesha, metadata inaijua Task.

Hatua 4 — Faili za mradi.

pip freeze > requirements.txt

.gitignore:

.venv/
__pycache__/
*.db

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

(Kisha unda repo GitHub na git remote add origin ... + git push -u origin main, kama Wiki 8.)

Maswali ya kumuuliza mwanafunzi

1. Dependency ni nini kwa maneno yako? — Function (au class) ambayo FastAPI inaiita kabla ya endpoint ili kumpa endpoint kitu anachohitaji.
2. task_id ya get_task_or_404 inatoka wapi? — Kutoka path /{task_id} kwa sababu jina linafanana.
3. Kwa nini delete_task inapokea session tena wakati get_task_or_404 tayari ina session? — Endpoint inahitaji session yenyewe kufanya delete/commit; FastAPI inampa ileile.
4. prefix="/tasks" + @router.get("/{task_id}") = path gani? — /tasks/{task_id}.
5. Kwa nini app/__init__.py inahitajika? — Inaifanya app kuwa package ili from app.models import ... ifanye kazi na fastapi dev app/main.py itambue module app.main.
6. Kwa nini *.db iko kwenye .gitignore? — Ni data ya local; kila mtu anajenga database yake; pia inaweza kuwa na data binafsi.
7. requirements.txt inasaidiaje? — Mtu mwingine anaweza pip install -r requirements.txt kupata libraries zilezile.

Makosa ya kawaida

Zoezi darasani

Zoezi 3.1 — Ongeza router ya pili app/routers/categories.py yenye model Category (id, name — unique, min_length 2) na endpoints POST /categories/ na GET /categories/. Iunganishe kwenye main.py.

Zoezi 3.2 — Ongeza category_id: int | None = Field(default=None, foreign_key="category.id") kwenye TaskBase. Kwenye create_task, kama category_id imetumwa lakini category haipo, rudisha 400 "Category haipo". (Futa database.db kwanza kwa sababu model imebadilika.)

Zoezi 3.3 — Ongeza endpoint GET /categories/{category_id}/tasks inayorudisha tasks zote za category hiyo (404 kama category haipo).

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

Kazi ya Nyumbani

Kuwasilisha: kipindi kijacho.

"Contacts API" — Daftari la Mawasiliano. Jenga API mpya kwa muundo wa leo (app/main.py, app/database.py, app/models.py, app/routers/contacts.py).

Model ya Contact:

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

• simu: str, lazima ianze na +255 au 0, ikifuatiwa na tarakimu 9. Dokezo (Field ya SQLModel haikagui pattern, tumia hii ya 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: str = "Dar es Salaam"
• favorite: bool = False

Endpoints:

1. POST /contacts/ → 201; ikiwa simu tayari ipo → 409
2. GET /contacts/ → pagination (offset, limit ≤ 50), filter mji, filter favorite
3. GET /contacts/{id} → 404 kama haipo (tumia dependency get_contact_or_404)
4. PATCH /contacts/{id}
5. DELETE /contacts/{id} → 204
6. GET /contacts/tafuta?jina=... → contacts ambazo jina lina neno hilo

Pia: requirements.txt, .gitignore, na README.md fupi yenye amri za kuendesha.

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

Ukiulizwa swali usilolijua

Sema: "Sijui kwa uhakika — tuangalie docs pamoja", kisha tafuta mbele yake. Maneno mazuri ya kutafuta: "fastapi sqlmodel one to many", "sqlmodel relationship", "pydantic field validator", "fastapi dependency yield". Vyanzo rasmi: 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/

Checklist ya mwisho wa wiki

• Anaandika Pydantic model yenye Field validation na kuelewa 422 inayotokana nayo
• Anajua FastAPI inavyoamua path / query / body
• Anatumia response_model kuficha fields (mfano password)
• Anajenga CRUD kamili yenye 201, 204, 404 sahihi
• Anatumia raise HTTPException(...)
• Anaelewa PATCH + model_dump(exclude_unset=True)
• Anaeleza table, row, column, primary key
• Anaunda engine, table, na kutumia Session (add, commit, refresh, get, exec(select(...)), delete)
• Anaelewa kwa nini tuna TaskBase / Task / TaskCreate / TaskPublic / TaskUpdate
• Anatumia SessionDep na kuandika dependency yake mwenyewe (get_..._or_404)
• Anagawa mradi kwa app/ + routers/ na kuendesha fastapi dev app/main.py
• Mradi uko GitHub na requirements.txt na .gitignore sahihi