Como otimizar operações assíncronas com FastAPI: melhores práticas e considerações importantes

Quando se trata de construir aplicações web eficientes e escaláveis, a programação assíncrona é um recurso indispensável, especialmente quando lidamos com operações de I/O que exigem tempo, como acessos a banco de dados, chamadas de API e operações de rede. Abaixo, exploraremos como podemos melhorar o desempenho da nossa aplicação utilizando FastAPI, considerando a implementação de funções assíncronas para tarefas concorrentes.

Primeiramente, ao utilizar FastAPI, a capacidade de executar funções assíncronas de maneira eficiente pode ser vista no uso de bibliotecas como asyncio, que permite que múltiplas tarefas sejam executadas ao mesmo tempo, sem bloquear o processo principal. Isso é de especial importância em operações de rede ou em transações de banco de dados, que muitas vezes podem ser lentas devido à latência. O benefício imediato do uso de asyncio e funções assíncronas é a resposta mais rápida da aplicação, já que o servidor pode processar múltiplas requisições enquanto aguarda respostas externas.

python
Copy code
from contextlib import contextmanager
from multiprocessing import Process
@contextmanager
def run_server_in_process():
    p = Process(target=run_server)
    p.start()
    time.sleep(2)  # Damos um tempo para o servidor iniciar
    print("Servidor rodando em um processo separado")
    yield
    p.terminate()

python
Copy code
async def make_requests_to_the_endpoint(n: int, path: str):

    async with AsyncClient(base_url="http://localhost:8000") as client:

        tasks = (client.get(path, timeout=float("inf")) for _ in range(n))
        await asyncio.gather(*tasks)

Em seguida, podemos organizar essas chamadas em uma função principal, onde realizamos n requisições para diferentes endpoints (sincronos e assíncronos) e calculamos o tempo necessário para atender a todas as requisições:

python
Copy code
async def main(n: int = 10):

    with run_server_in_process():
        begin = time.time()
        await make_requests_to_the_endpoint(n, "/sync")
        end = time.time()
        print(f"Tempo para {n} requisições no endpoint síncrono: {end - begin} segundos")
        begin = time.time()
        await make_requests_to_the_endpoint(n, "/async")
        end = time.time()

        print(f"Tempo para {n} requisições no endpoint assíncrono: {end - begin} segundos")

Ao executar a função main com o número padrão de requisições (10), podemos perceber que o tempo para ambos os endpoints (síncrono e assíncrono) é quase o mesmo. Contudo, quando aumentamos a quantidade de requisições, como para 100, a diferença começa a se tornar visível:

python
Copy code
if __name__ == "__main__":

    asyncio.run(main(n=100))

Aqui, o uso de programação assíncrona para realizar múltiplas requisições simultâneas resulta em um tempo significativamente menor para o endpoint assíncrono em comparação com o síncrono. Isso ilustra como operações assíncronas podem ser mais eficazes quando lidamos com tarefas que envolvem espera, como chamadas a bancos de dados ou APIs externas.

Entretanto, nem todas as operações se beneficiam da programação assíncrona. É importante entender que as operações assíncronas são mais vantajosas para tarefas de I/O e que o processamento intensivo de CPU (como operações de cálculos complexos) pode não ter um ganho significativo com a assíncronicidade. Além disso, ao integrar operações assíncronas em sua aplicação, é necessário tomar certos cuidados para garantir a integridade e a performance da aplicação.

A programação assíncrona também exige que tenhamos uma gestão cuidadosa de transações e sessões, pois as mudanças de estado podem ser feitas de forma concorrente, o que pode causar inconsistências se não forem gerenciadas corretamente. O uso de gerenciadores de contexto, como async with, ajuda a garantir que as transações sejam iniciadas e finalizadas de maneira adequada, preservando a integridade dos dados.

Além disso, um dos desafios da programação assíncrona é o tratamento de erros. A simultaneidade de múltiplas tarefas torna o manuseio de exceções mais complexo, sendo necessário utilizar blocos try-except para capturar e lidar com falhas que possam surgir durante a execução das tarefas assíncronas.

Outro ponto importante é a realização de testes. A execução de testes assíncronos exige que o framework de testes seja compatível com a programação assíncrona, utilizando async e await para garantir que o código seja testado corretamente.

A adoção da programação assíncrona, portanto, não se resume apenas à escrita do código com async e await, mas envolve uma série de boas práticas para garantir que a aplicação se beneficie de um desempenho superior, enquanto mantém a robustez, a segurança e a integridade. Com a combinação adequada de funções assíncronas, tratamento de erros, testes e gestão de transações, podemos construir sistemas que respondem de forma eficiente, mesmo sob alta carga de trabalho.

