Coverage for docs_src / tutorial / fastapi / update / tutorial002_py310.py: 100%
69 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-11-09 00:02 +0000
1from fastapi import FastAPI, HTTPException, Query 1abcdefghi
2from sqlmodel import Field, Session, SQLModel, create_engine, select 1abcdefghi
5class HeroBase(SQLModel): 1abcdefghi
6 name: str = Field(index=True) 1abcdefghi
7 secret_name: str 1abcdefghi
8 age: int | None = Field(default=None, index=True) 1abcdefghi
11class Hero(HeroBase, table=True): 1abcdefghi
12 id: int | None = Field(default=None, primary_key=True) 1abcdefghi
13 hashed_password: str = Field() 1abcdefghi
16class HeroCreate(HeroBase): 1abcdefghi
17 password: str 1abcdefghi
20class HeroPublic(HeroBase): 1abcdefghi
21 id: int 1abcdefghi
24class HeroUpdate(SQLModel): 1abcdefghi
25 name: str | None = None 1abcdefghi
26 secret_name: str | None = None 1abcdefghi
27 age: int | None = None 1abcdefghi
28 password: str | None = 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) 1jklmnopqr
42def hash_password(password: str) -> str: 1abcdefghi
43 # Use something like passlib here
44 return f"not really hashed {password} hehehe" 1jklmnopqr
47app = FastAPI() 1abcdefghi
50@app.on_event("startup") 1abcdefghi
51def on_startup(): 1abcdefghi
52 create_db_and_tables() 1jklmnopqr
55@app.post("/heroes/", response_model=HeroPublic) 1abcdefghi
56def create_hero(hero: HeroCreate): 1abcdefghi
57 hashed_password = hash_password(hero.password) 1jklmnopqr
58 with Session(engine) as session: 1jklmnopqr
59 extra_data = {"hashed_password": hashed_password} 1jklmnopqr
60 db_hero = Hero.model_validate(hero, update=extra_data) 1jklmnopqr
61 session.add(db_hero) 1jklmnopqr
62 session.commit() 1jklmnopqr
63 session.refresh(db_hero) 1jklmnopqr
64 return db_hero 1jklmnopqr
67@app.get("/heroes/", response_model=list[HeroPublic]) 1abcdefghi
68def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)): 1abcdefghi
69 with Session(engine) as session: 1jklmnopqr
70 heroes = session.exec(select(Hero).offset(offset).limit(limit)).all() 1jklmnopqr
71 return heroes 1jklmnopqr
74@app.get("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
75def read_hero(hero_id: int): 1abcdefghi
76 with Session(engine) as session: 1jklmnopqr
77 hero = session.get(Hero, hero_id) 1jklmnopqr
78 if not hero: 1jklmnopqr
79 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
80 return hero 1jklmnopqr
83@app.patch("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
84def update_hero(hero_id: int, hero: HeroUpdate): 1abcdefghi
85 with Session(engine) as session: 1jklmnopqr
86 db_hero = session.get(Hero, hero_id) 1jklmnopqr
87 if not db_hero: 1jklmnopqr
88 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
89 hero_data = hero.model_dump(exclude_unset=True) 1jklmnopqr
90 extra_data = {} 1jklmnopqr
91 if "password" in hero_data: 1jklmnopqr
92 password = hero_data["password"] 1jklmnopqr
93 hashed_password = hash_password(password) 1jklmnopqr
94 extra_data["hashed_password"] = hashed_password 1jklmnopqr
95 db_hero.sqlmodel_update(hero_data, update=extra_data) 1jklmnopqr
96 session.add(db_hero) 1jklmnopqr
97 session.commit() 1jklmnopqr
98 session.refresh(db_hero) 1jklmnopqr
99 return db_hero 1jklmnopqr