fastapi-best-practices

@app.get("/users")
def get_users():
    users = db.query(User).all()
    return users

@app.get("/data")
async def get_data():
    result = heavy_computation()
    return result

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

@app.get("/data")
def get_data():
    return heavy_computation()

db = get_database()

@app.get("/items")
async def get_items():
    return db.query(Item).all()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
async def get_items(db: Annotated[Session, Depends(get_db)]):
    return db.query(Item).all()

from pydantic import validator

class Item(BaseModel):
    name: str
    price: float

    class Config:
        orm_mode = True

    @validator("price")
    def price_positive(cls, v):
        if v <= 0:
            raise ValueError("must be positive")
        return v

from pydantic import BaseModel, field_validator, ConfigDict

class Item(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    name: str
    price: float

    @field_validator("price")
    @classmethod
    def price_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("must be positive")
        return v

@app.on_event("startup")
async def startup():
    app.state.db = await create_pool()

@app.on_event("shutdown")
async def shutdown():
    await app.state.db.close()

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db = await create_pool()
    yield
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

@app.post("/send-email")
async def send_email(email: str):
    asyncio.create_task(send_email_async(email))
    return {"status": "queued"}

@app.post("/send-email")
async def send_email(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_async, email)
    return {"status": "queued"}

# main.py - 500 lines of routes
@app.get("/users")
@app.get("/users/{id}")
@app.post("/items")
@app.get("/items")

# main.py
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

# routers/users.py
router = APIRouter()
@router.get("/")
@router.get("/{id}")

@app.get("/items/{id}")
async def get_item(id: int):
    item = await db.get(Item, id)
    return {"id": item.id, "name": item.name}

@app.get("/items/{id}", response_model=ItemOut)
async def get_item(id: int, db: Session = Depends(get_db)):
    item = await db.get(Item, id)
    if not item:
        raise HTTPException(status_code=404)
    return item

raise HTTPException(status_code=404, detail="Not found")
raise HTTPException(status_code=401, detail="Unauthorized")

from fastapi import status

raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")

@app.get("/me")
async def read_me(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/me")
async def read_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///db.sqlite")

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///db.sqlite"
    debug: bool = False
    model_config = {"env_file": ".env"}

settings = Settings()