mirror of
https://github.com/fastapi/fastapi.git
synced 2026-04-08 09:00:26 -04:00
✨ Add new tutorial for SQL databases with SQLModel (#12285)
This commit is contained in:
Sebastián Ramírez
committed by
GitHub
GitHub
parent
13a18f80b3
commit
7daaac2bc3
@@ -1,62 +0,0 @@
|
||||
from typing import List
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, Request, Response
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud, models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def db_session_middleware(request: Request, call_next):
|
||||
response = Response("Internal server error", status_code=500)
|
||||
try:
|
||||
request.state.db = SessionLocal()
|
||||
response = await call_next(request)
|
||||
finally:
|
||||
request.state.db.close()
|
||||
return response
|
||||
|
||||
|
||||
# Dependency
|
||||
def get_db(request: Request):
|
||||
return request.state.db
|
||||
|
||||
|
||||
@app.post("/users/", response_model=schemas.User)
|
||||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db, user=user)
|
||||
|
||||
|
||||
@app.get("/users/", response_model=List[schemas.User])
|
||||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
users = crud.get_users(db, skip=skip, limit=limit)
|
||||
return users
|
||||
|
||||
|
||||
@app.get("/users/{user_id}", response_model=schemas.User)
|
||||
def read_user(user_id: int, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user(db, user_id=user_id)
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
|
||||
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
||||
def create_item_for_user(
|
||||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
return crud.create_user_item(db=db, item=item, user_id=user_id)
|
||||
|
||||
|
||||
@app.get("/items/", response_model=List[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = crud.get_items(db, skip=skip, limit=limit)
|
||||
return items
|
||||
@@ -1,36 +0,0 @@
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import models, schemas
|
||||
|
||||
|
||||
def get_user(db: Session, user_id: int):
|
||||
return db.query(models.User).filter(models.User.id == user_id).first()
|
||||
|
||||
|
||||
def get_user_by_email(db: Session, email: str):
|
||||
return db.query(models.User).filter(models.User.email == email).first()
|
||||
|
||||
|
||||
def get_users(db: Session, skip: int = 0, limit: int = 100):
|
||||
return db.query(models.User).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
def create_user(db: Session, user: schemas.UserCreate):
|
||||
fake_hashed_password = user.password + "notreallyhashed"
|
||||
db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
|
||||
db.add(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
|
||||
def get_items(db: Session, skip: int = 0, limit: int = 100):
|
||||
return db.query(models.Item).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
|
||||
db_item = models.Item(**item.dict(), owner_id=user_id)
|
||||
db.add(db_item)
|
||||
db.commit()
|
||||
db.refresh(db_item)
|
||||
return db_item
|
||||
@@ -1,13 +0,0 @@
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
|
||||
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
||||
)
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
Base = declarative_base()
|
||||
@@ -1,55 +0,0 @@
|
||||
from typing import List
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud, models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
# Dependency
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post("/users/", response_model=schemas.User)
|
||||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db, user=user)
|
||||
|
||||
|
||||
@app.get("/users/", response_model=List[schemas.User])
|
||||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
users = crud.get_users(db, skip=skip, limit=limit)
|
||||
return users
|
||||
|
||||
|
||||
@app.get("/users/{user_id}", response_model=schemas.User)
|
||||
def read_user(user_id: int, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user(db, user_id=user_id)
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
|
||||
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
||||
def create_item_for_user(
|
||||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
return crud.create_user_item(db=db, item=item, user_id=user_id)
|
||||
|
||||
|
||||
@app.get("/items/", response_model=List[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = crud.get_items(db, skip=skip, limit=limit)
|
||||
return items
|
||||
@@ -1,26 +0,0 @@
|
||||
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from .database import Base
|
||||
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
email = Column(String, unique=True, index=True)
|
||||
hashed_password = Column(String)
|
||||
is_active = Column(Boolean, default=True)
|
||||
|
||||
items = relationship("Item", back_populates="owner")
|
||||
|
||||
|
||||
class Item(Base):
|
||||
__tablename__ = "items"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
title = Column(String, index=True)
|
||||
description = Column(String, index=True)
|
||||
owner_id = Column(Integer, ForeignKey("users.id"))
|
||||
|
||||
owner = relationship("User", back_populates="items")
|
||||
@@ -1,37 +0,0 @@
|
||||
from typing import List, Union
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class ItemBase(BaseModel):
|
||||
title: str
|
||||
description: Union[str, None] = None
|
||||
|
||||
|
||||
class ItemCreate(ItemBase):
|
||||
pass
|
||||
|
||||
|
||||
class Item(ItemBase):
|
||||
id: int
|
||||
owner_id: int
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
class UserBase(BaseModel):
|
||||
email: str
|
||||
|
||||
|
||||
class UserCreate(UserBase):
|
||||
password: str
|
||||
|
||||
|
||||
class User(UserBase):
|
||||
id: int
|
||||
is_active: bool
|
||||
items: List[Item] = []
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
@@ -1,50 +0,0 @@
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from ..database import Base
|
||||
from ..main import app, get_db
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite://"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
|
||||
def override_get_db():
|
||||
try:
|
||||
db = TestingSessionLocal()
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_create_user():
|
||||
response = client.post(
|
||||
"/users/",
|
||||
json={"email": "deadpool@example.com", "password": "chimichangas4life"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["email"] == "deadpool@example.com"
|
||||
assert "id" in data
|
||||
user_id = data["id"]
|
||||
|
||||
response = client.get(f"/users/{user_id}")
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["email"] == "deadpool@example.com"
|
||||
assert data["id"] == user_id
|
||||
@@ -1,60 +0,0 @@
|
||||
from fastapi import Depends, FastAPI, HTTPException, Request, Response
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud, models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def db_session_middleware(request: Request, call_next):
|
||||
response = Response("Internal server error", status_code=500)
|
||||
try:
|
||||
request.state.db = SessionLocal()
|
||||
response = await call_next(request)
|
||||
finally:
|
||||
request.state.db.close()
|
||||
return response
|
||||
|
||||
|
||||
# Dependency
|
||||
def get_db(request: Request):
|
||||
return request.state.db
|
||||
|
||||
|
||||
@app.post("/users/", response_model=schemas.User)
|
||||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db, user=user)
|
||||
|
||||
|
||||
@app.get("/users/", response_model=list[schemas.User])
|
||||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
users = crud.get_users(db, skip=skip, limit=limit)
|
||||
return users
|
||||
|
||||
|
||||
@app.get("/users/{user_id}", response_model=schemas.User)
|
||||
def read_user(user_id: int, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user(db, user_id=user_id)
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
|
||||
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
||||
def create_item_for_user(
|
||||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
return crud.create_user_item(db=db, item=item, user_id=user_id)
|
||||
|
||||
|
||||
@app.get("/items/", response_model=list[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = crud.get_items(db, skip=skip, limit=limit)
|
||||
return items
|
||||
@@ -1,36 +0,0 @@
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import models, schemas
|
||||
|
||||
|
||||
def get_user(db: Session, user_id: int):
|
||||
return db.query(models.User).filter(models.User.id == user_id).first()
|
||||
|
||||
|
||||
def get_user_by_email(db: Session, email: str):
|
||||
return db.query(models.User).filter(models.User.email == email).first()
|
||||
|
||||
|
||||
def get_users(db: Session, skip: int = 0, limit: int = 100):
|
||||
return db.query(models.User).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
def create_user(db: Session, user: schemas.UserCreate):
|
||||
fake_hashed_password = user.password + "notreallyhashed"
|
||||
db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
|
||||
db.add(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
|
||||
def get_items(db: Session, skip: int = 0, limit: int = 100):
|
||||
return db.query(models.Item).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
|
||||
db_item = models.Item(**item.dict(), owner_id=user_id)
|
||||
db.add(db_item)
|
||||
db.commit()
|
||||
db.refresh(db_item)
|
||||
return db_item
|
||||
@@ -1,13 +0,0 @@
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
|
||||
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
||||
)
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
Base = declarative_base()
|
||||
@@ -1,53 +0,0 @@
|
||||
from fastapi import Depends, FastAPI, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud, models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
# Dependency
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post("/users/", response_model=schemas.User)
|
||||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db, user=user)
|
||||
|
||||
|
||||
@app.get("/users/", response_model=list[schemas.User])
|
||||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
users = crud.get_users(db, skip=skip, limit=limit)
|
||||
return users
|
||||
|
||||
|
||||
@app.get("/users/{user_id}", response_model=schemas.User)
|
||||
def read_user(user_id: int, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user(db, user_id=user_id)
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
|
||||
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
||||
def create_item_for_user(
|
||||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
return crud.create_user_item(db=db, item=item, user_id=user_id)
|
||||
|
||||
|
||||
@app.get("/items/", response_model=list[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = crud.get_items(db, skip=skip, limit=limit)
|
||||
return items
|
||||
@@ -1,26 +0,0 @@
|
||||
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from .database import Base
|
||||
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
email = Column(String, unique=True, index=True)
|
||||
hashed_password = Column(String)
|
||||
is_active = Column(Boolean, default=True)
|
||||
|
||||
items = relationship("Item", back_populates="owner")
|
||||
|
||||
|
||||
class Item(Base):
|
||||
__tablename__ = "items"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
title = Column(String, index=True)
|
||||
description = Column(String, index=True)
|
||||
owner_id = Column(Integer, ForeignKey("users.id"))
|
||||
|
||||
owner = relationship("User", back_populates="items")
|
||||
@@ -1,35 +0,0 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class ItemBase(BaseModel):
|
||||
title: str
|
||||
description: str | None = None
|
||||
|
||||
|
||||
class ItemCreate(ItemBase):
|
||||
pass
|
||||
|
||||
|
||||
class Item(ItemBase):
|
||||
id: int
|
||||
owner_id: int
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
class UserBase(BaseModel):
|
||||
email: str
|
||||
|
||||
|
||||
class UserCreate(UserBase):
|
||||
password: str
|
||||
|
||||
|
||||
class User(UserBase):
|
||||
id: int
|
||||
is_active: bool
|
||||
items: list[Item] = []
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
@@ -1,47 +0,0 @@
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from ..database import Base
|
||||
from ..main import app, get_db
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
||||
)
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
|
||||
def override_get_db():
|
||||
try:
|
||||
db = TestingSessionLocal()
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_create_user():
|
||||
response = client.post(
|
||||
"/users/",
|
||||
json={"email": "deadpool@example.com", "password": "chimichangas4life"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["email"] == "deadpool@example.com"
|
||||
assert "id" in data
|
||||
user_id = data["id"]
|
||||
|
||||
response = client.get(f"/users/{user_id}")
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["email"] == "deadpool@example.com"
|
||||
assert data["id"] == user_id
|
||||
@@ -1,60 +0,0 @@
|
||||
from fastapi import Depends, FastAPI, HTTPException, Request, Response
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud, models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def db_session_middleware(request: Request, call_next):
|
||||
response = Response("Internal server error", status_code=500)
|
||||
try:
|
||||
request.state.db = SessionLocal()
|
||||
response = await call_next(request)
|
||||
finally:
|
||||
request.state.db.close()
|
||||
return response
|
||||
|
||||
|
||||
# Dependency
|
||||
def get_db(request: Request):
|
||||
return request.state.db
|
||||
|
||||
|
||||
@app.post("/users/", response_model=schemas.User)
|
||||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db, user=user)
|
||||
|
||||
|
||||
@app.get("/users/", response_model=list[schemas.User])
|
||||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
users = crud.get_users(db, skip=skip, limit=limit)
|
||||
return users
|
||||
|
||||
|
||||
@app.get("/users/{user_id}", response_model=schemas.User)
|
||||
def read_user(user_id: int, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user(db, user_id=user_id)
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
|
||||
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
||||
def create_item_for_user(
|
||||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
return crud.create_user_item(db=db, item=item, user_id=user_id)
|
||||
|
||||
|
||||
@app.get("/items/", response_model=list[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = crud.get_items(db, skip=skip, limit=limit)
|
||||
return items
|
||||
@@ -1,36 +0,0 @@
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import models, schemas
|
||||
|
||||
|
||||
def get_user(db: Session, user_id: int):
|
||||
return db.query(models.User).filter(models.User.id == user_id).first()
|
||||
|
||||
|
||||
def get_user_by_email(db: Session, email: str):
|
||||
return db.query(models.User).filter(models.User.email == email).first()
|
||||
|
||||
|
||||
def get_users(db: Session, skip: int = 0, limit: int = 100):
|
||||
return db.query(models.User).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
def create_user(db: Session, user: schemas.UserCreate):
|
||||
fake_hashed_password = user.password + "notreallyhashed"
|
||||
db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
|
||||
db.add(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
|
||||
def get_items(db: Session, skip: int = 0, limit: int = 100):
|
||||
return db.query(models.Item).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
|
||||
db_item = models.Item(**item.dict(), owner_id=user_id)
|
||||
db.add(db_item)
|
||||
db.commit()
|
||||
db.refresh(db_item)
|
||||
return db_item
|
||||
@@ -1,13 +0,0 @@
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
|
||||
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
||||
)
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
Base = declarative_base()
|
||||
@@ -1,53 +0,0 @@
|
||||
from fastapi import Depends, FastAPI, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud, models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
# Dependency
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post("/users/", response_model=schemas.User)
|
||||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db, user=user)
|
||||
|
||||
|
||||
@app.get("/users/", response_model=list[schemas.User])
|
||||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
users = crud.get_users(db, skip=skip, limit=limit)
|
||||
return users
|
||||
|
||||
|
||||
@app.get("/users/{user_id}", response_model=schemas.User)
|
||||
def read_user(user_id: int, db: Session = Depends(get_db)):
|
||||
db_user = crud.get_user(db, user_id=user_id)
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
|
||||
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
||||
def create_item_for_user(
|
||||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
return crud.create_user_item(db=db, item=item, user_id=user_id)
|
||||
|
||||
|
||||
@app.get("/items/", response_model=list[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = crud.get_items(db, skip=skip, limit=limit)
|
||||
return items
|
||||
@@ -1,26 +0,0 @@
|
||||
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from .database import Base
|
||||
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
email = Column(String, unique=True, index=True)
|
||||
hashed_password = Column(String)
|
||||
is_active = Column(Boolean, default=True)
|
||||
|
||||
items = relationship("Item", back_populates="owner")
|
||||
|
||||
|
||||
class Item(Base):
|
||||
__tablename__ = "items"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
title = Column(String, index=True)
|
||||
description = Column(String, index=True)
|
||||
owner_id = Column(Integer, ForeignKey("users.id"))
|
||||
|
||||
owner = relationship("User", back_populates="items")
|
||||
@@ -1,37 +0,0 @@
|
||||
from typing import Union
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class ItemBase(BaseModel):
|
||||
title: str
|
||||
description: Union[str, None] = None
|
||||
|
||||
|
||||
class ItemCreate(ItemBase):
|
||||
pass
|
||||
|
||||
|
||||
class Item(ItemBase):
|
||||
id: int
|
||||
owner_id: int
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
class UserBase(BaseModel):
|
||||
email: str
|
||||
|
||||
|
||||
class UserCreate(UserBase):
|
||||
password: str
|
||||
|
||||
|
||||
class User(UserBase):
|
||||
id: int
|
||||
is_active: bool
|
||||
items: list[Item] = []
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
@@ -1,47 +0,0 @@
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from ..database import Base
|
||||
from ..main import app, get_db
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
||||
)
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
|
||||
def override_get_db():
|
||||
try:
|
||||
db = TestingSessionLocal()
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_create_user():
|
||||
response = client.post(
|
||||
"/users/",
|
||||
json={"email": "deadpool@example.com", "password": "chimichangas4life"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["email"] == "deadpool@example.com"
|
||||
assert "id" in data
|
||||
user_id = data["id"]
|
||||
|
||||
response = client.get(f"/users/{user_id}")
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["email"] == "deadpool@example.com"
|
||||
assert data["id"] == user_id
|
||||
71
docs_src/sql_databases/tutorial001.py
Normal file
71
docs_src/sql_databases/tutorial001.py
Normal file
@@ -0,0 +1,71 @@
|
||||
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
Normal file
74
docs_src/sql_databases/tutorial001_an.py
Normal file
@@ -0,0 +1,74 @@
|
||||
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}
|
||||
73
docs_src/sql_databases/tutorial001_an_py310.py
Normal file
73
docs_src/sql_databases/tutorial001_an_py310.py
Normal file
@@ -0,0 +1,73 @@
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, Query
|
||||
from sqlmodel import Field, Session, SQLModel, create_engine, select
|
||||
|
||||
|
||||
class Hero(SQLModel, table=True):
|
||||
id: int | None = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
age: 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}
|
||||
73
docs_src/sql_databases/tutorial001_an_py39.py
Normal file
73
docs_src/sql_databases/tutorial001_an_py39.py
Normal file
@@ -0,0 +1,73 @@
|
||||
from typing import Annotated, 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
|
||||
|
||||
|
||||
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}
|
||||
69
docs_src/sql_databases/tutorial001_py310.py
Normal file
69
docs_src/sql_databases/tutorial001_py310.py
Normal file
@@ -0,0 +1,69 @@
|
||||
from fastapi import Depends, FastAPI, HTTPException, Query
|
||||
from sqlmodel import Field, Session, SQLModel, create_engine, select
|
||||
|
||||
|
||||
class Hero(SQLModel, table=True):
|
||||
id: int | None = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
age: 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}
|
||||
71
docs_src/sql_databases/tutorial001_py39.py
Normal file
71
docs_src/sql_databases/tutorial001_py39.py
Normal file
@@ -0,0 +1,71 @@
|
||||
from typing import 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}
|
||||
104
docs_src/sql_databases/tutorial002.py
Normal file
104
docs_src/sql_databases/tutorial002.py
Normal file
@@ -0,0 +1,104 @@
|
||||
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
Normal file
104
docs_src/sql_databases/tutorial002_an.py
Normal file
@@ -0,0 +1,104 @@
|
||||
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}
|
||||
103
docs_src/sql_databases/tutorial002_an_py310.py
Normal file
103
docs_src/sql_databases/tutorial002_an_py310.py
Normal file
@@ -0,0 +1,103 @@
|
||||
from typing import Annotated
|
||||
|
||||
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: int | None = Field(default=None, index=True)
|
||||
|
||||
|
||||
class Hero(HeroBase, table=True):
|
||||
id: 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: str | None = None
|
||||
age: int | None = None
|
||||
secret_name: 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}
|
||||
103
docs_src/sql_databases/tutorial002_an_py39.py
Normal file
103
docs_src/sql_databases/tutorial002_an_py39.py
Normal file
@@ -0,0 +1,103 @@
|
||||
from typing import Annotated, 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
|
||||
|
||||
|
||||
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}
|
||||
102
docs_src/sql_databases/tutorial002_py310.py
Normal file
102
docs_src/sql_databases/tutorial002_py310.py
Normal file
@@ -0,0 +1,102 @@
|
||||
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: int | None = Field(default=None, index=True)
|
||||
|
||||
|
||||
class Hero(HeroBase, table=True):
|
||||
id: 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: str | None = None
|
||||
age: int | None = None
|
||||
secret_name: 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_py39.py
Normal file
104
docs_src/sql_databases/tutorial002_py39.py
Normal file
@@ -0,0 +1,104 @@
|
||||
from typing import 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}
|
||||
Reference in New Issue
Block a user