Como configurar e testar aplicações FastAPI com Pytest

No desenvolvimento de aplicações web com FastAPI, a criação de testes automatizados é uma parte fundamental para garantir a integridade e o funcionamento correto da aplicação. Um dos frameworks mais utilizados para testes em Python é o pytest, que se integra perfeitamente ao FastAPI. A seguir, apresentamos como configurar o ambiente de testes e escrever testes tanto unitários quanto de integração para endpoints de uma aplicação FastAPI.

O primeiro passo para configurar o ambiente de testes é organizar a estrutura do projeto. A estrutura básica deve incluir um diretório principal para o código da aplicação e outro para os testes. A seguir, exemplificamos a estrutura do projeto:

css
Copy code
protoapp/

│─ protoapp/
│  └─ main.py
│─ tests/
│  └─ test_main.py
│─ pytest.ini

O arquivo pytest.ini é fundamental para configurar o pytest. Nele, podemos definir o caminho de onde o código da aplicação se encontra, adicionando ao PYTHONPATH. O conteúdo mínimo do arquivo deve ser o seguinte:

csharp
Copy code
[pytest]
pythonpath = .
protoapp

Em seguida, criamos um módulo de testes, por exemplo, test_main.py, onde escreveremos os testes para os endpoints da aplicação. Vamos começar com o teste de um endpoint básico /home da aplicação:

python
Copy code
import pytest

from httpx import ASGITransport, AsyncClient
from protoapp.main import app
@pytest.mark.asyncio
async def test_read_main():
    client = AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    )
    response = await client.get("/home")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

Este é um teste simples, onde verificamos se o endpoint /home retorna corretamente o status 200 e o conteúdo esperado. Para rodar os testes, basta executar o comando pytest no diretório principal do projeto.

Uma vez que o ambiente esteja configurado, podemos começar a escrever testes unitários para os endpoints. O TestClient do FastAPI é uma ferramenta excelente para realizar testes de integração de forma simples e eficiente. A criação de um teste unitário para o mesmo endpoint /home pode ser feita da seguinte forma:

Primeiro, criamos uma fixture no módulo conftest.py para configurar o cliente de teste, que será utilizado em múltiplos testes:

python
Copy code
import pytest

from fastapi.testclient import TestClient
from protoapp.main import app
@pytest.fixture(scope="function")
def test_client():
    client = TestClient(app)
    yield client

Agora podemos escrever um teste unitário no arquivo test_main.py, que utiliza a fixture do cliente de teste:

python
Copy code
def test_read_main_client(test_client):
    response = test_client.get("/home")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

Além dos testes unitários, é fundamental realizar testes de integração para verificar se diferentes partes da aplicação estão funcionando de forma conjunta, como a interação com bancos de dados ou outros serviços externos. Para ilustrar como configurar e testar endpoints que interagem com um banco de dados SQL, vamos adicionar uma configuração básica de banco de dados utilizando SQLAlchemy.

Primeiro, criamos o arquivo database.py dentro da pasta protoapp/, que irá configurar o banco de dados e a sessão de conexão:

python
Copy code
from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker, DeclarativeBase
DATABASE_URL = "sqlite:///./production.db"
engine = create_engine(DATABASE_URL)
Base = DeclarativeBase()
SessionLocal = sessionmaker(
    autocommit=False, autoflush=False, bind=engine
)

A seguir, criamos a classe Item que irá mapear a tabela do banco de dados:

python
Copy code
from sqlalchemy.orm import Mapped, mapped_column
class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)

    name: Mapped[str] = mapped_column(index=True)
    color: Mapped[str]

Com isso, criamos a sessão de banco de dados no arquivo main.py, para ser utilizada pelos endpoints:

python
Copy code
from protoapp.database import SessionLocal
def get_db_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Agora, podemos criar endpoints para adicionar e ler itens do banco de dados. Por exemplo, um endpoint POST para adicionar um item e um GET para buscar um item pelo ID:

python
Copy code
from fastapi import FastAPI, Depends

from sqlalchemy.orm import Session
from protoapp.database import SessionLocal
from protoapp.models import Item
app = FastAPI()
@app.post("/items/")
def create_item(item: Item, db: Session = Depends(get_db_session)):
    db.add(item)
    db.commit()
    return item
@app.get("/items/{item_id}")

def read_item(item_id: int, db: Session = Depends(get_db_session)):

    return db.query(Item).filter(Item.id == item_id).first()

Agora, podemos escrever testes para garantir que esses endpoints estão funcionando corretamente. Um teste de integração para o endpoint POST /items/ poderia ser feito da seguinte forma:

python
Copy code
def test_create_item(test_client):

    response = test_client.post("/items/", json={"name": "Test Item", "color": "Red"})

    assert response.status_code == 200
    assert response.json()["name"] == "Test Item"
    assert response.json()["color"] == "Red"

Esse teste verifica se o item foi corretamente inserido no banco de dados.

Importante, ao escrever testes de integração, lembre-se sempre de isolar o banco de dados utilizado nos testes para não afetar a base de dados real da aplicação. Isso pode ser feito configurando um banco de dados específico para os testes ou utilizando ferramentas como fixtures para limpar o banco de dados antes e depois de cada teste.

Como Gerenciar Relacionamentos em Bancos de Dados SQL para Sistemas de Ingressos

Relacionamentos de um-para-um são utilizados para agrupar informações específicas sobre um registro de forma lógica separada. No contexto de um sistema de ingressos, isso pode ser útil quando queremos associar detalhes específicos a um ingresso, como o assento e o tipo de ingresso. Esses detalhes, embora relacionados ao ingresso, precisam ser tratados separadamente para garantir maior flexibilidade e clareza na gestão de dados. A seguir, vamos mostrar como criar e manipular esse tipo de relacionamento em uma base de dados utilizando SQLAlchemy, o que nos permitirá estruturar um sistema robusto e bem organizado.

Configuração da Tabela de Ingressos e Detalhes

Primeiro, é necessário adicionar uma referência aos detalhes do ingresso na classe Ticket já existente. A tabela dos ingressos conterá os dados principais, como o preço e o nome do espetáculo, mas também terá um campo de relacionamento com a tabela de detalhes do ingresso:

python
Copy code
class Ticket(Base):

    __tablename__ = "tickets"
    id: Mapped[int] = mapped_column(primary_key=True)
    price: Mapped[float] = mapped_column(nullable=True)
    show: Mapped[str | None]
    user: Mapped[str | None]
    sold: Mapped[bool] = mapped_column(default=False)
    details: Mapped["TicketDetails"] = relationship(back_populates="ticket")

Em seguida, criamos a tabela TicketDetails para mapear os detalhes associados ao ingresso:

python
Copy code
from sqlalchemy import ForeignKey
class TicketDetails(Base):
    __tablename__ = "ticket_details"
    id: Mapped[int] = mapped_column(primary_key=True)
    ticket_id: Mapped[int] = mapped_column(ForeignKey("tickets.id"))
    ticket: Mapped["Ticket"] = relationship(back_populates="details")
    seat: Mapped[str | None]
    ticket_type: Mapped[str | None]

Essa estrutura estabelece uma relação de um-para-um entre as tabelas tickets e ticket_details, permitindo armazenar e manipular informações como o assento e o tipo de ingresso de forma organizada e separada.

Atualizando os Detalhes do Ingresso

Após configurar as tabelas, o próximo passo é criar operações CRUD (Create, Read, Update, Delete) para manipular esses dados. Para atualizar os detalhes de um ingresso, podemos criar uma função dedicada dentro do módulo de operações:

