Como Integrar Elasticsearch e FastAPI: Construção de Consultas e Melhoria de Performance com Cache

Ao trabalhar com FastAPI e Elasticsearch, o processo de integrar essas duas tecnologias exige não apenas a configuração do cliente Elasticsearch, mas também a criação de consultas eficientes que atendam às necessidades específicas da aplicação. A seguir, exploraremos como criar consultas no Elasticsearch, integrar o FastAPI com o Elasticsearch, e otimizar a performance utilizando Redis como cache.

python
Copy code
async def main():
    await create_index()
    await fill_elastichsearch()  # caso esteja usando Elasticsearch
if __name__ == "__main__":
    import asyncio
    asyncio.run(create_index())

Isso configurará o índice no Elasticsearch e permitirá que você adicione os dados (como músicas) no índice. Após a criação do índice, podemos focar na construção da consulta, que será responsável por buscar e ordenar os dados conforme nossa necessidade.

Construindo a Consulta

Para construir uma consulta que retorne as músicas mais populares em um determinado país, você pode definir a consulta como uma função Python que retorna um dicionário contendo a consulta. Esse dicionário será utilizado pelo cliente Elasticsearch para fazer a busca.

O exemplo de consulta a seguir é construído para buscar as dez músicas mais visualizadas de um país específico:

python
Copy code
def top_ten_songs_query(country) -> dict:

    views_field = f"views_per_country.{country}"
    query = {
        "bool": {
            "must": {"match_all": {}},
            "filter": [
                {"exists": {"field": views_field}}
            ],
        }
    }
    sort = {views_field: {"order": "desc"}}
    source = [
        "title", views_field, "album.title", "artist"
    ]
    return {
        "index": "songs_index",
        "query": query,
        "size": 10,
        "sort": sort,
        "source": source,
    }

Nessa consulta, estamos buscando todas as músicas que possuem dados de visualizações por país, filtrando apenas os campos relevantes para a resposta e ordenando os resultados de forma decrescente com base nas visualizações.

Integrando o FastAPI com Elasticsearch

Com o cliente Elasticsearch configurado e a consulta pronta, o próximo passo é integrar isso ao FastAPI. Abaixo segue a criação do endpoint que irá utilizar a consulta criada:

python
Copy code
from fastapi import APIRouter, Depends, HTTPException
from app.db_connection import es_client
router = APIRouter(prefix="/search", tags=["search"])
def get_elasticsearch_client():
    return es_client
@router.get("/top/ten/artists/{country}")
async def top_ten_artist_by_country(country: str, es_client=Depends(get_elasticsearch_client)):
    try:
        response = await es_client.search(*top_ten_songs_query(country))
    except BadRequestError as e:
        logger.error(e)
        raise HTTPException(status_code=400, detail="Invalid country")
    return [
        {
            "artist": record.get("key"),
            "views": record.get("views", {}).get("value")
        }

        for record in response["aggregations"]["top_ten_artists"]["buckets"]

    ]

Este endpoint consulta o Elasticsearch para retornar os artistas mais populares com base nas visualizações de músicas em um país específico. A função top_ten_songs_query é chamada para construir a consulta, e os resultados são retornados após um tratamento para extrair os campos necessários.

Melhorando o Desempenho com Redis

Para integrar o Redis ao FastAPI, o primeiro passo é instalar e configurar o cliente Redis. Após a instalação do Redis e do cliente Redis para Python, a configuração no código pode ser feita da seguinte maneira:

python
Copy code
from redis import asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost")
async def ping_redis_server():
    try:
        await redis_client.ping()
        logger.info("Conectado ao Redis")
    except Exception as e:
        logger.error(f"Erro ao conectar ao Redis: {e}")
        raise e

python
Copy code
@router.get("/top/ten/artists/{country}")
async def top_ten_artist_by_country(
    country: str, 
    es_client=Depends(get_elasticsearch_client), 
    redis_client=Depends(get_redis_client)
):
    # Verifica se os dados estão no cache
    cached_data = await redis_client.get(f"top_ten_artists_{country}")
    if cached_data:
        return cached_data
    try:
        # Se não estiver no cache, consulta o Elasticsearch
        response = await es_client.search(*top_ten_songs_query(country))
    except BadRequestError as e:
        logger.error(e)
        raise HTTPException(status_code=400, detail="Invalid country")
    data = [

        {"artist": record.get("key"), "views": record.get("views", {}).get("value")}

        for record in response["aggregations"]["top_ten_artists"]["buckets"]
    ]
    # Armazena no cache

    await redis_client.set(f"top_ten_artists_{country}", data, ex=3600)  # Expires in 1 hour

    return data