Como Testar e Depurar Endpoints de API com FastAPI

A criação de endpoints eficazes é apenas o primeiro passo na construção de uma aplicação robusta. A verdadeira força de um sistema vem da garantia de que ele funciona corretamente em diferentes cenários. Este processo exige uma combinação de boas práticas de desenvolvimento e uma abordagem rigorosa de testes e depuração. Quando usamos o FastAPI, uma das ferramentas mais poderosas para a construção de APIs rápidas e eficientes, entender como configurar e testar adequadamente os endpoints pode ser a chave para garantir o sucesso do projeto. Vamos explorar como fazer isso de maneira eficiente.

python
Copy code
from pydantic import BaseModel
class ItemSchema(BaseModel):
    name: str
    color: str

Com essa definição, podemos construir um endpoint para adicionar um item. Esse endpoint será responsável por receber dados, armazená-los em um banco de dados e retornar o ID do item. Aqui está um exemplo de como implementá-lo:

python
Copy code
from fastapi import Depends, Request, HTTPException, status

from sqlalchemy.orm import Session
@app.post("/item", response_model=int, status_code=status.HTTP_201_CREATED)
def add_item(item: ItemSchema, db_session: Session = Depends(get_db_session)):
    db_item = Item(name=item.name, color=item.color)
    db_session.add(db_item)
    db_session.commit()
    db_session.refresh(db_item)
    return db_item.id

python
Copy code
@app.get("/item/{item_id}", response_model=ItemSchema)

def get_item(item_id: int, db_session: Session = Depends(get_db_session)):

    item_db = db_session.query(Item).filter(Item.id == item_id).first()
    if item_db is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item_db

Aqui, o FastAPI irá buscar o item no banco de dados, e caso ele não seja encontrado, um erro 404 será retornado.

Agora que os endpoints para adição e recuperação de itens estão prontos, o próximo passo é realizar testes adequados. Para testar essas funções, precisamos garantir que o banco de dados de teste seja configurado corretamente. Para isso, uma boa prática é usar um banco de dados SQLite em memória, o que permite que os testes sejam rápidos e não afetem a base de dados de produção.

Para configurar o banco de dados de teste, seguimos alguns passos. Primeiro, criamos uma instância do banco de dados em memória:

python
Copy code
from sqlalchemy.pool import StaticPool

