Coverage for docs_src/tutorial/fastapi/update/tutorial001_py39.py: 100%
59 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-27 00:03 +0000
1from typing import Optional 1abcdefghi
3from fastapi import FastAPI, HTTPException, Query 1abcdefghi
4from sqlmodel import Field, Session, SQLModel, create_engine, select 1abcdefghi
7class HeroBase(SQLModel): 1abcdefghi
8 name: str = Field(index=True) 1abcdefghi
9 secret_name: str 1abcdefghi
10 age: Optional[int] = Field(default=None, index=True) 1abcdefghi
13class Hero(HeroBase, table=True): 1abcdefghi
14 id: Optional[int] = Field(default=None, primary_key=True) 1abcdefghi
17class HeroCreate(HeroBase): 1abcdefghi
18 pass 1abcdefghi
21class HeroPublic(HeroBase): 1abcdefghi
22 id: int 1abcdefghi
25class HeroUpdate(SQLModel): 1abcdefghi
26 name: Optional[str] = None 1abcdefghi
27 secret_name: Optional[str] = None 1abcdefghi
28 age: Optional[int] = None 1abcdefghi
31sqlite_file_name = "database.db" 1abcdefghi
32sqlite_url = f"sqlite:///{sqlite_file_name}" 1abcdefghi
34connect_args = {"check_same_thread": False} 1abcdefghi
35engine = create_engine(sqlite_url, echo=True, connect_args=connect_args) 1abcdefghi
38def create_db_and_tables(): 1abcdefghi
39 SQLModel.metadata.create_all(engine) 1abcdefghi
42app = FastAPI() 1abcdefghi
45@app.on_event("startup") 1abcdefghi
46def on_startup(): 1abcdefghi
47 create_db_and_tables() 1abcdefghi
50@app.post("/heroes/", response_model=HeroPublic) 1abcdefghi
51def create_hero(hero: HeroCreate): 1abcdefghi
52 with Session(engine) as session: 1abcdefghi
53 db_hero = Hero.model_validate(hero) 1abcdefghi
54 session.add(db_hero) 1abcdefghi
55 session.commit() 1abcdefghi
56 session.refresh(db_hero) 1abcdefghi
57 return db_hero 1abcdefghi
60@app.get("/heroes/", response_model=list[HeroPublic]) 1abcdefghi
61def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)): 1abcdefghi
62 with Session(engine) as session: 1abcdefghi
63 heroes = session.exec(select(Hero).offset(offset).limit(limit)).all() 1abcdefghi
64 return heroes 1abcdefghi
67@app.get("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
68def read_hero(hero_id: int): 1abcdefghi
69 with Session(engine) as session: 1abcdefghi
70 hero = session.get(Hero, hero_id) 1abcdefghi
71 if not hero: 1abcdefghi
72 raise HTTPException(status_code=404, detail="Hero not found") 1abcdefghi
73 return hero 1abcdefghi
76@app.patch("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
77def update_hero(hero_id: int, hero: HeroUpdate): 1abcdefghi
78 with Session(engine) as session: 1abcdefghi
79 db_hero = session.get(Hero, hero_id) 1abcdefghi
80 if not db_hero: 1abcdefghi
81 raise HTTPException(status_code=404, detail="Hero not found") 1abcdefghi
82 hero_data = hero.model_dump(exclude_unset=True) 1abcdefghi
83 db_hero.sqlmodel_update(hero_data) 1abcdefghi
84 session.add(db_hero) 1abcdefghi
85 session.commit() 1abcdefghi
86 session.refresh(db_hero) 1abcdefghi
87 return db_hero 1abcdefghi