Neste exemplo, o Redis armazena os resultados por uma hora, evitando consultas repetidas ao Elasticsearch e melhorando a performance do sistema.

Considerações Importantes

Ao trabalhar com Elasticsearch e FastAPI, é essencial entender a natureza assíncrona dessas ferramentas. O uso de asyncio e await é fundamental para garantir que o sistema possa lidar com múltiplas requisições simultâneas sem bloquear a execução.

Além disso, ao implementar cache com Redis, deve-se estar atento ao tempo de expiração dos dados armazenados. Definir um tempo de expiração adequado para os dados no cache ajuda a balancear a performance com a precisão das informações, já que dados obsoletos podem ser mantidos por tempo demais.

Como Criar e Utilizar Dependências e Middleware Personalizado no FastAPI

O FastAPI oferece uma série de recursos poderosos para facilitar o desenvolvimento de APIs robustas e eficientes. Entre esses recursos, destacam-se a injeção de dependências e a capacidade de criar middleware personalizado. Ambas as funcionalidades permitem uma organização mais eficiente do código, além de agregar flexibilidade e modularidade ao processo de desenvolvimento.

Injeção de Dependências com Classes

Para ilustrar isso, podemos imaginar um cenário em que precisamos de parâmetros como o intervalo de tempo, a categoria e um código em um ponto de extremidade da API. Em vez de passar cada um desses parâmetros separadamente para a função de tratamento da rota, podemos agrupá-los em uma classe e usá-los como uma dependência.

A implementação começa com a criação de uma classe que irá encapsular os parâmetros desejados. Por exemplo:

python
Copy code
from pydantic import BaseModel

class FilterParameters(BaseModel):
    time_range: str
    category: str
    code: str

Após definir essa classe, podemos usá-la como uma dependência em um ponto de extremidade. A declaração ficaria assim:

python
Copy code
from fastapi import Depends, FastAPI
app = FastAPI()
@app.get("/items")

def get_items(filters: FilterParameters = Depends(FilterParameters)):

    return {"filters": filters}

Aqui, o FastAPI automaticamente valida os parâmetros de entrada de acordo com a classe FilterParameters, que garante que todos os campos sejam fornecidos corretamente. Esse método aumenta a legibilidade e a modularidade do código, além de evitar a repetição dos parâmetros em cada função.

Middleware Personalizado

Para criar um middleware personalizado, precisamos primeiro criar uma classe que herda de BaseHTTPMiddleware, da biblioteca Starlette. A seguir, é preciso sobrescrever o método dispatch, que será responsável por interceptar as requisições e realizar as operações desejadas.

Exemplo de middleware que loga informações sobre a requisição:

python
Copy code
import logging

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
logger = logging.getLogger("uvicorn.error")
class ClientInfoMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        host_client = request.client.host
        requested_path = request.url.path
        method = request.method

        logger.info(f"host client {host_client} requested {method} {requested_path} endpoint")

        return await call_next(request)

Com o middleware definido, basta adicioná-lo ao seu aplicativo FastAPI da seguinte maneira:

python
Copy code
from fastapi import FastAPI
from app.middleware import ClientInfoMiddleware
app = FastAPI()
app.add_middleware(ClientInfoMiddleware)

Esse middleware, por exemplo, intercepta todas as requisições e registra o cliente que fez a solicitação, o caminho solicitado e o método HTTP. Em um cenário real, essas informações poderiam ser armazenadas em um banco de dados ou usadas para análises de segurança.

Importância do Uso de Middleware Personalizado

Middleware personalizado não se limita apenas à simples interceptação de requisições. Ele pode ser expandido para incluir funcionalidades avançadas, como redirecionamento de requisições com base no IP do cliente, filtragem de conteúdos, verificação de cabeçalhos de segurança e muito mais. O poder do middleware está em sua flexibilidade e capacidade de centralizar o processamento de dados e decisões antes que uma requisição atinja a lógica principal da aplicação.

Embora o FastAPI ofereça suporte para criar middleware de maneira simples, é altamente recomendado encapsular essa funcionalidade em classes. Isso não só melhora a legibilidade e a modularidade do código, mas também facilita a reutilização e manutenção.

Internacionalização e Localização

