mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Remove code examples for Python 3.8 in `sql_databases`

pull/14510/head
Yurii Motov 7 months ago
parent
3c84819842
commit
2108e84959
6 changed files with 6 additions and 363 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 71
      docs_src/sql_databases/tutorial001.py
  2. 74
      docs_src/sql_databases/tutorial001_an.py
  3. 104
      docs_src/sql_databases/tutorial002.py
  4. 104
      docs_src/sql_databases/tutorial002_an.py
  5. 8
      tests/test_tutorial/test_sql_databases/test_tutorial001.py
  6. 8
      tests/test_tutorial/test_sql_databases/test_tutorial002.py

71
docs_src/sql_databases/tutorial001.py
View File

@ -1,71 +0,0 @@
from typing import List, Union
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Union[int, None] = Field(default=None, primary_key=True)
name: str = Field(index=True)
age: Union[int, None] = Field(default=None, index=True)
secret_name: str
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/")
def create_hero(hero: Hero, session: Session = Depends(get_session)) -> Hero:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes(
session: Session = Depends(get_session),
offset: int = 0,
limit: int = Query(default=100, le=100),
) -> List[Hero]:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: Session = Depends(get_session)) -> Hero:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: Session = Depends(get_session)):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
session.delete(hero)
session.commit()
return {"ok": True}

74
docs_src/sql_databases/tutorial001_an.py
View File

@ -1,74 +0,0 @@
from typing import List, Union
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing_extensions import Annotated
class Hero(SQLModel, table=True):
id: Union[int, None] = Field(default=None, primary_key=True)
name: str = Field(index=True)
age: Union[int, None] = Field(default=None, index=True)
secret_name: str
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes(
session: SessionDep,
offset: int = 0,
limit: Annotated[int, Query(le=100)] = 100,
) -> List[Hero]:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
session.delete(hero)
session.commit()
return {"ok": True}

104
docs_src/sql_databases/tutorial002.py
View File

@ -1,104 +0,0 @@
from typing import List, Union
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select
class HeroBase(SQLModel):
name: str = Field(index=True)
age: Union[int, None] = Field(default=None, index=True)
class Hero(HeroBase, table=True):
id: Union[int, None] = Field(default=None, primary_key=True)
secret_name: str
class HeroPublic(HeroBase):
id: int
class HeroCreate(HeroBase):
secret_name: str
class HeroUpdate(HeroBase):
name: Union[str, None] = None
age: Union[int, None] = None
secret_name: Union[str, None] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: Session = Depends(get_session)):
db_hero = Hero.model_validate(hero)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(
session: Session = Depends(get_session),
offset: int = 0,
limit: int = Query(default=100, le=100),
):
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: Session = Depends(get_session)):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(
hero_id: int, hero: HeroUpdate, session: Session = Depends(get_session)
):
hero_db = session.get(Hero, hero_id)
if not hero_db:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
hero_db.sqlmodel_update(hero_data)
session.add(hero_db)
session.commit()
session.refresh(hero_db)
return hero_db
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: Session = Depends(get_session)):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
session.delete(hero)
session.commit()
return {"ok": True}

104
docs_src/sql_databases/tutorial002_an.py
View File

@ -1,104 +0,0 @@
from typing import List, Union
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing_extensions import Annotated
class HeroBase(SQLModel):
name: str = Field(index=True)
age: Union[int, None] = Field(default=None, index=True)
class Hero(HeroBase, table=True):
id: Union[int, None] = Field(default=None, primary_key=True)
secret_name: str
class HeroPublic(HeroBase):
id: int
class HeroCreate(HeroBase):
secret_name: str
class HeroUpdate(HeroBase):
name: Union[str, None] = None
age: Union[int, None] = None
secret_name: Union[str, None] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
db_hero = Hero.model_validate(hero)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(
session: SessionDep,
offset: int = 0,
limit: Annotated[int, Query(le=100)] = 100,
):
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
hero_db = session.get(Hero, hero_id)
if not hero_db:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
hero_db.sqlmodel_update(hero_data)
session.add(hero_db)
session.commit()
session.refresh(hero_db)
return hero_db
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
session.delete(hero)
session.commit()
return {"ok": True}

8
tests/test_tutorial/test_sql_databases/test_tutorial001.py
View File

@ -9,7 +9,7 @@ from sqlalchemy import StaticPool
from sqlmodel import SQLModel, create_engine
from sqlmodel.main import default_registry
from tests.utils import needs_py39, needs_py310
from tests.utils import needs_py310
def clear_sqlmodel():
@ -22,11 +22,9 @@ def clear_sqlmodel():
@pytest.fixture(
name="client",
params=[
"tutorial001",
pytest.param("tutorial001_py39", marks=needs_py39),
pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
"tutorial001_an",
pytest.param("tutorial001_an_py39", marks=needs_py39),
pytest.param("tutorial001_an_py39"),
pytest.param("tutorial001_an_py310", marks=needs_py310),
],
)

8
tests/test_tutorial/test_sql_databases/test_tutorial002.py
View File

@ -9,7 +9,7 @@ from sqlalchemy import StaticPool
from sqlmodel import SQLModel, create_engine
from sqlmodel.main import default_registry
from tests.utils import needs_py39, needs_py310
from tests.utils import needs_py310
def clear_sqlmodel():
@ -22,11 +22,9 @@ def clear_sqlmodel():
@pytest.fixture(
name="client",
params=[
"tutorial002",
pytest.param("tutorial002_py39", marks=needs_py39),
pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
"tutorial002_an",
pytest.param("tutorial002_an_py39", marks=needs_py39),
pytest.param("tutorial002_an_py39"),
pytest.param("tutorial002_an_py310", marks=needs_py310),
],
)

Write Preview
Loading…
Cancel
Save