python
Copy code
async def update_ticket_details(

    db_session: AsyncSession,
    ticket_id: int,
    updating_ticket_details: dict,
) -> bool:
    ticket_query = update(TicketDetails).where(TicketDetails.ticket_id == ticket_id)
    if updating_ticket_details != {}:
        ticket_query = ticket_query.values(updating_ticket_details)
    result = await db_session.execute(ticket_query)
    await db_session.commit()
    if result.rowcount == 0:
        return False
    return True

Esse código permite que os detalhes de um ingresso sejam atualizados de forma eficiente, retornando False caso nenhum registro tenha sido modificado.

Criando o Ingresso

Para criar um ingresso, incluindo automaticamente um registro de detalhes vazio para manter a consistência no banco de dados, podemos modificar a função de criação de ingressos da seguinte forma:

python
Copy code
async def create_ticket(
    db_session: AsyncSession,
    show_name: str,
    user: str = None,
    price: float = None,
) -> int:
    ticket = Ticket(
        show=show_name,
        user=user,
        price=price,
        details=TicketDetails(),
    )
    async with db_session.begin():
        db_session.add(ticket)
        await db_session.flush()
        ticket_id = ticket.id
        await db_session.commit()
    return ticket_id

Esse exemplo garante que sempre que um ingresso é criado, um registro vazio de detalhes de ingresso também seja inserido no banco de dados.

Relacionamentos de Muitos-para-Um

Em um sistema de ingressos, é comum que um ingresso esteja associado a um evento, e que um evento tenha vários ingressos. Esse é um exemplo clássico de um relacionamento de muitos-para-um. Para isso, vamos adicionar uma referência ao evento na tabela de ingressos:

python
Copy code
class Ticket(Base):

    __tablename__ = "tickets"
    # outros campos existentes
    event_id: Mapped[int | None] = mapped_column(ForeignKey("events.id"))
    event: Mapped["Event | None"] = relationship(back_populates="tickets")

A seguir, criamos a tabela Event para mapear os eventos:

python
Copy code
class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    tickets: Mapped[list["Ticket"]] = relationship(back_populates="event")

Com essa estrutura, um evento pode ter múltiplos ingressos associados a ele, mas cada ingresso estará vinculado a um único evento.

Relacionamentos de Muitos-para-Muitos

Em alguns casos, um evento pode ser patrocinado por múltiplos patrocinadores, e um patrocinador pode apoiar vários eventos. Esse é um exemplo de relacionamento de muitos-para-muitos, que requer o uso de uma tabela intermediária, também conhecida como tabela de associação.

Primeiro, adicionamos o campo de relacionamento na tabela Event:

python
Copy code
class Event(Base):

    __tablename__ = "events"
    # outros campos existentes
    sponsors: Mapped[list["Sponsor"]] = relationship(
        secondary="sponsorships", back_populates="events"
    )

Em seguida, criamos a tabela Sponsor para mapear os patrocinadores:

python
Copy code
class Sponsor(Base):
    __tablename__ = "sponsors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    events: Mapped[list["Event"]] = relationship(
        secondary="sponsorships", back_populates="sponsors"
    )

Por fim, a tabela de associação Sponsorship contém informações sobre o relacionamento entre o patrocinador e o evento, como o valor patrocinado:

python
Copy code
class Sponsorship(Base):

    __tablename__ = "sponsorships"
    event_id: Mapped[int] = mapped_column(ForeignKey("events.id"), primary_key=True)
    sponsor_id: Mapped[int] = mapped_column(ForeignKey("sponsors.id"), primary_key=True)
    amount: Mapped[float] = mapped_column(nullable=False, default=10)

Essa estrutura facilita o gerenciamento de múltiplos patrocinadores para um evento e vice-versa, além de permitir a inclusão de dados adicionais, como o valor da contribuição do patrocinador.

Considerações Finais

