Como Realizar Testes de Performance em Aplicações de Alta Demanda com Locust e FastAPI

Os testes de performance são essenciais para garantir que sua aplicação seja capaz de suportar cenários reais de uso, especialmente sob grande carga de tráfego. A implementação sistemática de testes de performance, a análise dos resultados e a otimização com base nas conclusões obtidas podem melhorar significativamente a responsividade, estabilidade e escalabilidade de seu sistema.

Preparando o Ambiente de Testes

Para começar, é necessário garantir que o Locust esteja instalado em seu ambiente virtual. Isso pode ser feito com o comando:

bash
Copy code
$ pip install locust

Aqui está um exemplo básico de como o conteúdo do arquivo pode ser estruturado:

python
Copy code
from locust import HttpUser, task
class ProtoappUser(HttpUser):
    host = "http://localhost:8000"
    @task
    def hello_world(self):
        self.client.get("/home")

Neste arquivo, definimos uma classe ProtoappUser, que representa um usuário acessando a aplicação em teste. O endereço do host é o da aplicação local, e a tarefa associada é uma requisição GET para o endpoint "/home".

Executando o Teste de Performance

Com o arquivo locustfile.py configurado, o próximo passo é iniciar o servidor FastAPI. Para isso, execute o comando:

bash
Copy code
$ uvicorn protoapp.main:app

Em seguida, em uma nova janela do terminal, execute o Locust com o seguinte comando:

bash
Copy code
$ locust

Além disso, o Locust pode ser executado em modo headless, ou seja, sem a interface web. Nesse caso, o comando seria:

bash
Copy code
$ locust --headless --users 10 --spawn-rate 1

Neste exemplo, o comando simula 10 usuários acessando a aplicação ao mesmo tempo, com uma taxa de 1 usuário por segundo.

Integrando os Testes de Performance ao Pipeline CI/CD

O Locust pode ser incorporado ao pipeline CI/CD para rodar testes a cada atualização de código, simulando o tráfego e garantindo que o desempenho da aplicação seja mantido ou melhorado com o tempo. Ao adotar essa abordagem, é possível realizar testes contínuos, tornando o processo de desenvolvimento mais robusto e eficiente.

Monitoramento e Análise dos Resultados

Além disso, é recomendável realizar uma análise pós-teste para comparar os resultados com os objetivos de performance da aplicação, como tempos de resposta aceitáveis e a quantidade máxima de usuários simultâneos. Esse processo de análise e ajuste contínuo é vital para manter a qualidade do software em longo prazo.

O Uso do Locust para Simulação de Tráfego

O Locust não se limita a testes simples de carga. A ferramenta permite a criação de cenários de tráfego mais complexos, simulando diferentes tipos de usuários e comportamentos. Por exemplo, é possível criar tarefas específicas que imitam navegações mais complexas, interações com formulários ou até mesmo chamadas de APIs externas.

Isso oferece uma flexibilidade significativa, permitindo que os testes se aproximem mais da experiência real de uso de uma aplicação, o que é crucial para identificar problemas que poderiam surgir em cenários de alta demanda.

Segurança e Escalabilidade

Além de realizar testes de performance, outro aspecto fundamental para aplicações de alta demanda é garantir a segurança e escalabilidade. Embora o Locust seja excelente para avaliar o desempenho da aplicação sob carga, é igualmente importante verificar como a aplicação se comporta quando submetida a diferentes tipos de ataques, como DoS (Denial of Service), e como ela gerencia o aumento de usuários de maneira eficiente.

Como Integrar o FastAPI com Redis para Melhoria de Desempenho e Cache de Dados

Ao desenvolver aplicações web eficientes e de alta performance, uma das melhores práticas é utilizar cache para armazenar dados temporários que são frequentemente acessados. O FastAPI, um framework Python de alto desempenho para a construção de APIs, possui integração fácil com diversos sistemas de cache, incluindo Redis. A seguir, detalhamos como você pode utilizar o Redis para otimizar a performance de um endpoint do FastAPI, melhorando assim a experiência do usuário e a escalabilidade da aplicação.

