Postagens das partes que compõem o projeto:
A) Atleta;
rom fastapi import FastAPI

app = FastAPI(title='workaoutapi')

if name"main":
import uvicorn
uvicorn.run(app, host= '0, 0, 0, port=8000')
from typing import Annotated, Optional
from pydantic import Field, PositiveFloat
from workout_api.categorias.schemas import CategoriaIn
from workout_api.centro_treinamento.schemas import CentroTreinamentoAtleta

from workout_api.contrib.schemas import BaseSchema, OutMixin

class Atleta(BaseSchema):
nome: Annotated[str, Field(description='Nome do atleta', example='Joao', max_length=50)]
cpf: Annotated[str, Field(description='CPF do atleta', example='12345678900', max_length=11)]
idade: Annotated[int, Field(description='Idade do atleta', example=25)]
peso: Annotated[PositiveFloat, Field(description='Peso do atleta', example=75.5)]
altura: Annotated[PositiveFloat, Field(description='Altura do atleta', example=1.70)]
sexo: Annotated[str, Field(description='Sexo do atleta', example='M', max_length=1)]
categoria: Annotated[CategoriaIn, Field(description='Categoria do atleta')]
centro_treinamento: Annotated[CentroTreinamentoAtleta, Field(description='Centro de treinamento do atleta')]
class AtletaIn(Atleta):
pass

class AtletaOut(Atleta, OutMixin):
pass

class AtletaUpdate(BaseSchema):
nome: Annotated[Optional[str], Field(None, description='Nome do atleta', example='Joao', max_length=50)]
idade: Annotated[Optional[int], Field(None, description='Idade do atleta', example=25)]

class AtletaModel(BaseModel):
tablename = 'atletas'

pk_id: Mapped[int] = mapped_column(Integer, primary_key=True)
nome: Mapped[str] = mapped_column(String(50), nullable=False)
cpf: Mapped[str] = mapped_column(String(11), unique=True, nullable=False)
idade: Mapped[int] = mapped_column(Integer, nullable=False)
peso: Mapped[float] = mapped_column(Float, nullable=False)
altura: Mapped[float] = mapped_column(Float, nullable=False)
sexo: Mapped[str] = mapped_column(String(1), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
categoria: Mapped['CategoriaModel'] = relationship(back_populates="atleta", lazy='selectin')
categoria_id: Mapped[int] = mapped_column(ForeignKey("categorias.pk_id"))
centro_treinamento: Mapped['CentroTreinamentoModel'] = relationship(back_populates="atleta", lazy='selectin')
centro_treinamento_id: Mapped[int] = mapped_column(ForeignKey("centros_treinamento.pk_id"))

class Atleta(BaseSchema):
nome: Annotated[str, Field(description='Nome do atleta', example='Joao', max_length=50)]
cpf: Annotated[str, Field(description='CPF do atleta', example='12345678900', max_length=11)]
idade: Annotated[int, Field(description='Idade do atleta', example=25)]
peso: Annotated[PositiveFloat, Field(description='Peso do atleta', example=75.5)]
altura: Annotated[PositiveFloat, Field(description='Altura do atleta', example=1.70)]
sexo: Annotated[str, Field(description='Sexo do atleta', example='M', max_length=1)]
categoria: Annotated[CategoriaIn, Field(description='Categoria do atleta')]
centro_treinamento: Annotated[CentroTreinamentoAtleta, Field(description='Centro de treinamento do atleta')]

class AtletaIn(Atleta):
pass

class AtletaOut(Atleta, OutMixin):
pass

class AtletaUpdate(BaseSchema):
nome: Annotated[Optional[str], Field(None, description='Nome do atleta', example='Joao', max_length=50)]
idade: Annotated[Optional[int], Field(None, description='Idade do atleta', example=25)]

version: "3"
services:
db:
image: postgres:11-alpine
environment:
POSTGRES_PASSWORD: workout
POSTGRES_USER: workout
POSTGRES_DB: workout
ports:
- "5432:5432"
@router.post(
'/',
summary='Criar um novo atleta',
status_code=status.HTTP_201_CREATED,
response_model=AtletaOut
)
async def post(
db_session: DatabaseDependency,
atleta_in: AtletaIn = Body(...)
):
categoria_nome = atleta_in.categoria.nome
centro_treinamento_nome = atleta_in.centro_treinamento.nome
categoria = (await db_session.execute(
select(CategoriaModel).filter_by(nome=categoria_nome))
).scalars().first()

if not categoria:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f'A categoria {categoria_nome} não foi encontrada.'
)

