Coverage for docs_src / tutorial / fastapi / relationships / tutorial001_py39.py: 100%
123 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-01-06 21:09 +0000
1from typing import Optional 1abcdefghi
3from fastapi import Depends, FastAPI, HTTPException, Query 1abcdefghi
4from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select 1abcdefghi
7class TeamBase(SQLModel): 1abcdefghi
8 name: str = Field(index=True) 1abcdefghi
9 headquarters: str 1abcdefghi
12class Team(TeamBase, table=True): 1abcdefghi
13 id: Optional[int] = Field(default=None, primary_key=True) 1abcdefghi
15 heroes: list["Hero"] = Relationship(back_populates="team") 1abcdefghi
18class TeamCreate(TeamBase): 1abcdefghi
19 pass 1abcdefghi
22class TeamPublic(TeamBase): 1abcdefghi
23 id: int 1abcdefghi
26class TeamUpdate(SQLModel): 1abcdefghi
27 id: Optional[int] = None 1abcdefghi
28 name: Optional[str] = None 1abcdefghi
29 headquarters: Optional[str] = None 1abcdefghi
32class HeroBase(SQLModel): 1abcdefghi
33 name: str = Field(index=True) 1abcdefghi
34 secret_name: str 1abcdefghi
35 age: Optional[int] = Field(default=None, index=True) 1abcdefghi
37 team_id: Optional[int] = Field(default=None, foreign_key="team.id") 1abcdefghi
40class Hero(HeroBase, table=True): 1abcdefghi
41 id: Optional[int] = Field(default=None, primary_key=True) 1abcdefghi
43 team: Optional[Team] = Relationship(back_populates="heroes") 1abcdefghi
46class HeroPublic(HeroBase): 1abcdefghi
47 id: int 1abcdefghi
50class HeroCreate(HeroBase): 1abcdefghi
51 pass 1abcdefghi
54class HeroUpdate(SQLModel): 1abcdefghi
55 name: Optional[str] = None 1abcdefghi
56 secret_name: Optional[str] = None 1abcdefghi
57 age: Optional[int] = None 1abcdefghi
58 team_id: Optional[int] = None 1abcdefghi
61class HeroPublicWithTeam(HeroPublic): 1abcdefghi
62 team: Optional[TeamPublic] = None 1abcdefghi
65class TeamPublicWithHeroes(TeamPublic): 1abcdefghi
66 heroes: list[HeroPublic] = [] 1abcdefghi
69sqlite_file_name = "database.db" 1abcdefghi
70sqlite_url = f"sqlite:///{sqlite_file_name}" 1abcdefghi
72connect_args = {"check_same_thread": False} 1abcdefghi
73engine = create_engine(sqlite_url, echo=True, connect_args=connect_args) 1abcdefghi
76def create_db_and_tables(): 1abcdefghi
77 SQLModel.metadata.create_all(engine) 1jklmnopqr
80def get_session(): 1abcdefghi
81 with Session(engine) as session: 1jklmnopqr
82 yield session 1jklmnopqr
85app = FastAPI() 1abcdefghi
88@app.on_event("startup") 1abcdefghi
89def on_startup(): 1abcdefghi
90 create_db_and_tables() 1jklmnopqr
93@app.post("/heroes/", response_model=HeroPublic) 1abcdefghi
94def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate): 1abcdefghi
95 db_hero = Hero.model_validate(hero) 1jklmnopqr
96 session.add(db_hero) 1jklmnopqr
97 session.commit() 1jklmnopqr
98 session.refresh(db_hero) 1jklmnopqr
99 return db_hero 1jklmnopqr
102@app.get("/heroes/", response_model=list[HeroPublic]) 1abcdefghi
103def read_heroes( 1abcdefghi
104 *,
105 session: Session = Depends(get_session),
106 offset: int = 0,
107 limit: int = Query(default=100, le=100),
108):
109 heroes = session.exec(select(Hero).offset(offset).limit(limit)).all() 1jklmnopqr
110 return heroes 1jklmnopqr
113@app.get("/heroes/{hero_id}", response_model=HeroPublicWithTeam) 1abcdefghi
114def read_hero(*, session: Session = Depends(get_session), hero_id: int): 1abcdefghi
115 hero = session.get(Hero, hero_id) 1jklmnopqr
116 if not hero: 1jklmnopqr
117 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
118 return hero 1jklmnopqr
121@app.patch("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
122def update_hero( 1abcdefghi
123 *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
124):
125 db_hero = session.get(Hero, hero_id) 1jklmnopqr
126 if not db_hero: 1jklmnopqr
127 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
128 hero_data = hero.model_dump(exclude_unset=True) 1jklmnopqr
129 db_hero.sqlmodel_update(hero_data) 1jklmnopqr
130 session.add(db_hero) 1jklmnopqr
131 session.commit() 1jklmnopqr
132 session.refresh(db_hero) 1jklmnopqr
133 return db_hero 1jklmnopqr
136@app.delete("/heroes/{hero_id}") 1abcdefghi
137def delete_hero(*, session: Session = Depends(get_session), hero_id: int): 1abcdefghi
138 hero = session.get(Hero, hero_id) 1jklmnopqr
139 if not hero: 1jklmnopqr
140 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
141 session.delete(hero) 1jklmnopqr
142 session.commit() 1jklmnopqr
143 return {"ok": True} 1jklmnopqr
146@app.post("/teams/", response_model=TeamPublic) 1abcdefghi
147def create_team(*, session: Session = Depends(get_session), team: TeamCreate): 1abcdefghi
148 db_team = Team.model_validate(team) 1jklmnopqr
149 session.add(db_team) 1jklmnopqr
150 session.commit() 1jklmnopqr
151 session.refresh(db_team) 1jklmnopqr
152 return db_team 1jklmnopqr
155@app.get("/teams/", response_model=list[TeamPublic]) 1abcdefghi
156def read_teams( 1abcdefghi
157 *,
158 session: Session = Depends(get_session),
159 offset: int = 0,
160 limit: int = Query(default=100, le=100),
161):
162 teams = session.exec(select(Team).offset(offset).limit(limit)).all() 1jklmnopqr
163 return teams 1jklmnopqr
166@app.get("/teams/{team_id}", response_model=TeamPublicWithHeroes) 1abcdefghi
167def read_team(*, team_id: int, session: Session = Depends(get_session)): 1abcdefghi
168 team = session.get(Team, team_id) 1jklmnopqr
169 if not team: 1jklmnopqr
170 raise HTTPException(status_code=404, detail="Team not found") 1jklmnopqr
171 return team 1jklmnopqr
174@app.patch("/teams/{team_id}", response_model=TeamPublic) 1abcdefghi
175def update_team( 1abcdefghi
176 *,
177 session: Session = Depends(get_session),
178 team_id: int,
179 team: TeamUpdate,
180):
181 db_team = session.get(Team, team_id) 1jklmnopqr
182 if not db_team: 1jklmnopqr
183 raise HTTPException(status_code=404, detail="Team not found") 1jklmnopqr
184 team_data = team.model_dump(exclude_unset=True) 1jklmnopqr
185 db_team.sqlmodel_update(team_data) 1jklmnopqr
186 session.add(db_team) 1jklmnopqr
187 session.commit() 1jklmnopqr
188 session.refresh(db_team) 1jklmnopqr
189 return db_team 1jklmnopqr
192@app.delete("/teams/{team_id}") 1abcdefghi
193def delete_team(*, session: Session = Depends(get_session), team_id: int): 1abcdefghi
194 team = session.get(Team, team_id) 1jklmnopqr
195 if not team: 1jklmnopqr
196 raise HTTPException(status_code=404, detail="Team not found") 1jklmnopqr
197 session.delete(team) 1jklmnopqr
198 session.commit() 1jklmnopqr
199 return {"ok": True} 1jklmnopqr