Outro aspecto importante ao desenvolver APIs robustas é garantir que sua aplicação seja capaz de se adaptar a diferentes regiões e culturas. Isso envolve a implementação de internacionalização (i18n) e localização (l10n). O i18n refere-se ao processo de preparar um software para ser adaptado a diferentes idiomas e culturas, enquanto o l10n envolve a adaptação real do conteúdo para um mercado específico, como formatos de moeda ou unidades de medida.

No FastAPI, uma forma comum de implementar i18n e l10n é utilizando o cabeçalho HTTP Accept-Language, que informa ao servidor as preferências de idioma do cliente. A partir dessa informação, é possível adaptar as respostas da API para o idioma desejado.

Um exemplo de como isso pode ser feito é o uso da biblioteca babel, que facilita a resolução de códigos de idiomas. Primeiro, definimos os idiomas suportados pela aplicação:

python
Copy code
SUPPORTED_LOCALES = ['en_US', 'fr_FR']

Em seguida, podemos implementar uma função que, com base no cabeçalho Accept-Language, seleciona o idioma mais adequado para o cliente:

python
Copy code
from fastapi import Request
def get_locale(request: Request):
    accept_language = request.headers.get('Accept-Language', 'en')
    return accept_language

Com isso, é possível personalizar a resposta da API com base no idioma preferido do cliente. Isso se aplica tanto ao conteúdo textual quanto a outros aspectos, como unidades de medida ou formatação de data/hora.

Considerações Finais

Ao utilizar dependências e middleware no FastAPI, você não só organiza seu código de forma mais eficiente, mas também ganha flexibilidade e escalabilidade, permitindo a adição de novas funcionalidades e a melhoria contínua da aplicação. As práticas descritas aqui são apenas o começo: a injeção de dependências e o middleware personalizado podem ser expandidos para cobrir uma infinidade de casos de uso, tornando sua API mais modular e manutenível.

A internacionalização e localização, por sua vez, são conceitos fundamentais para a criação de aplicações que atendem usuários em diversas partes do mundo, garantindo uma experiência consistente e culturalmente relevante para cada um deles.

Como Desenvolver Middleware para Modificação de Respostas em FastAPI

O uso de middleware no desenvolvimento de APIs com FastAPI é uma prática fundamental para manipular tanto as requisições quanto as respostas. A modificação de respostas, por exemplo, pode ser necessária em uma série de casos, como adicionar cabeçalhos personalizados, logar detalhes da resposta, entre outros. A seguir, vamos explorar como criar um middleware que permite a modificação das respostas de uma aplicação FastAPI.

Para implementar um middleware que altere os cabeçalhos da resposta, começamos criando uma classe chamada ExtraHeadersResponseMiddleware. O objetivo desse middleware será adicionar cabeçalhos personalizados a cada resposta gerada pela aplicação. Vamos ver como isso pode ser feito:

Definindo o Middleware

python
Copy code
class ExtraHeadersResponseMiddleware:
    def __init__(self, app: ASGIApp, headers: Sequence[tuple[str, str]]):
        self.app = app
        self.headers = headers

python
Copy code
async def __call__(self, scope: Scope, receive: Receive, send: Send):
    if scope["type"] != "http":
        return await self.app(scope, receive, send)

Modificando a Resposta

O principal trabalho do middleware é modificar a resposta antes que ela seja enviada de volta para o cliente. Para isso, criamos uma função auxiliar chamada send_with_extra_headers, que é responsável por adicionar os cabeçalhos extras à resposta.

python
Copy code
async def send_with_extra_headers(message: Message):
    if message["type"] == "http.response.start":
        headers = MutableHeaders(scope=message)
        for key, value in self.headers:
            headers.append(key, value)
        await send(message)

Após a modificação dos cabeçalhos, o middleware chama o send com a mensagem já modificada, garantindo que os cabeçalhos extras sejam incluídos na resposta final.

Adicionando o Middleware à Aplicação FastAPI

Uma vez que o middleware foi definido, o próximo passo é adicioná-lo à instância da aplicação FastAPI. Isso é feito utilizando o método add_middleware, onde passamos a classe ExtraHeadersResponseMiddleware e os cabeçalhos que queremos adicionar.

python
Copy code
app.add_middleware(

    ExtraHeadersResponseMiddleware,
    headers=(
        ("new-header", "fastapi-cookbook"),
        ("another-header", "fastapi-cookbook"),
    ),
)

Ao executar o servidor, por exemplo, com uvicorn main:app, podemos testar a modificação de cabeçalhos. Ao chamar um endpoint da aplicação, podemos verificar que os dois cabeçalhos personalizados foram adicionados à resposta:

