Add source examples for docs

This commit is contained in:
Sebastián Ramírez
2021-08-24 14:50:16 +02:00
parent e1cfd99a96
commit e6308146f7
126 changed files with 5778 additions and 0 deletions

View File

@ -0,0 +1,107 @@
from typing import List, Optional
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select
class HeroBase(SQLModel):
name: str
secret_name: str
age: Optional[int] = None
class Hero(HeroBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
class HeroCreate(HeroBase):
pass
class HeroRead(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, 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=HeroRead)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
db_hero = Hero.from_orm(hero)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroRead])
def read_heroes(
*,
session: Session = Depends(get_session),
offset: int = 0,
limit: int = Query(default=100, lte=100),
):
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroRead)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
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=HeroRead)
def update_hero(
*, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
):
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.dict(exclude_unset=True)
for key, value in hero_data.items():
setattr(db_hero, key, value)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
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}

View File

@ -0,0 +1,125 @@
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from .main import Hero, app, get_session
@pytest.fixture(name="session")
def session_fixture():
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
@pytest.fixture(name="client")
def client_fixture(session: Session):
def get_session_override():
return session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
yield client
app.dependency_overrides.clear()
def test_create_hero(client: TestClient):
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpond"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None
def test_create_hero_incomplete(client: TestClient):
# No secret_name
response = client.post("/heroes/", json={"name": "Deadpond"})
assert response.status_code == 422
def test_create_hero_invalid(client: TestClient):
# secret_name has an invalid type
response = client.post(
"/heroes/",
json={
"name": "Deadpond",
"secret_name": {"message": "Do you wanna know my secret identity?"},
},
)
assert response.status_code == 422
def test_read_heroes(session: Session, client: TestClient):
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
session.add(hero_1)
session.add(hero_2)
session.commit()
response = client.get("/heroes/")
data = response.json()
assert response.status_code == 200
assert len(data) == 2
assert data[0]["name"] == hero_1.name
assert data[0]["secret_name"] == hero_1.secret_name
assert data[0]["age"] == hero_1.age
assert data[0]["id"] == hero_1.id
assert data[1]["name"] == hero_2.name
assert data[1]["secret_name"] == hero_2.secret_name
assert data[1]["age"] == hero_2.age
assert data[1]["id"] == hero_2.id
def test_read_hero(session: Session, client: TestClient):
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
session.add(hero_1)
session.commit()
response = client.get(f"/heroes/{hero_1.id}")
data = response.json()
assert response.status_code == 200
assert data["name"] == hero_1.name
assert data["secret_name"] == hero_1.secret_name
assert data["age"] == hero_1.age
assert data["id"] == hero_1.id
def test_update_hero(session: Session, client: TestClient):
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
session.add(hero_1)
session.commit()
response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpuddle"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] == hero_1.id
def test_delete_hero(session: Session, client: TestClient):
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
session.add(hero_1)
session.commit()
response = client.delete(f"/heroes/{hero_1.id}")
hero_in_db = session.get(Hero, hero_1.id)
assert response.status_code == 200
assert hero_in_db is None

View File

@ -0,0 +1,32 @@
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from .main import app, get_session # (1)
def test_create_hero():
engine = create_engine(
"sqlite:///testing.db", connect_args={"check_same_thread": False}
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
def get_session_override():
return session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app) # (2)
response = client.post( # (3)
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
app.dependency_overrides.clear()
data = response.json() # (4)
assert response.status_code == 200 # (5)
assert data["name"] == "Deadpond" # (6)
assert data["secret_name"] == "Dive Wilson" # (7)
assert data["age"] is None # (8)
assert data["id"] is not None # (9)

View File

@ -0,0 +1,32 @@
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from .main import app, get_session # (1)
def test_create_hero():
engine = create_engine(
"sqlite:///testing.db", connect_args={"check_same_thread": False}
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
def get_session_override(): # (2)
return session # (3)
app.dependency_overrides[get_session] = get_session_override # (4)
client = TestClient(app)
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
app.dependency_overrides.clear() # (5)
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpond"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None

View File

@ -0,0 +1,33 @@
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from .main import app, get_session # (1)
def test_create_hero():
engine = create_engine( # (2)
"sqlite:///testing.db", connect_args={"check_same_thread": False}
)
SQLModel.metadata.create_all(engine) # (3)
with Session(engine) as session: # (4)
def get_session_override():
return session # (5)
app.dependency_overrides[get_session] = get_session_override # (4)
client = TestClient(app)
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
app.dependency_overrides.clear()
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpond"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None
# (6)

View File

@ -0,0 +1,35 @@
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool # (1)
from .main import app, get_session
def test_create_hero():
engine = create_engine(
"sqlite://", # (2)
connect_args={"check_same_thread": False},
poolclass=StaticPool, # (3)
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
def get_session_override():
return session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
app.dependency_overrides.clear()
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpond"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None

View File

@ -0,0 +1,37 @@
import pytest # (1)
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from .main import app, get_session
@pytest.fixture(name="session") # (2)
def session_fixture(): # (3)
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session # (4)
def test_create_hero(session: Session): # (5)
def get_session_override():
return session # (6)
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
app.dependency_overrides.clear()
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpond"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None

View File

@ -0,0 +1,41 @@
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from .main import app, get_session
@pytest.fixture(name="session")
def session_fixture():
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
@pytest.fixture(name="client") # (1)
def client_fixture(session: Session): # (2)
def get_session_override(): # (3)
return session
app.dependency_overrides[get_session] = get_session_override # (4)
client = TestClient(app) # (5)
yield client # (6)
app.dependency_overrides.clear() # (7)
def test_create_hero(client: TestClient): # (8)
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
data = response.json()
assert response.status_code == 200
assert data["name"] == "Deadpond"
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None