centro_treinamento = (await db_session.execute(
select(CentroTreinamentoModel).filter_by(nome=centro_treinamento_nome))
).scalars().first()

if not centro_treinamento:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f'O centro de treinamento {centro_treinamento_nome} não foi encontrado.'
)
try:
atleta_out = AtletaOut(id=uuid4(), created_at=datetime.utcnow(), **atleta_in.model_dump())
atleta_model = AtletaModel(**atleta_out.model_dump(exclude={'categoria', 'centro_treinamento'}))

  atleta_model.categoria_id = categoria.pk_id
  atleta_model.centro_treinamento_id = centro_treinamento.pk_id

  db_session.add(atleta_model)
  await db_session.commit()

except Exception:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Ocorreu um erro ao inserir os dados no banco'
)

return atleta_out

@router.get(
'/',
summary='Consultar todos os Atletas',
status_code=status.HTTP_200_OK,
response_model=list[AtletaOut],
)
async def query(db_session: DatabaseDependency) -> list[AtletaOut]:
atletas: list[AtletaOut] = (await db_session.execute(select(AtletaModel))).scalars().all()

return [AtletaOut.model_validate(atleta) for atleta in atletas]

@router.get(
'/{id}',
summary='Consulta um Atleta pelo id',
status_code=status.HTTP_200_OK,
response_model=AtletaOut,
)
async def get(id: UUID4, db_session: DatabaseDependency) -> AtletaOut:
atleta: AtletaOut = (
await db_session.execute(select(AtletaModel).filter_by(id=id))
).scalars().first()

if not atleta:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f'Atleta não encontrado no id: {id}'
)

return atleta

@router.patch(
'/{id}',
summary='Editar um Atleta pelo id',
status_code=status.HTTP_200_OK,
response_model=AtletaOut,
)
async def patch(id: UUID4, db_session: DatabaseDependency, atleta_up: AtletaUpdate = Body(...)) -> AtletaOut:
atleta: AtletaOut = (
await db_session.execute(select(AtletaModel).filter_by(id=id))
).scalars().first()

if not atleta:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f'Atleta não encontrado no id: {id}'
)

atleta_update = atleta_up.model_dump(exclude_unset=True)
for key, value in atleta_update.items():
setattr(atleta, key, value)

await db_session.commit()
await db_session.refresh(atleta)

return atleta

@router.delete(
'/{id}',
summary='Deletar um Atleta pelo id',
status_code=status.HTTP_204_NO_CONTENT
)
async def delete(id: UUID4, db_session: DatabaseDependency) -> None:
atleta: AtletaOut = (
await db_session.execute(select(AtletaModel).filter_by(id=id))
).scalars().first()

if not atleta:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f'Atleta não encontrado no id: {id}'
)

await db_session.delete(atleta)
await db_session.commit()

B) Categorias;

class CategoriaModel(BaseModel):
tablename = 'categorias'

pk_id: Mapped[int] = mapped_column(Integer, primary_key=True)
nome: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
atleta: Mapped['AtletaModel'] = relationship(back_populates="categoria")

from typing import Annotated

from pydantic import UUID4, Field
from workout_api.contrib.schemas import BaseSchema

class CategoriaIn(BaseSchema):
nome: Annotated[str, Field(description='Nome da categoria', example='Scale', max_length=10)]

class CategoriaOut(CategoriaIn):
id: Annotated[UUID4, Field(description='Identificador da categoria')]

@router.post(
'/',
summary='Criar uma nova Categoria',
status_code=status.HTTP_201_CREATED,
response_model=CategoriaOut,
)
async def post(
db_session: DatabaseDependency,
categoria_in: CategoriaIn = Body(...)
) -> CategoriaOut:
categoria_out = CategoriaOut(id=uuid4(), **categoria_in.model_dump())
categoria_model = CategoriaModel(**categoria_out.model_dump())

db_session.add(categoria_model)
await db_session.commit()

return categoria_out