Em primeiro lugar, a integração de cache com Redis em FastAPI começa com a verificação da existência de dados armazenados no Redis antes de buscar as informações de uma fonte primária, como um banco de dados ou um sistema de pesquisa como o Elasticsearch. No caso do nosso exemplo, vamos considerar que desejamos armazenar a lista dos "dez principais artistas" de um país específico. O fluxo de trabalho para isso é simples:

1. Inicialmente, verificamos se a chave associada ao país já existe no Redis. Se a chave estiver presente, os dados são retornados diretamente do cache, o que garante um tempo de resposta mais rápido para o usuário. Isso é feito com o comando:

   python
   Copy code
   cache_key = f"top_ten_artists_{country}"
   cached_data = await redis_client.get(cache_key)
   if cached_data:
       logger.info(f"Returning cached data for {country}")
       return json.loads(cached_data)
   

2. Caso os dados não estejam disponíveis no cache, buscamos as informações diretamente no Elasticsearch. Essa busca pode gerar uma exceção se a consulta for inválida (por exemplo, quando um país não existe ou a consulta retorna um erro). Para lidar com isso, capturamos a exceção e retornamos uma mensagem de erro para o usuário:

   python
   Copy code
   try:
       response = await es_client.search(*top_ten_artists_query(country))
   except BadRequestError as e:
       logger.error(e)
       raise HTTPException(status_code=400, detail="Invalid country")
   

3. Após a consulta bem-sucedida ao Elasticsearch, extraímos os dados relevantes e os formatamos de acordo com o que precisamos para a nossa aplicação. A lista de artistas é montada e pronta para ser armazenada no Redis, para que possa ser utilizada em futuras requisições sem a necessidade de uma nova consulta ao Elasticsearch.

   python
   Copy code
   artists = [
   
       {"artist": record.get("key"), "views": record.get("views", {}).get("value")}
   
       for record in response["aggregations"]["top_ten_artists"]["buckets"]
   ]
   

4. A última etapa é armazenar os dados no Redis com um tempo de expiração. Isso significa que, após um período de tempo pré-definido, os dados serão automaticamente removidos do cache, forçando a consulta ao Elasticsearch novamente. Esse tempo de expiração é importante para garantir que o cache não seja usado por um período excessivo, evitando a exibição de dados desatualizados:

   python
   Copy code
   await redis_client.set(cache_key, json.dumps(artists), ex=3600)
   

   A partir desse momento, qualquer nova requisição ao mesmo endpoint durante a próxima hora (3600 segundos) retornará os dados diretamente do Redis, economizando tempo de processamento e melhorando a performance da aplicação.

A grande vantagem dessa estratégia é a significativa redução do tempo de resposta em chamadas subsequentes, já que a consulta ao Elasticsearch, que pode ser relativamente lenta dependendo do tamanho dos dados, é evitada. Em vez disso, o Redis fornece os dados de maneira muito mais rápida. Quando o tempo de expiração do cache é atingido, a aplicação automaticamente faz uma nova consulta ao Elasticsearch e armazena os dados atualizados.

Ao trabalhar com Redis e FastAPI, é essencial lembrar que o gerenciamento de cache deve ser feito com cautela. É importante definir corretamente o tempo de expiração para evitar que os dados fiquem desatualizados. Em sistemas onde a frequência de atualização dos dados é alta, o uso de um tempo de expiração menor pode ser preferível. Por outro lado, em sistemas mais estáveis, onde os dados não mudam frequentemente, é possível aumentar esse tempo, economizando recursos.

