🧩 Exercícios: Sessões Assíncronas e Operações CRUD
Spec Backend Com Python E Fastapi • Trilha Progressiva em 4 Níveis
🎯 Nível 1: Fundamentos
Problema 06.1 — Criação de Registro com AsyncSession e Commit
Contexto: Criação de Registro com AsyncSession e Commit no contexto de Sessões Assíncronas e Operações CRUD.
Requisitos de Execução:
- Implementar rotina assíncrona para instanciar e persistir uma entidade no banco via
session.add()eawait session.commit().
Resultado Esperado
Registro persistido no banco assíncrono com ID gerado.
📤 Instruções de Entrega (Microsoft Teams)
- Salve o arquivo como:
Atividade_06_1_SeuNome - Envie na tarefa:
Atividade Cap 06 - Sessões Assíncronas e Operações CRUD
🔑 Gabarito de Código & Solução Comentada
# async_crud_create.py
import asyncio
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 = Field(default=None, primary_key=True)
nome: str
preco: float
async def criar_produto_demo():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
async with AsyncSession(engine) as session:
p = Produto(nome="Mouse Gamer", preco=150.0)
session.add(p)
await session.commit()
await session.refresh(p)
print(f"Produto Criado: ID={p.id}, Nome={p.nome}")
asyncio.run(criar_produto_demo())
🔍 Nível 2: Prática
Problema 06.2 — Consulta de Registros com Filtros e Ordenação via select()
Contexto: Consulta de Registros com Filtros e Ordenação via select() no contexto de Sessões Assíncronas e Operações CRUD.
Requisitos de Execução:
- Executar consulta assíncrona usando
select(Produto).where(Produto.preco > 100).order_by(Produto.preco).
Resultado Esperado
2 produtos filtrados e ordenados por preço.
📤 Instruções de Entrega (Microsoft Teams)
- Salve o arquivo como:
Atividade_06_2_SeuNome - Envie na tarefa:
Atividade Cap 06 - Sessões Assíncronas e Operações CRUD
🔑 Gabarito de Código & Solução Comentada
# async_query.py
import asyncio
from sqlmodel import SQLModel, Field, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
class Item(SQLModel, table=True):
id: int = Field(default=None, primary_key=True)
titulo: str
valor: float
async def buscar_itens():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all)
async with AsyncSession(engine) as session:
session.add_all([Item(titulo="A", valor=50.0), Item(titulo="B", valor=150.0)])
await session.commit()
stmt = select(Item).where(Item.valor > 100.0)
res = await session.exec(stmt)
itens = res.all()
print("Itens encontrados:", [i.titulo for i in itens])
asyncio.run(buscar_itens())
⚡ Nível 3: Integração
Problema 06.3 — Atualização Parcial de Registro (PATCH) com Verificação de Existência
Contexto: Atualização Parcial de Registro (PATCH) com Verificação de Existência no contexto de Sessões Assíncronas e Operações CRUD.
Requisitos de Execução:
- Implementar endpoint PATCH que busque o registro por ID, lance 404 se inexistente e atualize apenas os campos enviados.
Resultado Esperado
PATCH /produtos/1 -> 200 OK com campo atualizado.
📤 Instruções de Entrega (Microsoft Teams)
- Salve o arquivo como:
Atividade_06_3_SeuNome - Envie na tarefa:
Atividade Cap 06 - Sessões Assíncronas e Operações CRUD
🔑 Gabarito de Código & Solução Comentada
# partial_update.py
from sqlmodel import SQLModel, Field
from typing import Optional
class ProdutoUpdate(SQLModel):
nome: Optional[str] = None
preco: Optional[float] = None
def aplicar_patch(produto_db, dados_update: ProdutoUpdate):
update_dict = dados_update.model_dump(exclude_unset=True)
for chave, valor in update_dict.items():
setattr(produto_db, chave, valor)
return produto_db
class ProdutoMock:
def __init__(self, id, nome, preco): self.id, self.nome, self.preco = id, nome, preco
p = ProdutoMock(1, "Notebook Antigo", 2000.0)
patch_data = ProdutoUpdate(preco=1800.0)
p_atualizado = aplicar_patch(p, patch_data)
print(f"Preço atualizado: R$ {p_atualizado.preco}")
🏆 Nível 4: Desafio Corporativo
Problema 06.4 — Soft Delete e Auditoria de Exclusão Lógica
Contexto: Soft Delete e Auditoria de Exclusão Lógica no contexto de Sessões Assíncronas e Operações CRUD.
Requisitos de Execução:
- Implementar exclusão lógica (
is_deleted=True, deleted_at=datetime.utcnow()) preservando integridade referencial.
Resultado Esperado
Registro marcado como deletado sem remoção física do banco.
📤 Instruções de Entrega (Microsoft Teams)
- Salve o arquivo como:
Atividade_06_4_SeuNome - Envie na tarefa:
Atividade Cap 06 - Sessões Assíncronas e Operações CRUD
🔑 Gabarito de Código & Solução Comentada
# soft_delete.py
from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field
class RegistroAuditavel(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
def soft_delete(self):
self.is_deleted = True
self.deleted_at = datetime.utcnow()
reg = RegistroAuditavel(descricao="Contrato 2026")
reg.soft_delete()
print(f"Registro Excluído Logicamente: {reg.is_deleted}, Em: {reg.deleted_at}")