from sqlalchemy import create_engine
engine = create_engine(
    "sqlite:///:memory:",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
Base.metadata.create_all(bind=engine)  # Vincula o engine

Em seguida, definimos uma fábrica de sessões de banco de dados dedicada aos testes:

python
Copy code
from sqlalchemy.orm import sessionmaker
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Com isso, podemos criar um fixture de teste que utilizará essa sessão de teste:

python
Copy code
import pytest

@pytest.fixture
def test_db_session():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

Além disso, devemos adaptar o test_client para usar essa nova sessão de banco de dados, substituindo a sessão padrão pela sessão de teste:

python
Copy code
from protoapp.main import app, get_db_session
@pytest.fixture(scope="function")
def test_client(test_db_session):
    client = TestClient(app)
    app.dependency_overrides[get_db_session] = lambda: test_db_session
    return client

python
Copy code
def test_client_can_add_read_the_item_from_database(test_client, test_db_session):
    response = test_client.get("/item/1")
    assert response.status_code == 404

    response = test_client.post("/item", json={"name": "ball", "color": "red"})

    assert response.status_code == 201
    item_id = response.json()
    item = test_db_session.query(Item).filter(Item.id == item_id).first()
    assert item is not None
    response = test_client.get(f"/item/{item_id}")
    assert response.status_code == 200
    assert response.json() == {"name": "ball", "color": "red"}

Além disso, é importante garantir que a cobertura dos testes seja alta, o que pode ser feito utilizando ferramentas como o pytest-cov. Com isso, podemos garantir que todas as partes do código estão sendo testadas, inclusive os endpoints da API. A aplicação de marcadores no pytest também facilita a execução de grupos de testes específicos, como por exemplo, os testes de integração:

python
Copy code
@pytest.mark.integration

def test_client_can_add_read_the_item_from_database(test_client, test_db_session):
    # Conteúdo do teste

Ao utilizar marcadores, podemos agrupar os testes conforme suas funcionalidades, tornando o processo de execução e manutenção dos testes mais eficiente.

É fundamental entender que testes não são apenas uma forma de verificar se a aplicação está funcionando corretamente, mas também de garantir que, ao modificar ou adicionar novos recursos, o comportamento da aplicação continue consistente. Além disso, ao realizar testes unitários e de integração, conseguimos identificar pontos de falha de forma antecipada, permitindo uma resposta rápida e uma aplicação mais robusta.

Como Integrar o FastAPI com o gRPC: Um Guia Completo para Construir um Gateway REST-gRPC

O primeiro passo é configurar o ambiente. Para isso, você deve instalar as bibliotecas necessárias, incluindo o fastapi, uvicorn, grpcio e grpcio-tools. Uma maneira prática de garantir que todas as dependências sejam instaladas é usar um arquivo requirements.txt, ou você pode instalar os pacotes diretamente com o seguinte comando:

ruby
Copy code
$ pip install fastapi uvicorn grpcio grpcio-tools

Com o ambiente pronto, podemos começar a construir o servidor gRPC básico. O servidor gRPC que vamos criar terá um método que recebe uma mensagem do cliente e envia uma resposta de volta. Para isso, precisamos primeiro definir o serviço e as mensagens no arquivo .proto.

1. Criação do Arquivo .proto: O primeiro passo é criar um arquivo chamado grpcserver.proto no diretório raiz do projeto. Nele, definimos o serviço GrpcServer e os tipos de mensagem que nosso serviço utilizará. Exemplo de definição do serviço e das mensagens:

proto
Copy code
syntax = "proto3";
service GrpcServer {
  rpc GetServerResponse(Message) returns (MessageResponse);
}
message Message {
  string message = 1;
}
message MessageResponse {
  string message = 1;
  bool received = 2;
}

A partir desse arquivo .proto, podemos gerar o código Python necessário para implementar o servidor e o cliente gRPC usando o compilador protoc. Execute o seguinte comando:

shell
Copy code
$ python -m grpc_tools.protoc --proto_path=. ./grpcserver.proto --python_out=. --grpc_python_out=.

Esse comando gerará dois arquivos: grpcserver_pb2_grpc.py (que contém a classe para construir o servidor) e grpcserver_pb2.py (que define as mensagens Message e MessageResponse).

2. Implementação do Servidor gRPC: Agora, vamos criar o servidor gRPC. Para isso, criamos o arquivo grpc_server.py e implementamos a classe Service, que vai herdar de GrpcServerServicer e implementar o método GetServerResponse. O código abaixo define o servidor que escuta requisições e envia respostas para o cliente:

python
Copy code
from grpcserver_pb2 import MessageResponse

from grpcserver_pb2_grpc import GrpcServerServicer
import logging
class Service(GrpcServerServicer):
    async def GetServerResponse(self, request, context):
        message = request.message
        logging.info(f"Received message: {message}")
        result = f"Hello I am up and running, received: {message}"
        return MessageResponse(message=result, received=True)

Com o servidor definido, o próximo passo é configurá-lo para rodar na porta 50051. Adicionamos o seguinte código:

python
Copy code
import grpc

from grpcserver_pb2_grpc import add_GrpcServerServicer_to_server
import logging
import asyncio
async def serve():
    server = grpc.aio.server()
    add_GrpcServerServicer_to_server(Service(), server)
    server.add_insecure_port("[::]:50051")
    logging.info("Starting server on port 50051")
    await server.start()
    await server.wait_for_termination()
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(serve())

Com isso, você tem um servidor gRPC simples que pode ser iniciado com o comando:

shell
Copy code
$ python ./grpc_server.py

Se tudo estiver configurado corretamente, você verá a mensagem de log INFO:root:Starting server on port 50051 no terminal.

3. Criação do Gateway FastAPI: Agora, vamos criar o gateway REST utilizando o FastAPI. Esse gateway será responsável por receber as requisições HTTP e redirecioná-las para o servidor gRPC. Primeiramente, criamos o diretório app com o arquivo main.py, onde definimos a aplicação FastAPI.

No arquivo main.py, começamos importando o FastAPI e configurando o servidor:

python
Copy code
from fastapi import FastAPI
import grpc
from grpcserver_pb2_grpc import GrpcServerStub
from grpcserver_pb2 import Message
from pydantic import BaseModel
app = FastAPI()

Em seguida, criamos um modelo de resposta com o Pydantic para refletir a estrutura do MessageResponse do gRPC:

python
Copy code
class GRPCResponse(BaseModel):

    message: str
    received: bool

Agora, configuramos a comunicação com o servidor gRPC. Para isso, utilizamos o canal grpc.aio.insecure_channel:

python
Copy code
grpc_channel = grpc.aio.insecure_channel("localhost:50051")

E finalmente, criamos o endpoint /grpc, que recebe uma mensagem e envia uma requisição ao servidor gRPC, retornando a resposta do servidor para o cliente HTTP:

python
Copy code
@app.get("/grpc")

async def call_grpc(message: str) -> GRPCResponse:
    async with grpc_channel as channel:
        grpc_stub = GrpcServerStub(channel)
        response = await grpc_stub.GetServerResponse(Message(message=message))
        return response

Para rodar a aplicação FastAPI, basta executar o seguinte comando:

ruby
Copy code
$ uvicorn app.main:app

Agora, você pode acessar a documentação interativa da API em http://localhost:8000/docs e testar a chamada ao endpoint /grpc. Quando a requisição for realizada, você verá o log no terminal do servidor gRPC com a mensagem recebida.

É importante destacar que, além de chamadas simples como a do tipo Unary RPC (um único envio e resposta), o gRPC oferece suporte a outras formas de comunicação, como streaming de mensagens de/para o servidor ou comunicação bidirecional. Esses tipos de RPC oferecem soluções poderosas para cenários que exigem maior flexibilidade na comunicação.

Embora o processo de integração do FastAPI com o gRPC seja relativamente simples, compreender as diferenças entre os tipos de RPC e saber quando utilizá-los é crucial para tirar o máximo proveito do gRPC.

Como integrar FastAPI com outras bibliotecas Python para criar endpoints GraphQL e ML

A primeira etapa consiste em instalar as dependências necessárias para que o FastAPI e o GraphQL funcionem corretamente. Para isso, podemos usar o arquivo requirements.txt ou o comando pip install fastapi uvicorn strawberry-graphql[fastapi] diretamente no terminal. Uma vez que o ambiente esteja pronto, podemos começar a construir as funcionalidades.

Em primeiro lugar, criamos um módulo database.py que simula uma base de dados simples com uma lista de usuários. Aqui, definimos a classe User com atributos como id, username, phone_number e country usando o BaseModel da biblioteca Pydantic. Este modelo serve para estruturar os dados dos usuários e facilitar a integração com o GraphQL, que, por sua vez, exigirá uma estrutura de dados definida.

Em seguida, criamos o arquivo graphql_utils.py, onde configuramos o GraphQL. Começamos definindo um modelo User usando a biblioteca Strawberry, que facilita a criação de tipos e consultas GraphQL em Python. O modelo User incluirá os mesmos campos presentes na base de dados, garantindo consistência na consulta dos dados. A seguir, implementamos a query que receberá o parâmetro country e retornará uma lista de usuários de determinado país.

A integração entre FastAPI e GraphQL se dá por meio do GraphQLRouter da biblioteca strawberry.fastapi. Criamos o schema GraphQL e, em seguida, adicionamos o roteador ao objeto FastAPI no arquivo main.py. Através desse processo, podemos facilmente criar um endpoint em FastAPI que manipula solicitações GraphQL na rota /graphql. O endpoint se torna interativo ao ser acessado no navegador, permitindo ao desenvolvedor realizar consultas diretamente pela interface gráfica, simplificando o processo de interação com os dados.

json
Copy code
{
  "data": {
    "users": [
      {
        "username": "user1",
        "country": "USA",
        "phoneNumber": "1234567890"
      }
    ]
  }
}

Esse modelo de integração entre REST e GraphQL abre vastas possibilidades para ampliar a flexibilidade nas consultas e modificações de dados, especialmente em aplicações que envolvem grandes volumes de informações.

Agora, exploraremos como a mesma estrutura do FastAPI pode ser aplicada para integrar modelos de Machine Learning. O Joblib, uma biblioteca popular para serialização de modelos, pode ser usado para carregar e salvar modelos treinados, como o modelo de previsão de doenças desenvolvido com scikit-learn. FastAPI facilita o uso de modelos ML em produção, permitindo criar APIs que podem ser acessadas remotamente para realizar previsões.

Ao desenvolver uma aplicação como um "médico inteligente" que realiza diagnósticos baseados em sintomas, é possível combinar o uso de ML com uma API construída em FastAPI. Para isso, começamos configurando o ambiente com o Joblib, além de instalar as dependências necessárias como scikit-learn e huggingface_hub, que nos permite baixar modelos treinados diretamente do Hugging Face.

No código de exemplo, o modelo é carregado na memória ao iniciar a aplicação FastAPI, e um dicionário ml_model mantém o modelo carregado disponível para todas as requisições. A API recebe os sintomas do paciente, realiza a previsão usando o modelo carregado e retorna o diagnóstico, simplificando o processo de análise. O endpoint da API pode ser configurado para receber os sintomas do paciente e retornar o diagnóstico predito pelo modelo ML.

Ao integrar essas ferramentas, o desenvolvimento de uma aplicação de ML robusta e eficiente torna-se mais acessível. A combinação de FastAPI com GraphQL e ML possibilita criar soluções ágeis, escaláveis e facilmente testáveis, aumentando a produtividade e a inovação no desenvolvimento de software.

O que deve ser considerado além do que foi descrito: