API ágeis, escaláveis & Python

Cada vez mais a demanda por APIs rápidas, escaláveis e fáceis de manter tem crescido exponencialmente. No mesmo sentido, o desenvolvimento de API assíncronas tem se tornado essencial para aplicações que exigem alta concorrência e baixa latência.

Este artigo apresenta a construção de uma API “simples” voltada para a gestão de treinamentos de CrossFit, desenvolvida como um laboratório prático para demonstrar o uso completo de Python, FastAPI e o ORM SQLAlchemy em um fluxo assíncrono de ponta a ponta, incluíndo os testes de integração com PyTest.

Para assegurar a qualidade do código e a confiabilidade dos endpoints, incorporamos testes integrados utilizando PyTest com suporte a operações assíncronas. Esses testes cobrem desde a validação dos modelos Pydantic até a verificação do comportamento da camada de persistência, proporcionando um feedback rápido durante a implementação.

O objetivo final é disponibilizar um protótipo funcional que ilustra boas práticas de arquitetura, documentação da API, organização de código e otimização de desempenho em ambientes assíncronos. Além disso, oferecer um exemplo funcional e bem estruturado que sirva como ponto de partida para projetos reais que exijam alta performance, manutenção simplificada e cobertura de testes abrangente - seja para treinamentos esportivas ou qualquer outra aplicação que necessite de uma API moderna e responsiva.

Requisitos funcionais

Para o laboratório, as funcionalidades consideradas são:

Atleta de crossfit está inserido em apenas uma categoria (nível CrosFit).
Cada categoria pode conter vários atletas.
Um atleta pode treinar em diferentes centros de treinamento.
Os centros de treinamento pode abrigar vários atletas treinando.

As tecnologias

Por que FastAPI?

FastAPI foi escolhido como framework principal por duas razões fundamentais:

Modelo Declarativo + Validação Automática – Utiliza type hints e Pydantic para validar payloads com baixo esforço adicional.
Suporte Nativo a Async/Await - Cada rota pode ser declarada como async, permitindo que o servidor (Uvicorn ou Hypercorn) manipule milhares de conexões simultâneas sem bloquear o loop de eventos.

Essas características reduzem drasticamente a quantidade de padrões e garantem alta performance out‑of‑the‑box.

Persistência Assíncrona com SQLAlchemy

O SQLAlchemy oferece um driver assíncrono (asyncpg para PostgreSQL).
Ele combina:

ORM completo – Mapeia classes Python para tabelas relacionais.
Sessão assíncrona – Operações de CRUD podem ser executadas dentro de coroutines Python, preservando a coerência transacional.
Integração com Alembic – Migrações de esquema são geridas de forma a aproveitar da implementação assícrona do driver do banco.

Exemplo parcial do Modelo de Domínio:

from datetime import datetime, timezone

from sqlalchemy import DateTime, ForeignKey, String, func
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class ORMBase(AsyncAttrs, DeclarativeBase):
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=datetime.now(timezone.utc),
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=datetime.now(timezone.utc),
        server_default=func.now(),
        server_onupdate=func.now(),
        onupdate=func.now(),
    )


class Category(ORMBase):
    __tablename__ = "category"

    name: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)

    def __repr__(self):
        return "<Category(id={}, name={})>".format(self.id, self.name)

    athletes: Mapped[list["AthleteCategory"]] = relationship(
        back_populates="categories", cascade="all, delete-orphan"
    )

Camada de API com FastAPI

Estrutura de diretórios principais:

├── docs                # documentações (solution design, ER, UML, OpenAPI)
├── alembic             # scripts de migrações de banco
└── src
    ├── api
    │   ├── controllers # tramento de funcionalidades
    │   ├── main.py     # ponto de entrada
    │   ├── models 
    │   │   ├── dto     # Pydantic models (DTO)
    │   │   ├── orm     # SQLAlchemy models
    │   ├── services    # Handlers de serviços (ex: Context Manager DB Async, Bus)
    │   ├── setup       # Setup base (ex: api, conexão DB)
    │   └── views       # Rotas
    └── tests
        ├── category    # Casos de testes para Categoria
        ├── conftest.py # Fixtures de testes
        └── health      # Casos de testes para saúde da aplicação

Migrações com Alembic (Modo Assíncrono)

Embora Alembic seja síncrono por padrão, foi adaptado para execução com o driver asyncpg.

É também mantido como parte do pipeline CI/CD em que as migrações são aplicados antes de iniciar os testes ou mesmo a aplicação em ambiente produtigo (staging/production).
Exemplo parcial de env.py adaptado:

#...
def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """In this scenario we need to create an Engine
    and associate a connection with the context.

    """

    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()
#...

Testes Automatizados com PyTest (Assíncrono)

As fixtures de banco poderiam ser em memória, porém, com teste de integração, as rotinas são realizadas diretamente na aplicação. Desta forma, já são diretamente validados os módulos de banco existentes na aplicação.

Exemplo parcial de casos te testes de Categorias:

class TestCategory:

    async def test_category_post_success(
        self, aioclient: AsyncClient, category_in, category_out
    ):
        # Given - setting the stage
        # category_in

        # When - the action
        response = await aioclient.post(url=self._CATEGORY_URL_PREFIX, json=category_in)

        # Then - the outcome
        assert response.status_code == status.HTTP_201_CREATED
        content = response.json()
        content["name"] == category_in["name"]
        category_out["id"] = content["id"]
        category_out["name"] = content["name"]
    # ...

Os testes utilizam as bibliotecas httpx.AsyncClient para executar a API em memória, garantindo que todo o fluxo – validação Pydantic, camada de CRUD e sessão assíncrona – seja exercitado. Alem disso, a esteira CI/CD abrange análises de cobertura de testes e linter de código.

Conclusão

Mostrei aqui como combinar as ferramentas mais modernas do ecossistema Python para criar uma API totalmente assíncrona, testável e pronta para produção. Embora o caso de uso seja a gestão de treinamentos de CrossFit, a arquitetura apresentada – FastAPI + SQLAlchemy + Alembic + PyTest – pode ser reaplicada a qualquer domínio que exija:

Respostas em tempo real;
Escalabilidade horizontal;
Manutenção simplificada graças a tipagem forte e validação automática;
Garantia de qualidade por meio de testes integrados.

A partir daqui, você pode expandir a solução adicionando autenticação JWT, websockets para push de resultados ao vivo, integrar um cache distribuído (ex: Valkey ou Redis) para reduzir a latência das consultas mais frequentes, ou ainda implementar recursos event-driven (ex: Kafka) para disponibilizar dados para outras análises.

O código-fonte para a API descrita acima está em https://gitlab.com/adrianovieira/workout-api

Bora divertir mais… ;)

Também publicado em https://dev.to/adrianovieira/agile-scalable-apis-python-2ma

Adriano Vieira