Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis

💻 Exemplo 1 — Dependência de Sessão Assíncrona Integrada ao FastAPI

from typing import Annotated, AsyncIterator
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import SQLModel, Field, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

class Produto(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    nome: str
    preco: float

engine = create_async_engine("sqlite+aiosqlite:///./catalogo.db")
app = FastAPI()

async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSession(engine) as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

@app.on_event("startup")
async def on_startup() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

@app.post("/produtos", status_code=201)
async def criar_produto(nome: str, preco: float, session: SessionDep) -> Produto:
    produto = Produto(nome=nome, preco=preco)
    session.add(produto)
    await session.commit()
    await session.refresh(produto)
    return produto

@app.get("/produtos/{produto_id}")
async def obter_produto(produto_id: int, session: SessionDep) -> Produto:
    produto = await session.get(Produto, produto_id)
    if produto is None:
        raise HTTPException(404, "Produto não encontrado")
    return produto

💻 Exemplo 2 — Consulta Filtrada e Atualização Parcial (PATCH)

from typing import Optional
from sqlmodel import SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession

class ProdutoUpdate(SQLModel):
    nome: Optional[str] = None
    preco: Optional[float] = None

async def listar_por_faixa_preco(session: AsyncSession, minimo: float, maximo: float):
    stmt = select(Produto).where(Produto.preco.between(minimo, maximo)).order_by(Produto.preco)
    resultado = await session.exec(stmt)
    return resultado.all()

async def atualizar_parcial(session: AsyncSession, produto_id: int, dados: ProdutoUpdate) -> Produto | None:
    produto = await session.get(Produto, produto_id)
    if produto is None:
        return None
    for campo, valor in dados.model_dump(exclude_unset=True).items():
        setattr(produto, campo, valor)
    session.add(produto)
    await session.commit()
    await session.refresh(produto)
    return produto

💻 Exemplo 3 — Soft Delete Assíncrono com Auditoria

from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field
from sqlmodel.ext.asyncio.session import AsyncSession

class Pedido(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    descricao: str
    is_deleted: bool = Field(default=False)
    deleted_at: Optional[datetime] = None

async def soft_delete_pedido(session: AsyncSession, pedido_id: int) -> bool:
    pedido = await session.get(Pedido, pedido_id)
    if pedido is None or pedido.is_deleted:
        return False
    pedido.is_deleted = True
    pedido.deleted_at = datetime.utcnow()
    session.add(pedido)
    await session.commit()
    return True