text
Copy code
another-header: fastapi-cookbook
content-length: 28
content-type: application/json
date: Thu,23 May 2024 09:24:41 GMT
new-header: fastapi-cookbook
server: uvicorn

Esse processo é simples, mas fundamental para quem deseja controlar as respostas de sua API de maneira eficaz e com flexibilidade.

Considerações Finais

É importante lembrar que o uso de middleware para adicionar ou modificar cabeçalhos pode ser útil em situações em que, por exemplo, você precisa garantir que todas as respostas contenham cabeçalhos de segurança, como Content-Security-Policy ou X-Frame-Options, ou até cabeçalhos que indicam a versão da API.

Em um ambiente de produção, é sempre recomendado que você defina com clareza quais cabeçalhos serão manipulados e com quais valores, pois a modificação inadequada de cabeçalhos pode afetar a segurança e a performance da aplicação.

Como Gerenciar Erros e Exceções de Forma Eficiente em Aplicações FastAPI

Gerenciar erros e exceções não se limita a capturar problemas inesperados, mas também envolve projetar proativamente a sua aplicação para responder de maneira adequada a diversos cenários de erro. A capacidade de lidar com erros de forma personalizada e eficaz é essencial para garantir que suas aplicações FastAPI não apenas sejam funcionais, mas também amigáveis para o usuário e seguras. Neste contexto, abordaremos como personalizar o tratamento de erros, validar dados, lidar com exceções e testar esses cenários para garantir a resiliência de suas aplicações.

O FastAPI fornece suporte integrado para o manuseio de exceções e erros. Quando um erro ocorre, o FastAPI retorna uma resposta em JSON contendo detalhes sobre o erro, o que é extremamente útil para depuração. No entanto, existem situações em que você pode querer personalizar essas respostas de erro para melhorar a experiência do usuário ou reforçar a segurança. A criação de manipuladores de erros personalizados é uma forma eficaz de ajustar as mensagens de erro a um padrão que seja mais adequado ao seu caso específico.

python
Copy code
from fastapi import FastAPI, HTTPException
from starlette.responses import JSONResponse
@app.exception_handler(HTTPException)

async def http_exception_handler(request, exc):

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "message": "Oops! Algo deu errado"
        },
    )

Neste exemplo, a função http_exception_handler é utilizada para capturar erros do tipo HTTPException. Sempre que um erro desse tipo for gerado em sua aplicação, o FastAPI usará esse manipulador para retornar uma resposta personalizada. Para testar esse comportamento, você pode criar um novo ponto de extremidade que levante uma exceção HTTP:

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

async def raise_exception():
    raise HTTPException(status_code=400)

Ao acessar o ponto de extremidade http://localhost:8000/error_endpoint, a resposta será a seguinte:

json
Copy code
{
    "message": "Oops! Algo deu errado"
}

Além disso, o FastAPI usa o Pydantic para validação de dados. Quando um dado enviado em uma requisição não está em conformidade com o modelo definido, o FastAPI automaticamente levanta uma exceção e retorna uma resposta de erro. Por exemplo, ao tentar chamar o ponto de extremidade POST /book com um valor numérico no campo title, onde se espera uma string, o FastAPI irá gerar um erro de validação.

Se você desejar personalizar essa resposta de erro, o FastAPI permite a criação de um manipulador específico para erros de validação, como mostrado abaixo:

python
Copy code
import json

from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request, exc: RequestValidationError
):
    return PlainTextResponse(
        "Este é um erro de validação:",
        status_code=status.HTTP_400_BAD_REQUEST,
    )

Neste exemplo, o manipulador de erros para RequestValidationError retornará uma resposta de texto simples, exibindo detalhes do erro de validação. Isso permite que você forneça mensagens de erro mais detalhadas e compreensíveis para os usuários ou desenvolvedores que estão interagindo com sua API.

Essas personalizações de resposta garantem que sua aplicação lide de maneira mais eficiente e amigável com os erros, proporcionando uma melhor experiência tanto para os desenvolvedores quanto para os usuários. Além disso, você pode aprimorar o controle de erros criando tratamentos específicos para diferentes tipos de exceções, garantindo que sua aplicação seja robusta e resiliente em qualquer situação.

É importante que, ao trabalhar com FastAPI, você também se atente a aspectos como a validação eficiente de dados, a proteção contra entradas maliciosas, e o impacto das exceções na performance geral da aplicação. A personalização de erros, embora crucial, deve ser sempre feita com um equilíbrio entre usabilidade e segurança, para evitar que informações sensíveis sejam expostas ou que o comportamento da API se torne confuso para o usuário final.