Pular para o conteúdo

    Capítulo 28, Avançado

    Cache

    Calcular ou buscar de novo o que não mudou é desperdício. O cache guarda uma resposta por um tempo, e a pergunta difícil é a outra: quando ela deixa de valer.

    Um cache com validade (TTL)

    A ideia: guardar o resultado junto do momento em que foi calculado, e reaproveitá-lo enquanto não passar o TTL (time to live, o tempo de validade). Depois disso, recalcular. Eu escrevo o relógio como um parâmetro, para o teste controlar o tempo e não precisar esperar de verdade:

    avancado/cap28_cache.pylinhas 10 a 61
    import threading
    import time
    from typing import Callable
    
    
    class CacheTTL:
        def __init__(self, ttl: float, relogio: Callable[[], float] = time.monotonic):
            self.ttl = ttl
            self.relogio = relogio
            self._dados: dict[str, tuple[float, object]] = {}
            self._trava = threading.Lock()
    
        def obter_ou_calcular(self, chave: str, calcular: Callable[[], object]) -> object:
            with self._trava:
                agora = self.relogio()
                guardado = self._dados.get(chave)
                if guardado is not None and agora - guardado[0] < self.ttl:
                    return guardado[1]
                valor = calcular()
                self._dados[chave] = (agora, valor)
                return valor
    
        def invalidar(self, chave: str) -> None:
            with self._trava:
                self._dados.pop(chave, None)
    
    
    class RelogioFalso:
        def __init__(self):
            self.agora = 0.0
    
        def __call__(self) -> float:
            return self.agora
    
    
    relogio = RelogioFalso()
    cache = CacheTTL(ttl=60, relogio=relogio)
    calculos = []
    
    
    def cotacao_cara() -> float:
        calculos.append(1)
        return 5.43
    
    
    print([cache.obter_ou_calcular("usd", cotacao_cara) for _ in range(3)], len(calculos))
    relogio.agora = 59.9
    cache.obter_ou_calcular("usd", cotacao_cara)
    print(len(calculos))
    relogio.agora = 60.1
    cache.obter_ou_calcular("usd", cotacao_cara)
    print(len(calculos))
    
    Saída
    [5.43, 5.43, 5.43] 1
    1
    2
    

    Três chamadas, um cálculo só. Aos 59,9 segundos, o cache ainda vale. Aos 60,1, expirou e recalculou. Três detalhes do código importam: o relógio é o time.monotonic (que não volta nem pula quando o relógio do sistema é ajustado), há uma trava (o servidor atende requisições em várias threads, e uma variável global sem trava corrompe o cache), e existe um invalidar, porque às vezes você sabe que o dado mudou e não vai esperar o TTL.

    Em uma rota

    avancado/cap28_cache.pylinhas 66 a 92
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    
    app = FastAPI()
    chamadas_externas = []
    
    
    def buscar_cotacao_externa() -> float:
        chamadas_externas.append(1)
        return 5.43
    
    
    @app.get("/cotacao")
    def cotacao():
        valor = cache.obter_ou_calcular("usd", buscar_cotacao_externa)
        return {"usd": valor}
    
    
    relogio.agora = 1000.0
    cache.invalidar("usd")
    cliente = TestClient(app)
    for _ in range(3):
        cliente.get("/cotacao")
    print("chamadas à API externa:", len(chamadas_externas))
    relogio.agora = 1061.0
    cliente.get("/cotacao")
    print("depois que o TTL passou:", len(chamadas_externas))
    
    Saída
    chamadas à API externa: 1
    depois que o TTL passou: 2
    

    `lru_cache` para funções puras

    Para uma função cujo resultado depende só dos argumentos (sem I/O, sem relógio), o functools.lru_cache do Python resolve:

    avancado/cap28_cache.pylinhas 97 a 107
    from functools import lru_cache
    
    
    @lru_cache(maxsize=128)
    def fatorial(n: int) -> int:
        return 1 if n < 2 else n * fatorial(n - 1)
    
    
    fatorial(50)
    fatorial(50)
    print(fatorial.cache_info())
    
    Saída
    CacheInfo(hits=1, misses=50, maxsize=128, currsize=50)
    

    O cache_info mostra os acertos (hits) e as falhas (misses). Mas, cuidado com a armadilha: o lru_cache não tem validade, e não deve ser usado em função que devolve dados que mudam (uma consulta ao banco) nem em função que depende de quem está logado: o resultado do primeiro usuário seria servido a todos os outros.

    O cache do HTTP: deixar o cliente guardar

    Outra forma de cache é pedir ao cliente (navegador, CDN) que guarde a resposta. O cabeçalho Cache-Control diz por quanto tempo. E o ETag permite uma revalidação barata: o servidor devolve uma "impressão digital" do conteúdo, e na próxima vez o cliente pergunta "mudou desde esta impressão?". Se não mudou, o servidor responde 304 Not Modified sem corpo:

    avancado/cap28_cache.pylinhas 112 a 135
    import hashlib
    import json
    
    from fastapi import Request, Response
    
    CATALOGO = {"itens": ["caneta", "caderno", "régua"]}
    
    
    @app.get("/catalogo")
    def catalogo(request: Request, response: Response):
        corpo = json.dumps(CATALOGO, sort_keys=True).encode()
        etag = '"' + hashlib.sha256(corpo).hexdigest()[:16] + '"'
        if request.headers.get("if-none-match") == etag:
            return Response(status_code=304, headers={"ETag": etag})
        response.headers["ETag"] = etag
        response.headers["Cache-Control"] = "public, max-age=60"
        return CATALOGO
    
    
    primeira = cliente.get("/catalogo")
    etag = primeira.headers["etag"]
    print(primeira.status_code, primeira.headers["cache-control"], len(primeira.content) > 0)
    revalidada = cliente.get("/catalogo", headers={"If-None-Match": etag})
    print(revalidada.status_code, revalidada.content)
    
    Saída
    200 public, max-age=60 True
    304 b''
    

    Na segunda chamada, o cliente mandou a impressão digital que tinha, o servidor viu que nada mudou e respondeu 304 com o corpo vazio: economizou a transferência, embora ainda tenha feito a viagem de rede.

    Os problemas do cache

    ProblemaO que aconteceComo lidar
    Dado velhoO cache serve o que já mudouTTL curto, ou invalidar quando o dado muda
    Vários processosCada --workers tem o seu cache em memória, e eles divergemCache compartilhado (Redis ou equivalente)
    Dado por usuárioO cache entrega os dados de um usuário a outroInclua o usuário na chave do cache
    Estouro de memóriaO cache cresce sem limiteLimite de tamanho (maxsize) e remoção dos mais antigos
    AvalancheO cache expira e centenas de requisições recalculam ao mesmo tempoA trava do exemplo acima (cada chave calcula uma vez)

    "Só há duas coisas difíceis em computação"

    A frase famosa é que são a invalidação de cache e dar nomes às coisas. Ela é verdadeira: guardar é fácil, e saber quando deixou de valer é o trabalho de verdade. Eu começo sem cache, meço onde está o gargalo (capítulo 48 do curso de Python, "Desempenho e profiling"), e só então escolho o que guardar, com o TTL mais curto que o problema aceita.

    Exercício 1

    Invalidar quando o dado muda

    Use o CacheTTL para guardar perfil:ana. Depois de uma "atualização", chame invalidar("perfil:ana") e confira que o próximo acesso recalcula. Use um relógio falso próprio para o teste.