Coverage for docs_src / tutorial / fastapi / teams / tutorial001_py310.py: 100%
117 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-11-09 00:02 +0000
1from fastapi import Depends, FastAPI, HTTPException, Query 1abcdefghi
2from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select 1abcdefghi
5class TeamBase(SQLModel): 1abcdefghi
6 name: str = Field(index=True) 1abcdefghi
7 headquarters: str 1abcdefghi
10class Team(TeamBase, table=True): 1abcdefghi
11 id: int | None = Field(default=None, primary_key=True) 1abcdefghi
13 heroes: list["Hero"] = Relationship(back_populates="team") 1abcdefghi
16class TeamCreate(TeamBase): 1abcdefghi
17 pass 1abcdefghi
20class TeamPublic(TeamBase): 1abcdefghi
21 id: int 1abcdefghi
24class TeamUpdate(SQLModel): 1abcdefghi
25 name: str | None = None 1abcdefghi
26 headquarters: str | None = None 1abcdefghi
29class HeroBase(SQLModel): 1abcdefghi
30 name: str = Field(index=True) 1abcdefghi
31 secret_name: str 1abcdefghi
32 age: int | None = Field(default=None, index=True) 1abcdefghi
34 team_id: int | None = Field(default=None, foreign_key="team.id") 1abcdefghi
37class Hero(HeroBase, table=True): 1abcdefghi
38 id: int | None = Field(default=None, primary_key=True) 1abcdefghi
40 team: Team | None = Relationship(back_populates="heroes") 1abcdefghi
43class HeroPublic(HeroBase): 1abcdefghi
44 id: int 1abcdefghi
47class HeroCreate(HeroBase): 1abcdefghi
48 pass 1abcdefghi
51class HeroUpdate(SQLModel): 1abcdefghi
52 name: str | None = None 1abcdefghi
53 secret_name: str | None = None 1abcdefghi
54 age: int | None = None 1abcdefghi
55 team_id: int | None = None 1abcdefghi
58sqlite_file_name = "database.db" 1abcdefghi
59sqlite_url = f"sqlite:///{sqlite_file_name}" 1abcdefghi
61connect_args = {"check_same_thread": False} 1abcdefghi
62engine = create_engine(sqlite_url, echo=True, connect_args=connect_args) 1abcdefghi
65def create_db_and_tables(): 1abcdefghi
66 SQLModel.metadata.create_all(engine) 1jklmnopqr
69def get_session(): 1abcdefghi
70 with Session(engine) as session: 1jklmnopqr
71 yield session 1jklmnopqr
74app = FastAPI() 1abcdefghi
77@app.on_event("startup") 1abcdefghi
78def on_startup(): 1abcdefghi
79 create_db_and_tables() 1jklmnopqr
82@app.post("/heroes/", response_model=HeroPublic) 1abcdefghi
83def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate): 1abcdefghi
84 db_hero = Hero.model_validate(hero) 1jklmnopqr
85 session.add(db_hero) 1jklmnopqr
86 session.commit() 1jklmnopqr
87 session.refresh(db_hero) 1jklmnopqr
88 return db_hero 1jklmnopqr
91@app.get("/heroes/", response_model=list[HeroPublic]) 1abcdefghi
92def read_heroes( 1abcdefghi
93 *,
94 session: Session = Depends(get_session),
95 offset: int = 0,
96 limit: int = Query(default=100, le=100),
97):
98 heroes = session.exec(select(Hero).offset(offset).limit(limit)).all() 1jklmnopqr
99 return heroes 1jklmnopqr
102@app.get("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
103def read_hero(*, session: Session = Depends(get_session), hero_id: int): 1abcdefghi
104 hero = session.get(Hero, hero_id) 1jklmnopqr
105 if not hero: 1jklmnopqr
106 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
107 return hero 1jklmnopqr
110@app.patch("/heroes/{hero_id}", response_model=HeroPublic) 1abcdefghi
111def update_hero( 1abcdefghi
112 *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
113):
114 db_hero = session.get(Hero, hero_id) 1jklmnopqr
115 if not db_hero: 1jklmnopqr
116 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
117 hero_data = hero.model_dump(exclude_unset=True) 1jklmnopqr
118 db_hero.sqlmodel_update(hero_data) 1jklmnopqr
119 session.add(db_hero) 1jklmnopqr
120 session.commit() 1jklmnopqr
121 session.refresh(db_hero) 1jklmnopqr
122 return db_hero 1jklmnopqr
125@app.delete("/heroes/{hero_id}") 1abcdefghi
126def delete_hero(*, session: Session = Depends(get_session), hero_id: int): 1abcdefghi
127 hero = session.get(Hero, hero_id) 1jklmnopqr
128 if not hero: 1jklmnopqr
129 raise HTTPException(status_code=404, detail="Hero not found") 1jklmnopqr
130 session.delete(hero) 1jklmnopqr
131 session.commit() 1jklmnopqr
132 return {"ok": True} 1jklmnopqr
135@app.post("/teams/", response_model=TeamPublic) 1abcdefghi
136def create_team(*, session: Session = Depends(get_session), team: TeamCreate): 1abcdefghi
137 db_team = Team.model_validate(team) 1jklmnopqr
138 session.add(db_team) 1jklmnopqr
139 session.commit() 1jklmnopqr
140 session.refresh(db_team) 1jklmnopqr
141 return db_team 1jklmnopqr
144@app.get("/teams/", response_model=list[TeamPublic]) 1abcdefghi
145def read_teams( 1abcdefghi
146 *,
147 session: Session = Depends(get_session),
148 offset: int = 0,
149 limit: int = Query(default=100, le=100),
150):
151 teams = session.exec(select(Team).offset(offset).limit(limit)).all() 1jklmnopqr
152 return teams 1jklmnopqr
155@app.get("/teams/{team_id}", response_model=TeamPublic) 1abcdefghi
156def read_team(*, team_id: int, session: Session = Depends(get_session)): 1abcdefghi
157 team = session.get(Team, team_id) 1jklmnopqr
158 if not team: 1jklmnopqr
159 raise HTTPException(status_code=404, detail="Team not found") 1jklmnopqr
160 return team 1jklmnopqr
163@app.patch("/teams/{team_id}", response_model=TeamPublic) 1abcdefghi
164def update_team( 1abcdefghi
165 *,
166 session: Session = Depends(get_session),
167 team_id: int,
168 team: TeamUpdate,
169):
170 db_team = session.get(Team, team_id) 1jklmnopqr
171 if not db_team: 1jklmnopqr
172 raise HTTPException(status_code=404, detail="Team not found") 1jklmnopqr
173 team_data = team.model_dump(exclude_unset=True) 1jklmnopqr
174 db_team.sqlmodel_update(team_data) 1jklmnopqr
175 session.add(db_team) 1jklmnopqr
176 session.commit() 1jklmnopqr
177 session.refresh(db_team) 1jklmnopqr
178 return db_team 1jklmnopqr
181@app.delete("/teams/{team_id}") 1abcdefghi
182def delete_team(*, session: Session = Depends(get_session), team_id: int): 1abcdefghi
183 team = session.get(Team, team_id) 1jklmnopqr
184 if not team: 1jklmnopqr
185 raise HTTPException(status_code=404, detail="Team not found") 1jklmnopqr
186 session.delete(team) 1jklmnopqr
187 session.commit() 1jklmnopqr
188 return {"ok": True} 1jklmnopqr