FastAPI

project
└── app
    ├── __init__.py
    ├── main.py
    ├── tests.py
    ├── config.py
    ├── database.py
    ├── users
    │   ├── __init__.py
    │   ├── api.py
    │   ├── models.py
    │   ├── routers.py
    │   └── schemas.py
    ├── items
    │   ├── __init__.py
    │   ├── routers.py
    │   ├── schemas.py
    │   └── templates
    │       └── item.html
    └── static
        └── simple.css

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"message": "Hello World!"}

uvicorn app.main:app --reload

from .items.routers import router as items_router

app = FastAPI(
    title="fastapi-sandbox",
    description="A sandbox to play with *FastAPI* framework.",
    openapi_tags=[{"name": "items", "description": "Manage *items*."}]
)

app.include_router(items_router)

from datetime import datetime
from typing import Annotated
from fastapi import APIRouter, Request, Body, Cookie, File, Form, Header, Path, Query, UploadFile
from .schemas import Item

router = APIRouter(
    prefix="/items",
    tags=["items"],
)

@router.get("/{item_id}")
async def read_item(item_id: int, item: Item) -> Item:
    """
    The function return type can be defined as follows:
    - builtin type
    - Pydantic model
    - Response
    - PlainTextResponse
    - HtmlResponse
    - JSONResponse
    - RedirectResponse

    Default response class JSONResponse can be overridden with:
    - app = FastAPI(default_response_class=HtmlResponse)

    The path decorator's parameter `response_class` define response class:
    - @app.get("/items/", response_class=HTMLResponse)

    The path decorator's parameter `response_model` define response model:
    - @app.get("/portal", response_model=None)
    - @app.get("/items/", response_model=list[Item])
    - @app.get("/user/", response_model=BaseUser, response_model_exclude_unset=True)

    The path decorator's parameter `status_code` define response status code:
    - @app.post("/items/", status_code=201)
    - @app.post("/items/", status_code=status.HTTP_201_CREATED)

    The path decorator's parameters `tags`, `summary` and `description` are used in interactive docs:
    - @app.get("/items/", tags=["items"], summary="Create an item", description="Create an item description")

    Exception can be defined as follows:
    - raise HTTPException(status_code=404, detail="Item not found")
    """
    return item

@router.put("/{item_id}")
async def create_item(item_id: int, item: Item, item_query: str):
    """
    The function parameters will be recognized as follows:
    - If the parameter is also declared in the path, it will be used as a path parameter: `item_id`
    - If the parameter is of a builtin type, it will be interpreted as a query parameter: `item_query`
    - If the parameter is of a Pydantic model type, it will be interpreted as a request body (json string): `item`
    """
    result = {"item_id": item_id, **item.dict()}
    if item_query:
        result.update({"item_query": item_query})
    return result

@router.patch("/{item_id}")
async def update_item(
        item_id: Annotated[int, Path(title="Item ID", gt=0, le=100)],
        item_query: Annotated[str | None, Query(alias="item-query", title="Query string",
        description="Query string for the items to search",
        max_length=10, regex=r"^\w+$", deprecated=True)] = None,
        item: Item = None,
        available: Annotated[bool, Body()] = None,
        date: Annotated[datetime, Cookie()] = datetime.now(),
        user_agent: Annotated[str | None, Header()] = None):
    """
    The function parameters can be defined as follows:
    - `Path`: a path parameter
    - `Query`: a query parameter
    - `Body`: a request body
    - `Cookie`: a cookie parameter
    - `Header`: a header parameter

    See also: [Dependency Injection](https://fastapi.tiangolo.com/tutorial/dependencies/) system (function or class).
    """
    result = {"item_id": item_id, "item": item, "available": available, "date": date, "User-Agent": user_agent}
    if item_query:
        result.update({"item_query": item_query})
    return result

@router.post("/file/", tags=["file"])
async def create_file(
        token: Annotated[str, Form()],
        file: Annotated[bytes | None, File()] = None,
        upload_file: UploadFile | None = None):
    """
    The function parameters can be defined as follows:
    - `Form`: a form parameter
    - `File`: a file parameter
    - `UploadFile`: a spooled file parameter

    You can declare multiple File and Form parameters, but you can't also declare Body fields.
    """
    return {"token": token, "file": file, "upload_file": upload_file}

from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator

class Image(BaseModel):
    url: HttpUrl
    name: str

class Item(BaseModel):
    name: str
    description: str | None = Field(title="Description", max_length=300)
    price: float = Field(lt=100, description="The price must be lesser than 100")
    tags: set[str] = set()
    image: Image | None = None

    @field_validator("price")
    def price_must_be_positive(cls, value):
        if value <= 0:
            raise ValueError(f"The price must be greater than zero, received: {value}")
        return value

    model_config = ConfigDict(json_schema_extra={
        "example": {
            "name": "Foo",
            "description": "Foo description",
            "price": 42.0,
            "tags": [
                "foo",
                "bar",
                "baz"
            ],
            "image": {
                "url": "http://example.com/foo.jpg",
                "name": "The Foo"
            }
        }
    })

from functools import lru_cache
from pydantic import EmailStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    app_name: str = "fastapi-sandbox"
    admin_email: EmailStr = "admin@example.com"
    database_url: str = "sqlite:///db.sqlite3"

    model_config = SettingsConfigDict(env_file=".env")

@lru_cache()
def get_settings():
    return Settings()

settings = get_settings()

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="app/static"), name="static")

from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="app/items/templates")

@router.get("/{item_id}", response_class=HTMLResponse)
async def read_item(request: Request, item_id: int):
    return templates.TemplateResponse("item.html", {"request": request, "item_id": item_id})

<!DOCTYPE html>
<html lang="fr">
<head>
    <meta charset="UTF-8">
    <title>Item Details</title>
    <link rel="stylesheet" href="{{ url_for('static', path='/simple.css') }}">
</head>
<body>
    <h1>Item ID: {{ item_id }}</h1>
</body>
</html>

from fastapi import BackgroundTasks

def write_notification(email: str, message=""):
    print(f"notification for {email}: {message}")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

from .users.routers import router as users_router

app.include_router(users_router)

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from .config import settings

engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import database
from . import api, models, schemas

models.Base.metadata.create_all(bind=database.engine)

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

router = APIRouter(
    prefix="/users",
    tags=["users"],
)

@router.post("/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = api.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return api.create_user(db=db, user=user)

@router.get("/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = api.get_users(db, skip=skip, limit=limit)
    return users

@router.get("/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = api.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@router.post("/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return api.create_user_item(db=db, item=item, user_id=user_id)

@router.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = api.get_items(db, skip=skip, limit=limit)
    return items

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

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=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, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

from pydantic import BaseModel, ConfigDict, EmailStr

class ItemBase(BaseModel):
    title: str
    description: str | None = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    model_config = ConfigDict(from_attributes=True)

class UserBase(BaseModel):
    email: EmailStr

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    model_config = ConfigDict(from_attributes=True)

pytest app/tests.py

from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_root_endpoint():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World!"}

def test_items_template():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.template.name == 'item.html'
    assert "request" in response.context

def test_correct_user():
    json = {"email": "myuser@example.com", "password": "mypassword"}
    response = client.post("/users/", json=json)
    assert response.status_code == 200

def test_wrong_user():
    json = {"email": "myuser@example.com"}
    response = client.post("/users/", json=json)
    assert response.status_code != 200