! Como desenvolvedores de back-end, sabemos que todo milissegundo conta. Mas como podemos fazer as APIs responderem mais rapidamente? A resposta está no cache.

O cache é uma técnica que armazena dados frequentemente acessados ​​na memória, permitindo que as APIs respondam instantaneamente em vez de consultar um banco de dados mais lento todas as vezes. Pense nisso como manter os principais ingredientes (sal, pimenta, óleo) na bancada da cozinha, em vez de buscá -los da despensa cada vez que você cozinha - isso economiza tempo e torna o processo mais eficiente. Da mesma forma, o cache reduz os tempos de resposta da API armazenando dados comumente solicitados em um local rápido e acessível, como Redis.

Bibliotecas necessárias para serem instaladas

para se conectar com o Redis Cache com FASTAPI, exigimos que as seguintes bibliotecas sejam pré-instaladas.

install instale fastapi uvicorn aiocache pydantic

pip install fastapi uvicorn aiocache pydantic

pydantic é para criar tabelas e estruturas de banco de dados. O Aiocache executará operações assíncronas no cache. O Uvicorn é responsável pelo servidor em execução.

Redis Setup and Verification:

A configuração do Redis diretamente em um sistema Windows não é possível neste momento. Portanto, ele deve ser configurado e executado no subsistema Windows para Linux. As instruções para a instalação do WSL são dadas abaixo

POST Instalando o WSL, os seguintes comandos são obrigados a instalar Redis

sudo apt atualização sudo apt install redis-server sudo systemctl start redis

sudo apt update
sudo apt install redis-server
sudo systemctl start redis

Para testar a conectividade do servidor Redis, o seguinte comando é usado

redis-cli

sudo apt update
sudo apt install redis-server
sudo systemctl start redis

Após este comando, ele entrará em um terminal virtual da porta 6379. Nesse terminal, os comandos Redis podem ser digitados e testados.

Configurando o aplicativo FASTAPI

! Usaremos Redis para armazenar respostas em cache.

Etapa 1: Defina o modelo pydantic para dados do usuário

usaremos o Pydantic para definir nosso modelo de usuário, que representa a estrutura da resposta da API.

de Pydantic Import Basemodel Usuário da classe (Basemodel): id: int Nome: Str E -mail: str Idade: Int

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    age: int

Para evitar repetir a lógica de cache para cada ponto de extremidade, criaremos um decorador de cache reutilizável usando a biblioteca Aiocache. Este decorador tentará recuperar a resposta de Redis antes de chamar a função real.

import json de functools importando envoltórios Do cache de importação de Aiocache De fastapi importar httpexception DEF CACHE_RESPONSEIRO (TTL: int = 60, namespace: str = "main"): "" " Decorador de armazenamento em cache para pontos de extremidade do FASTAPI. TTL: Hora de viver para o cache em segundos. Namespace: espaço para nome para chaves de cache em redis. "" " Def Decorator (func): @wraps (func) Wrapper assíncrono (*args, ** kwargs): user_id = kwargs.get ('user_id') ou args [0] # assumindo que o ID do usuário é o primeiro argumento cache_key = f "{namespace}: user: {user_id}" cache = cache.redis (endpoint = "localhost", porta = 6379, namespace = namespace) # Tente recuperar dados do cache cached_value = wait cache.get (cache_key) se cached_value: Retornar JSON.Loads (CACHED_VALUE) # Retornar dados em cache # Ligue para a função real se o cache não for atingido Resposta = Wait Func (*args, ** Kwargs) tentar: # Armazene a resposta em Redis com um TTL aguarda cache.set (cache_key, json.dumps (resposta), ttl = ttl) Exceto exceção como e: Aumente httPexception (status_code = 500, detalhe = f "Dados de cache de erro: {e}") resposta de retorno Retornar invólucro Decorador de retorno

import json
from functools import wraps
from aiocache import Cache
from fastapi import HTTPException

def cache_response(ttl: int = 60, namespace: str = "main"):
    """
    Caching decorator for FastAPI endpoints.

    ttl: Time to live for the cache in seconds.
    namespace: Namespace for cache keys in Redis.
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            user_id = kwargs.get('user_id') or args[0]  # Assuming the user ID is the first argument
            cache_key = f"{namespace}:user:{user_id}"

            cache = Cache.REDIS(endpoint="localhost", port=6379, namespace=namespace)

            # Try to retrieve data from cache
            cached_value = await cache.get(cache_key)
            if cached_value:
                return json.loads(cached_value)  # Return cached data

            # Call the actual function if cache is not hit
            response = await func(*args, **kwargs)

            try:
                # Store the response in Redis with a TTL
                await cache.set(cache_key, json.dumps(response), ttl=ttl)
            except Exception as e:
                raise HTTPException(status_code=500, detail=f"Error caching data: {e}")

            return response
        return wrapper
    return decorator

agora implementaremos uma rota FASTAPI que recupera as informações do usuário com base em um ID de usuário. A resposta será armazenada em cache usando o Redis para um acesso mais rápido nas solicitações subsequentes.

de FASTAPI import FASTAPI App = FASTAPI () # Exemplo de dados representando usuários em um banco de dados usuários_db = { 1: {"ID": 1, "Nome": "Alice", "Email": "[email protected]", "Age": 25}, 2: {"id": 2, "nome": "bob", "email": "[email protected]", "idade": 30}, 3: {"ID": 3, "Nome": "Charlie", "Email": "[email protected]", "Age": 22}, } @app.get ("/users/{user_id}") @Cache_Response (ttl = 120, namespace = "usuários") ASYNC def get_user_details (user_id: int): # Simule uma chamada de banco de dados recuperando dados de usuários_db Usuário = usuários_db.get (user_id) Se não o usuário: Aumente httPexception (status_code = 404, detalhe = "Usuário não encontrado") Retornar usuário

from fastapi import FastAPI

app = FastAPI()

# Sample data representing users in a database
users_db = {
    1: {"id": 1, "name": "Alice", "email": "[email protected]", "age": 25},
    2: {"id": 2, "name": "Bob", "email": "[email protected]", "age": 30},
    3: {"id": 3, "name": "Charlie", "email": "[email protected]", "age": 22},
}

@app.get("/users/{user_id}")
@cache_response(ttl=120, namespace="users")
async def get_user_details(user_id: int):
    # Simulate a database call by retrieving data from users_db
    user = users_db.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

Inicie o seu aplicativo FASTAPI executando:

uvicorn Main: App - -Reload

uvicorn main:app --reload

http://127.0.0.1:8000/users/1

;

http://127.0.0.1:8000/users/1

você pode verificar o cache inspecionando as teclas armazenadas em redis. Abra a CLI Redis:

redis-cli Chaves *

você receberá todas as chaves que foram armazenadas no Redis até ttl.

redis-cli
KEYS *

Primeiro pedido

: Quando os dados do usuário são solicitados pela primeira vez, a API o busca no banco de dados (usuários_db) e armazena o resultado em redis com um tempo de vida (TTL) de 120 segundos.

Solicitações subsequentes:

;

TTL (hora de viver):

Após 120 segundos, a entrada do cache expira e os dados são buscados no banco de dados novamente na próxima solicitação, atualizando o cache.

Conclusão

Neste tutorial, demonstramos como implementar o cache do Redis em um aplicativo FASTAPI usando um exemplo simples de detalhes do usuário. Ao armazenar em cache as respostas da API, você pode melhorar significativamente o desempenho do seu aplicativo, principalmente para dados que não mudam com frequência.

Por favor, faça o UpVote e compartilhe se achar útil este artigo.