@router.get(
'/',
summary='Consultar todas as Categorias',
status_code=status.HTTP_200_OK,
response_model=list[CategoriaOut],
)
async def query(db_session: DatabaseDependency) -> list[CategoriaOut]:
categorias: list[CategoriaOut] = (await db_session.execute(select(CategoriaModel))).scalars().all()

return categorias

@router.get(
'/{id}',
summary='Consulta uma Categoria pelo id',
status_code=status.HTTP_200_OK,
response_model=CategoriaOut,
)
async def get(id: UUID4, db_session: DatabaseDependency) -> CategoriaOut:
categoria: CategoriaOut = (
await db_session.execute(select(CategoriaModel).filter_by(id=id))
).scalars().first()

if not categoria:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f'Categoria não encontrada no id: {id}'
)

return categoria

C) Centro de Treinamento;

from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from workout_api.contrib.models import BaseModel
from workout_api.atleta.models import AtletaModel

class CentroTreinamentoModel(BaseModel):
tablename = 'centros_treinamento'

pk_id: Mapped[int] = mapped_column(Integer, primary_key=True)
nome: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
endereco: Mapped[str] = mapped_column(String(60), nullable=False)
proprietario: Mapped[str] = mapped_column(String(30), nullable=False)
atleta: Mapped['AtletaModel'] = relationship(back_populates='centro_treinamento')

ss CentroTreinamentoModel(BaseModel):
tablename = 'centros_treinamento'

pk_id: Mapped[int] = mapped_column(Integer, primary_key=True)
nome: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
endereco: Mapped[str] = mapped_column(String(60), nullable=False)
proprietario: Mapped[str] = mapped_column(String(30), nullable=False)
atleta: Mapped['AtletaModel'] = relationship(back_populates='centro_treinamento')

m typing import Annotated

from pydantic import Field, UUID4
from workout_api.contrib.schemas import BaseSchema

class CentroTreinamentoIn(BaseSchema):
nome: Annotated[str, Field(description='Nome do centro de treinamento', example='CT King', max_length=20)]
endereco: Annotated[str, Field(description='Endereço do centro de treinamento', example='Rua X, Q02', max_length=60)]
proprietario: Annotated[str, Field(description='Proprietario do centro de treinamento', example='Marcos', max_length=30)]

class CentroTreinamentoAtleta(BaseSchema):
nome: Annotated[str, Field(description='Nome do centro de treinamento', example='CT King', max_length=20)]

class CentroTreinamentoOut(CentroTreinamentoIn):
id: Annotated[UUID4, Field(description='Identificador do centro de treinamento')]

from uuid import uuid4
from fastapi import APIRouter, Body, HTTPException, status
from pydantic import UUID4
from workout_api.centro_treinamento.schemas import CentroTreinamentoIn, CentroTreinamentoOut
from workout_api.centro_treinamento.models import CentroTreinamentoModel

from workout_api.contrib.dependencies import DatabaseDependency
from sqlalchemy.future import select

router = APIRouter()

@router.post(
'/',
summary='Criar um novo Centro de treinamento',
status_code=status.HTTP_201_CREATED,
response_model=CentroTreinamentoOut,
)
async def post(
db_session: DatabaseDependency,
centro_treinamento_in: CentroTreinamentoIn = Body(...)
) -> CentroTreinamentoOut:
centro_treinamento_out = CentroTreinamentoOut(id=uuid4(), **centro_treinamento_in.model_dump())
centro_treinamento_model = CentroTreinamentoModel(**centro_treinamento_out.model_dump())

db_session.add(centro_treinamento_model)
await db_session.commit()

return centro_treinamento_out

@router.get(
'/',
summary='Consultar todos os centros de treinamento',
status_code=status.HTTP_200_OK,
response_model=list[CentroTreinamentoOut],
)
async def query(db_session: DatabaseDependency) -> list[CentroTreinamentoOut]:
centros_treinamento_out: list[CentroTreinamentoOut] = (
await db_session.execute(select(CentroTreinamentoModel))
).scalars().all()

return centros_treinamento_out

@router.get(
'/{id}',
summary='Consulta um centro de treinamento pelo id',
status_code=status.HTTP_200_OK,
response_model=CentroTreinamentoOut,
)
async def get(id: UUID4, db_session: DatabaseDependency) -> CentroTreinamentoOut:
centro_treinamento_out: CentroTreinamentoOut = (
await db_session.execute(select(CentroTreinamentoModel).filter_by(id=id))
).scalars().first()