Outro ponto a considerar é a possível complexidade do gerenciamento do cache quando múltiplos micro-serviços ou instâncias de servidores estão envolvidos. A configuração de um Redis em cluster ou o uso de mecanismos de cache distribuído pode ser necessário para garantir que todos os servidores tenham acesso consistente aos dados em cache.

Em suma, a integração do FastAPI com Redis oferece uma solução poderosa para otimizar a performance de APIs, especialmente em cenários onde os dados são frequentemente acessados mas raramente alterados. Com uma configuração simples e práticas adequadas de gerenciamento de cache, é possível criar aplicações web rápidas e escaláveis, proporcionando uma experiência mais eficiente para os usuários.

Como Gerenciar Conexões WebSocket no FastAPI

Durante o processo de conexão, o servidor deve verificar a mensagem recebida: "Welcome to the chat room!". Isso indica que o ponto de extremidade WebSocket foi criado corretamente e está funcionando adequadamente. O parâmetro WebSocket no ponto de extremidade representa uma conexão WebSocket individual. Quando o método websocket.accept() é aguardado, o servidor estabelece a conexão com o cliente (um processo tecnicamente conhecido como "handshake" HTTP). Em seguida, o comando websocket.send_text() envia uma mensagem para o cliente, e finalmente, websocket.close() encerra a conexão.

Essa configuração, embora mínima e com pouca utilidade prática, representa o básico para uma conexão WebSocket. Em seguida, veremos como trocar mensagens entre o cliente e o servidor por meio de um ponto de extremidade WebSocket.

As conexões WebSocket permitem comunicação bidirecional entre clientes e servidores, possibilitando a troca em tempo real de mensagens. Com essa funcionalidade, é possível criar aplicações como salas de bate-papo, onde as mensagens são enviadas e recebidas de maneira contínua e dinâmica. No exemplo que iremos seguir, configuraremos um ponto de extremidade WebSocket para que o servidor possa receber mensagens do cliente e imprimi-las no terminal.

Para implementar isso, começamos definindo o "logger" (registro de logs) usando o pacote uvicorn, que é o mesmo utilizado pelo FastAPI para gerenciamento de logs. No arquivo main.py, escrevemos o seguinte código:

python
Copy code
import logging
logger = logging.getLogger("uvicorn")

Agora, vamos modificar a função ws_endpoint, responsável pelo ponto de extremidade WebSocket:

python
Copy code
@app.websocket("/ws")

async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    while True:
        data = await websocket.receive_text()
        logger.info(f"Message received: {data}")
        await websocket.send_text("Message received!")

Para testar, inicie o servidor com o comando uvicorn app.main:app no terminal e abra o Postman. Em seguida, crie uma nova requisição WebSocket e conecte-se ao endereço ws://localhost:8000/ws. Quando a conexão for estabelecida, você verá no terminal a mensagem: "INFO: connection open", e na resposta do Postman aparecerá: "Welcome to the chat room!". Agora, envie uma mensagem do Postman, como "Hello FastAPI application". Você verá no terminal: "INFO: Message received: Hello FastAPI application", e no painel de mensagens do cliente, a resposta será: "Message received!".

Agora que o servidor está configurado para receber mensagens do cliente, você tem uma comunicação bidirecional funcionando por meio de WebSockets. Esse tipo de comunicação em tempo real é a base para muitas aplicações modernas, como chats, jogos online, e outros sistemas que exigem troca instantânea de dados entre o cliente e o servidor.

Ao desenvolver aplicações com WebSockets, é essencial também gerenciar as desconexões corretamente. Isso garante que a comunicação entre cliente e servidor não seja interrompida de maneira abrupta, o que poderia causar falhas no sistema.

python
Copy code
from fastapi.websockets import WebSocketDisconnect
@app.websocket("/ws")

async def ws_endpoint(websocket: WebSocket):

    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    try:
        while True:
            data = await websocket.receive_text()
            logger.info(f"Message received: {data}")
            await websocket.send_text("Message received!")
    except WebSocketDisconnect:
        logger.warning("Connection closed by the client")