A implementação e a manipulação de diferentes tipos de relacionamentos em bancos de dados SQL são fundamentais para garantir que o sistema seja flexível e escalável. No exemplo do sistema de ingressos, vimos como implementar relacionamentos de um-para-um, muitos-para-um e muitos-para-muitos, o que proporciona uma maneira eficiente de organizar dados relacionados entre si. Esses conceitos são essenciais não apenas para sistemas de ingressos, mas para qualquer aplicação que envolva entidades interconectadas, como sistemas de comércio eletrônico, gestão de eventos e muito mais.

Além disso, ao lidar com dados relacionais, é importante considerar o desempenho das consultas. Consultas eficientes não só melhoram a experiência do usuário, mas também reduzem custos operacionais e aumentam a capacidade de escalabilidade do sistema. A otimização das consultas SQL deve ser um processo contínuo, onde pequenas melhorias podem ter um impacto significativo no desempenho geral do sistema.

Como integrar o OAuth2 para proteger as conexões WebSocket no FastAPI

Primeiramente, é necessário definir o objeto oauth2_scheme_for_ws que será utilizado para a autenticação. No exemplo abaixo, estamos utilizando a classe OAuth2WebSocketPasswordBearer para criar esse objeto:

python
Copy code
from app.ws_password_bearer import OAuth2WebSocketPasswordBearer
oauth2_scheme_for_ws = OAuth2WebSocketPasswordBearer(
    tokenUrl="/token"
)

O argumento tokenUrl especifica o ponto de extremidade que será utilizado para recuperar o token de autenticação. Este ponto de extremidade deve ser desenvolvido de acordo com o método de resolução de token que você utilizará em sua aplicação.

Em seguida, é possível criar uma função que recupere o nome de usuário a partir do token. O exemplo a seguir ilustra como isso pode ser feito, simulando a resolução de um token:

python
Copy code
def get_username_from_token(

    token: str = Depends(oauth2_scheme_for_ws),
) -> str:
    user = fake_token_resolver(token)
    if not user:
        raise WebSocketException(
            code=status.HTTP_401_UNAUTHORIZED,
            reason="Invalid authentication credentials"
        )
    return user.username

A função fake_token_resolver tem como objetivo simular o processo de resolução de um token. Ela é apenas uma implementação de exemplo e não oferece segurança real. No mundo real, seria ideal utilizar tokens JWT ou um provedor externo de autenticação para resolver tokens de forma segura, como é discutido nos capítulos sobre OAuth2 e JWT.

Agora que temos a função para recuperar o nome de usuário, podemos aplicar a segurança no endpoint WebSocket /secured-ws. O código a seguir mostra como isso pode ser feito no arquivo main.py:

python
Copy code
from fastapi import Depends, WebSocket

from app.security import get_username_from_token
@app.websocket("/secured-ws")
async def secured_websocket(
    websocket: WebSocket,
    username: str = Depends(get_username_from_token)
):
    # restante do endpoint

Esse código aplica a segurança ao ponto de extremidade WebSocket, exigindo um token de autenticação válido para acessar a conexão. Ao tentar se conectar a esse WebSocket utilizando uma ferramenta como o Postman, o usuário será rejeitado se não fornecer um token válido. O erro de autorização ocorrerá antes do início da conexão, evitando que dados sensíveis sejam transmitidos.

Para testar a conexão, é necessário recuperar o token de autenticação e incluí-lo no cabeçalho da solicitação WebSocket no Postman. O token pode ser obtido por meio de um ponto de extremidade dedicado ou, se você estiver utilizando a função de geração de tokens falsa do repositório do GitHub, basta anexar a string tokenizada ao nome de usuário. Por exemplo, para o usuário johndoe, o token seria tokenizedjohndoe. Esse token deve ser incluído no cabeçalho da solicitação WebSocket, como mostrado abaixo:

• Chave: Authorization

• Valor: Bearer tokenizedjohndoe

Essa abordagem de usar OAuth2 para proteger WebSockets oferece uma camada adicional de segurança em suas aplicações FastAPI. No entanto, vale ressaltar que esta solução é ideal para fins de exemplo e não deve ser utilizada em ambientes de produção sem a devida implementação de segurança adequada, como o uso de tokens JWT em vez de tokens fictícios.