if not centro_treinamento_out:
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND, 
        detail=f'Centro de treinamento não encontrado no id: {id}'
    )

return centro_treinamento_out

D) Configurações

from typing import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from workout_api.configs.settings import settings

engine = create_async_engine(settings.DB_URL, echo=False)
async_session = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)

async def get_session() -> AsyncGenerator:
async with async_session() as session:
yield session
from pydantic import Field
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
DB_URL: str = Field(default='postgresql+asyncpg://workout:workout@localhost/workout')

settings = Settings()

DESAFIO

#Instalação das Dependências

pip install fastapi uvicorn pymongo motor pytest

#Configuração do MongoDB

from pydantic import BaseModel
from typing import Optional

class Atleta(BaseModel):
    nome: str
    cpf: str
    centro_treinamento: Optional[str] = None
    categoria: Optional[str] = None

#Configuração da Aplicação FastAPI

from fastapi import FastAPI
from app.routes import router

app = FastAPI()

app.include_router(router)

#Configuração do Pytest

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_atleta():
    response = client.post("/atletas", json={
        "nome": "João Silva",
        "cpf": "12345678901",
        "centro_treinamento": "Centro A",
        "categoria": "Categoria 1"
    })
    assert response.status_code == 200
    assert response.json() == {
        "nome": "João Silva",
        "cpf": "12345678901",
        "centro_treinamento": "Centro A",
        "categoria": "Categoria 1"
    }

def test_get_atletas():
    response = client.get("/atletas")
    assert response.status_code == 200
    assert len(response.json()) > 0

**Projeto FastAPI com TDD

Neste projeto, você aprenderá na prática como implementar o TDD em uma aplicação utilizando FastAPI juntamente com Pytest. Criaremos juntos uma API utilizando o banco de dados MongoDB e realizando testes unitários e de integração. Além disso, você pode conhecer boas práticas de como documentar um projeto.**

pip install fastapi uvicorn pymongo motor pytest

#Configuração do MongoDB

from pydantic import BaseModel
from typing import Optional

class Atleta(BaseModel):
nome: str
cpf: str
centro_treinamento: Optional[str] = None
categoria: Optional[str] = None
#Configuração da Aplicação FastAPI

from fastapi import FastAPI
from app.routes import router

app = FastAPI()

app.include_router(router)
#Configuração do Pytest

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_atleta():
response = client.post("/atletas", json={
"nome": "João Silva",
"cpf": "12345678901",
"centro_treinamento": "Centro A",
"categoria": "Categoria 1"
})
assert response.status_code == 200
assert response.json() == {
"nome": "João Silva",
"cpf": "12345678901",
"centro_treinamento": "Centro A",
"categoria": "Categoria 1"
}

def test_get_atletas():
response = client.get("/atletas")
assert response.status_code == 200
assert len(response.json()) > 0

DESAFIO

Instalação das Dependências

pip install fastapi sqlalchemy fastapi-pagination uvicorn

#Criação do Modelo de Dados

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

class Atleta(Base):
    __tablename__ = "atletas"

    id = Column(Integer, primary_key=True, autoincrement=True)
    nome = Column(String(255), nullable=False)
    cpf = Column(String(11), unique=True, nullable=False)
    centro_treinamento = Column(String(255))
    categoria = Column(String(255))

    # Relacionamento com tabela de treinos (opcional)
    # treinos = relationship("Treino", backref="atleta")

#Configuração do Banco de Dados:

from` sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database import Base

DATABASE_URL = "postgresql://user:password@host:port/database"

engine = create_engine(DATABASE_URL)
Base.metadata.create_all(engine)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

#Implementação da API Assíncrona

``
from fastapi import FastAPI, Depends, HTTPException
from fastapi_pagination import Page, add_pagination
from sqlalchemy.orm import Session
from models import Atleta
from database import SessionLocal

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    global session
    session = SessionLocal()

@app.on_event("shutdown")
async def shutdown_event():
    global session
    session.close()

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

Endpoint para buscar todos os atletas

@app.get("/atletas", response_model=Page[Atleta])
async def get_all_atletas(db: Session = Depends(get_db), limit: int | None = None, offset: int | None = None):
    atletas = db.query(Atleta).order_by(Atleta.id).limit(limit).offset(offset).all()
    return Page(atletas, len(atletas))

Endpoint para buscar atletas por nome e CPF

@app.get("/atletas/search", response_model=List[Atleta])
async def get_atleta_by_nome_cpf(nome: str | None = None, cpf: str | None = None, db: Session = Depends(get_db)):
    query = db.query(Atleta)

    if nome:
        query = query.filter(Atleta.nome.ilike(f"%{nome}%"))

    if cpf:
        query = query.filter(Atleta.cpf == cpf)

    atletas = query.all()

    if not atletas:
        raise HTTPException(status_code=404, detail="Atleta não encontrado")

    return atletas

Endpoint para cadastrar um novo atleta

@app.post("/atletas")
async def create_atleta(atleta: Atleta, db: Session = Depends(get_db)):
    db.add(atleta)
    db.commit()
    db.refresh(atleta)
    return atleta

add_pagination(app)

#Obter sessão do banco de dados

def get_db(session: Session = Depends(SessionLocal)):
return session

#Endpoint para buscar todos os atletas

@app.get("/atletas", response_model=Page[Atleta])
async def get_all_atletas(db: Session = Depends(get_db), limit: int | None = None, offset: int | None = None):
atletas = db.query(Atleta).order_by(Atleta.id).limit(limit).offset(offset).all()
return Page(atletas, len(atletas))

#Endpoint para buscar atletas por nome e CPF

@app.get("/atletas", response_model=List[Atleta])
async def get_atleta_by_nome_cpf(nome: str | None = None, cpf: str | None = None, db: Session = Depends(get_db)):
query = db.query(Atleta)

if nome:
    query = query.filter(Atleta.nome.ilike(f"%{nome}%"))

if cpf:
    query = query.filter(Atleta.cpf == cpf)

atletas = query.all()

if not atletas:
    raise HTTPException(status_code=404, detail="Atleta não encontrado")

return atletas

#Endpoint para cadastrar um novo atleta

@app.post("/atletas")
async def

Instalação das Dependências
bash
pip install fastapi sqlalchemy fastapi-pagination uvicorn

Criação do Modelo de Dados
Python
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

from database import Base

class Atleta(Base):
tablename = "atletas"

id = Column(Integer, primary_key=True, autoincrement=True)
nome = Column(String(255), nullable=False)
cpf = Column(String(11), unique=True, nullable=False)
centro_treinamento = Column(String(255))
categoria = Column(String(255))

# Relacionamento com tabela de treinos (opcional)
# treinos = relationship("Treino", backref="atleta")

Configuração do Banco de Dados:
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("postgresql://user:password@host:port/database")
Base.metadata.create_all(engine)

SessionLocal = Session(autocommit=False, autoflush=False, bind=engine)

Implementação da API Assíncrona
Python
from fastapi import FastAPI, Depends, HTTPException
from fastapi_pagination import Page, add_pagination
from sqlalchemy.orm import Session

from models import Atleta
from database import SessionLocal

app = FastAPI()

@app.on_event("startup")
async def startup_event():
global session
session = SessionLocal()

@app.on_event("shutdown")
async def shutdown_event():
global session
session.close()

Obter sessão do banco de dados

def get_db(session: Session = Depends(SessionLocal)):
return session

Endpoint para buscar todos os atletas

@app.get("/atletas", response_model=Page[Atleta])
async def get_all_atletas(db: Session = Depends(get_db), limit: int | None = None, offset: int | None = None):
atletas = db.query(Atleta).order_by(Atleta.id).limit(limit).offset(offset).all()
return Page(atletas, len(atletas))

Endpoint para buscar atletas por nome e CPF

@app.get("/atletas", response_model=List[Atleta])
async def get_atleta_by_nome_cpf(nome: str | None = None, cpf: str | None = None, db: Session = Depends(get_db)):
query = db.query(Atleta)

if nome:
    query = query.filter(Atleta.nome.ilike(f"%{nome}%"))

if cpf:
    query = query.filter(Atleta.cpf == cpf)

atletas = query.all()

if not atletas:
    raise HTTPException(status_code=404, detail="Atleta não encontrado")

return atletas

Endpoint para cadastrar um novo atleta

@app.post("/atletas")
async def