Agora, se o cliente se desconectar, o servidor não gerará um erro, e em vez disso, registrará uma mensagem de aviso: "Connection closed by the client". Isso permite que o servidor continue funcionando sem falhas mesmo quando o cliente decide encerrar a conexão.

Além disso, o servidor também pode encerrar a conexão de forma controlada, por exemplo, ao receber uma mensagem específica, como a palavra "disconnect". Veja como implementar isso:

python
Copy code
@app.websocket("/ws")

async def ws_endpoint(websocket: WebSocket):

    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    while True:
        data = await websocket.receive_text()
        if data == "disconnect":
            await websocket.send_text("Disconnecting...")
            break
        logger.info(f"Message received: {data}")
        await websocket.send_text("Message received!")

Nesse caso, quando o servidor receber a mensagem "disconnect", ele enviará uma resposta ao cliente e encerrará a conexão de forma limpa. Esse tipo de controle é essencial para aplicações que exigem um gerenciamento eficaz das conexões WebSocket, garantindo uma comunicação contínua e sem falhas.

O protocolo WebSocket se destaca por sua eficiência na troca de dados em tempo real, oferecendo vantagens significativas em relação a outras abordagens de comunicação cliente-servidor, como long polling ou requisições periódicas via HTTP.

Como Integrar o FastAPI com o LangChain para Criar Assistentes de IA Dinâmicos

O LangChain é uma interface versátil para trabalhar com Modelos de Linguagem de Grande Escala (LLM), permitindo que desenvolvedores criem aplicativos de LLM e os integrem com fontes de dados externas e fluxos de trabalho de software. Desde seu lançamento em 2022, tornou-se rapidamente um dos projetos de código aberto mais populares no GitHub. Neste capítulo, exploraremos como integrar o FastAPI com o LangChain para criar um assistente de IA capaz de fornecer recomendações dinâmicas em uma loja de eletrônicos. A proposta é criar uma aplicação que utilize um modelo de Geração Aumentada por Recuperação (RAG), utilizando dados personalizados, como um documento de perguntas frequentes (FAQ), para treinar o modelo.

Além do fastapi e uvicorn, será preciso instalar pacotes adicionais para trabalhar com LangChain e com a integração do Cohere. Você pode instalar os pacotes usando o arquivo requirements.txt ou diretamente com o pip:

ruby
Copy code
$ pip install fastapi uvicorn python-dotenv
$ pip install langchain
$ pip install langchain-community langchain-cohere
$ pip install chromadb unstructured

1. Definindo os Prompts

Para este caso específico, vamos usar um modelo de chat que recebe uma lista de mensagens como entrada. O LangChain oferece objetos de template para mensagens específicas, como a mensagem do sistema e a mensagem do usuário. A primeira parte do processo envolve definir os prompts a serem usados.

python
Copy code
template: str = """
Você é um chatbot de suporte ao cliente.
Você auxilia os usuários com dúvidas gerais e problemas técnicos.
Você responderá à pergunta: {question}
Sua resposta será baseada apenas no conhecimento do contexto abaixo com o qual você foi treinado.
-----------
{context}
-----------
Se você não souber a resposta, pedirá ao usuário para reformular a pergunta ou o redirecionará para o suporte em [email protected]
Seja sempre amigável e útil. Ao final da conversa, pergunte ao usuário se ele está satisfeito com a resposta.
Se a resposta for positiva, diga adeus e finalize a conversa.
"""

Com esse template, podemos definir a mensagem do sistema como segue:

python
Copy code
from langchain.prompts import SystemMessagePromptTemplate

system_message_prompt = SystemMessagePromptTemplate.from_template(template)

A mensagem do usuário não precisa de contexto adicional e pode ser configurada assim:

python
Copy code
from langchain.prompts import HumanMessagePromptTemplate
human_message_prompt = HumanMessagePromptTemplate.from_template("{question}")

Por fim, as duas mensagens podem ser agrupadas sob um objeto de template de chat, que unifica os prompts:

python
Copy code
from langchain.prompts import ChatPromptTemplate

chat_prompt_template = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])

2. Processando e Vetorizando os Documentos

Para que nosso assistente de IA responda de forma inteligente às perguntas dos usuários, ele precisa analisar documentos específicos. Vamos criar uma pasta docs no diretório raiz do projeto para armazenar esses documentos. O arquivo de FAQ, por exemplo, pode ser baixado diretamente de um repositório GitHub ou criado manualmente. A única regra é garantir que cada pergunta e resposta estejam separadas por uma linha em branco.

No entanto, para otimizar as buscas por similaridade, precisaremos dividir os documentos em pedaços menores e armazená-los como vetores. O LangChain facilita isso com seu banco de dados vetorial em memória, o ChromaDB. Para dividir os documentos, utilizamos um "text splitter" baseado em caracteres, que divide o texto de acordo com um tamanho pré-definido.

3. Construindo a Cadeia do Modelo

4. Criando o Endpoint

A última etapa é criar o endpoint no FastAPI para que o assistente de IA possa interagir com os usuários. Para isso, utilizamos a funcionalidade de gerenciamento de contexto do FastAPI, que permite que o estado da aplicação seja mantido entre as requisições. Assim, a cada nova consulta, o assistente mantém o histórico de mensagens e pode gerar respostas mais contextuais.

python
Copy code
from fastapi import FastAPI, Body, Request

from typing import Annotated
from handlers import generate_chat_completion
app = FastAPI(title="Assistente de Compras Ecotech", lifespan=lifespan)
@app.post("/query")
async def query_chat_bot(request: Request, query: Annotated[str, Body(min_length=1)]) -> str:
    answer = await generate_chat_completion(query, request.state.messages)
    return answer

Esse endpoint lida com requisições POST, onde a consulta do usuário é passada como parâmetro. A resposta do modelo é retornada ao usuário, criando uma interação dinâmica e contínua.

Além disso, uma vez que o assistente esteja em operação, seria interessante adicionar funcionalidades extras, como um endpoint GET /messages que retorna todas as mensagens trocadas ou um endpoint POST /restart-conversation que reinicia a conversa, apagando todas as mensagens anteriores.

Considerações Finais

Como Implementar Middleware de Segurança e Webhooks em FastAPI

FastAPI é uma poderosa ferramenta para criar APIs rápidas e eficientes, mas, como qualquer aplicação web, é importante garantir que apenas fontes confiáveis possam acessar a aplicação e que eventos em tempo real possam ser comunicados de forma eficiente entre sistemas. Dois conceitos essenciais nesse contexto são o middleware de segurança, como o TrustedHostMiddleware, e a implementação de webhooks para comunicação assíncrona. Neste capítulo, vamos explorar como implementar essas funcionalidades em FastAPI, garantindo a segurança da aplicação e a integração com sistemas externos.

O middleware de segurança TrustedHostMiddleware é uma ferramenta fundamental para garantir que a sua aplicação web só aceite requisições de hosts confiáveis. Ao ativar esse middleware, podemos definir explicitamente quais hosts podem interagir com a aplicação, evitando requisições maliciosas ou indesejadas. No exemplo abaixo, configuramos o middleware para permitir apenas requisições originadas de "localhost":

python
Copy code
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["localhost"],
)

Essa configuração impede que a aplicação aceite requisições de qualquer outro host, protegendo-a de acessos indesejados. Se um host não autorizado tentar acessar a aplicação, a resposta será uma mensagem de erro, como "Invalid host header". Esse tipo de segurança é crucial para evitar ataques do tipo DNS rebinding ou outras formas de acesso indevido. Para testar essa configuração, basta executar o servidor com o comando:

bash
Copy code
$ uvicorn main:app --host=0.0.0.0

Isso tornará a aplicação acessível pela rede local, mas com o middleware em funcionamento, qualquer tentativa de acesso de um host não permitido resultará em erro.

Além disso, a integração com webhooks tem se tornado cada vez mais comum nas arquiteturas modernas. Webhooks são mecanismos que permitem a comunicação assíncrona entre sistemas, ou seja, quando um evento ocorre em um sistema, ele pode notificar outros sistemas em tempo real. Por exemplo, imagine que você tenha uma API que deseja notificar outros serviços sempre que uma requisição for realizada. Abaixo, veremos como configurar esse sistema em FastAPI.

Primeiro, criamos um sistema simples de registro de URLs de webhooks. Quando um serviço deseja se inscrever para receber notificações de eventos, ele registra sua URL no sistema. Para isso, utilizamos um endpoint que armazena as URLs em memória, usando o state da aplicação:

python
Copy code
from fastapi import FastAPI, Request, Body

from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield {"webhook_urls": []}
app = FastAPI(lifespan=lifespan)
@app.post("/register-webhook-url")
async def add_webhook_url(request: Request, url: str = Body()):
    if not url.startswith("http"):
        url = f"http://{url}"
    request.state.webhook_urls.append(url)
    return {"url added": url}

Essa configuração permite que os sistemas se registrem para receber notificações, enviando a URL onde desejam ser notificados. O próximo passo é definir o comportamento dos callbacks de webhook. Sempre que um evento ocorrer, a aplicação precisa enviar uma notificação para todos os URLs registrados. Isso é feito com o seguinte código, que usa a biblioteca httpx para fazer requisições assíncronas aos URLs registrados:

python
Copy code
import logging

from httpx import AsyncClient
from pydantic import BaseModel
client = AsyncClient()
logger = logging.getLogger("uvicorn")
class Event(BaseModel):
    host: str
    path: str
    time: str
    body: str | None = None
async def send_event_to_url(url: str, event: Event):
    logger.info(f"Sending event to {url}")
    try:
        await client.post(
            f"{url}/fastapi-webhook",
            json=event.dict(),
        )
    except Exception as e:
        logger.error(f"Error sending webhook event to {url}: {e}")

Agora, vamos configurar o middleware para interceptar todas as requisições HTTP e, sempre que uma requisição ocorrer, notificar todos os URLs registrados. O middleware é responsável por criar o evento e disparar as notificações para os sistemas interessados:

python
Copy code
from asyncio import create_task

from datetime import datetime
from fastapi import Request
from starlette.types import ASGIApp, Receive, Scope, Send
class WebhookSenderMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        if scope["type"] == "http":
            message = await receive()
            body = message.get("body", b"")
            request = Request(scope=scope)
            event = Event(
                host=request.client.host,
                path=request.url.path,
                time=datetime.now().isoformat(),
                body=body,
            )
            urls = request.state.webhook_urls
            for url in urls:
                await create_task(send_event_to_url(url, event))
        await self.app(scope, receive, send)

É importante notar que, embora o exemplo fornecido use a memória da aplicação para armazenar os URLs dos webhooks, em aplicações reais, é recomendável usar um banco de dados ou outra solução persistente para armazenar essas URLs. Além disso, deve-se implementar uma lógica de tratamento de falhas, como re-tentativas em caso de falha de rede ou de resposta do sistema receptor, para garantir que os eventos não sejam perdidos.

A implementação de middleware de segurança, como o TrustedHostMiddleware, e a configuração de webhooks permitem que sua aplicação FastAPI seja mais segura e interativa. Esses componentes são cruciais para garantir que apenas fontes confiáveis possam acessar a aplicação e para permitir a comunicação em tempo real com sistemas externos. Essas práticas são fundamentais para sistemas modernos e escaláveis, que exigem integração com outros serviços e notificações em tempo real.