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