Ao implementar essa funcionalidade, você garante que somente usuários autenticados poderão acessar as comunicações WebSocket, oferecendo uma camada extra de segurança.

Como integrar FastAPI com modelos de aprendizado de máquina e APIs externas

No processo de desenvolvimento de sistemas baseados em FastAPI, a integração com modelos de aprendizado de máquina (ML) pode proporcionar funcionalidades poderosas e interativas. Um exemplo disso é a criação de um sistema de diagnóstico médico baseado em sintomas, onde a API recebe parâmetros como sintomas e retorna um diagnóstico usando um modelo ML previamente treinado. Para implementar isso, começamos criando o endpoint que receberá os sintomas como parâmetros e retornará o diagnóstico adequado.

O primeiro passo é a criação de um objeto de parâmetros dinâmico usando o Pydantic. Considerando que temos 132 sintomas possíveis, mas limitando-nos aos 10 primeiros para facilitar a interação, podemos criar a classe Symptoms com base na lista de sintomas. Para isso, utilizamos a função create_model do Pydantic, que cria modelos dinamicamente. O código abaixo mostra como criar esse modelo:

python
Copy code
from pydantic import create_model

from app.utils import symptoms_list
query_params = { 
    symp: (bool, False) 
    for symp in symptoms_list[:10] 
}
Symptoms = create_model("Symptoms", **query_params)

Esse modelo agora aceita parâmetros dinâmicos para os 10 primeiros sintomas, permitindo que o usuário envie esses dados como parte da requisição. A seguir, criamos o endpoint GET /diagnosis, que utilizará esse modelo de sintomas para gerar o diagnóstico.

python
Copy code
@app.get("/diagnosis")
async def get_diagnosis(
    symptoms: Annotated[Symptoms, Depends()],
):
    array = [
        int(value)
        for _, value in symptoms.model_dump().items()
    ]
    array.extend(
        [0] * (len(symptoms_list) - len(array))
    )
    diseases = ml_model["doctor"].predict([array])
    return {
        "diseases": [disease for disease in diseases]
    }

Com esse processo, não apenas é possível integrar um único modelo de aprendizado de máquina em uma aplicação FastAPI, mas também é viável integrar múltiplos modelos na mesma aplicação, utilizando a mesma abordagem. Isso oferece uma enorme flexibilidade, permitindo o uso de diferentes modelos de ML conforme a necessidade.

Além disso, uma prática importante ao integrar modelos de ML em sistemas de produção é garantir que o modelo utilizado esteja atualizado e treinado com dados relevantes. Caso contrário, a precisão do diagnóstico ou da recomendação pode ser comprometida. Isso torna essencial a manutenção contínua e a atualização dos modelos para manter a eficácia da aplicação.

python
Copy code
from cohere import AsyncClient
from dotenv import load_dotenv
load_dotenv()
SYSTEM_MESSAGE = (
    "Você é um chef especializado em culinária italiana, "
    "com experiência em receitas tradicionais e truques de cozinha."
)
client = AsyncClient()

async def generate_chat_completion(user_query=" ", messages=[]):

    try:
        response = await client.chat(
            message=user_query,
            model="command-r-plus",
            preamble=SYSTEM_MESSAGE,
            chat_history=messages,
        )
        messages.extend([
            ChatMessage(role="USER", message=user_query),
            ChatMessage(role="CHATBOT", message=response.text)
        ])
        return response.text
    except ApiError as e:
        raise HTTPException(status_code=e.status_code, detail=e.body)

Esse sistema de chatbot pode ser alimentado com uma consulta do usuário, que receberá uma resposta em tempo real, como se estivesse conversando com um chef de cozinha. Esse exemplo de integração com o Cohere é particularmente útil para criar assistentes virtuais especializados, que vão além das respostas simples e conseguem gerar interações complexas e contextuais.