Uso

Cache¶

Um cache em memória está disponível para funções síncronas e assíncronas. Porém, é fundamental destacar que o cache será zerado ou limpo sempre que o programa for interrompido.

from cachetoolz import cache

# O import acima é equivalente a isso
from cachetoolz import AsyncInMemory, Cache
cache = Cache(AsyncInMemory())

@cache()
def sub(x, y):
    return x - y

@cache()
async def mul(x, y):
    return x * y

Armazena em cache uma chamada de função e a armazena no namespace.

O decorador simples, @cache, é suportado, bem como uma chamada com argumentos de palavra-chave @cache(ttl=7200).

Parâmetro	Tipo	Descrição	Padrão
`backend`	Union[AsyncBackendABC, BackendABC]	Backend de cache	obrigatório

Com backend assíncrono redis

from cachetoolz import AsyncRedisBackend, Cache
cache = Cache(AsyncRedisBackend())

Com backend sincrono redis

from cachetoolz import RedisBackend, Cache
cache = Cache(RedisBackend())

Para obter mais detalhes sobre backends, consulte a seção backends.

@cache¶

Parâmetro	Tipo	Descrição	Padrão
`ttl`	`int`, `float`, `timedelta`	cache ttl (tempo de vida)	`math.inf`
`namespace`	`str`	namespace para cache	`"default"`
`typed`	`bool`	Se typed for definido como True, argumentos de função de tipos diferentes serão armazenados em cache separadamente	`False`
`keygen`	`cachetoolz.types.KeyGenerator`	função para gerar uma chave identificadora de cache	`cachetoolz.utils.default_keygen`

Exemplos:

Um cache simples

>>> @cache
def func(*args, **kwargs):
    ...

Especifique um namespace

@cache(namespace='bar')
def func(*args, **kwargs):
    ...

Defina um tempo de expiração em segundos

@cache(ttl=60)
def func(*args, **kwargs):
    ...

Use timedelta para definir a expiração

from datetime import timedelta
@cache(ttl=timedelta(days=1))
def func(*args, **kwargs):
    ...

Diferencie o cache com base nos tipos de argumento

@cache(typed=True)
def func(*args, **kwargs):
    ...

Usando um keygen personalizado

def custom_keygen(
    typed: bool, func: Func, *args: P.args, **kwargs: P.kwargs
) -> str:
    """
    Construa uma chave para uma função.

    Parameters
    ----------
    typed
        Se typed for definido como True, argumentos de função
        de tipos diferentes serão armazenados em cache separadamente.
    func
        Função
    args
        Argumentos posicionais de função.
    kwargs
        Argumentos nomeados de função.

    Returns
    -------
        Chave identificadora de cache.

     """

@cache(keygen=custom_keygen)
def func(*args, **kwargs):
    ...

Altere o backend de cache que será usado em uma função específica

@cache(backend=RedisBackend())
def func(): ...

@cache.clear¶

Limpa todos os caches de todos os namespaces.

Este decorador limpará todos os caches contidos nos namespaces especificados assim que a função decorada for executada.

Parâmetro	Tipo	Descrição	Padrão
`namespaces`	`Sequence[str]`	namespaces a ser limpos.	`('default',)`

Exemplos:

Um cache simples e limpo

@cache.clear
def func(*args, **kwargs):
    ...

Definindo os namespaces a serem limpos

@cache.clear(namespaces=['foo'])
def func(*args, **kwargs):
    ...

Altere o backend clear que será usado em uma função específica

@cache.clear(backend=RedisBackend())
def func(): ...

Backend¶

Em Memória¶

Estão disponíveis backends na memória síncronos e assíncronos. Apenas instancie-os, eles não precisam de parâmetros.

from cachetoolz import AsyncInMemory, InMemory

async_in_memory = AsyncInMemory()
sync_in_memory = InMemory()

Funções assíncronas podem ser decoradas quando o backend síncrono está sendo usado, mas é importante ter cuidado, pois pode haver possíveis erros ou inconsistências ao tentar acessar o backend.

Redis¶

Com o Redis, você tem a flexibilidade de escolher entre usar o backend assíncrono ou síncrono, simplesmente especificando a string de conexão.

RedisBackend¶

Parâmetro	Tipo	Descrição	Padrão
`url`	`str`	URL do Redis.	`'redis://localhost:6379'`
`kwargs`	`dict[str, Any]`	Recebe os mesmos argumentos do construtor que `redis.client.Redis.from_url`. O parâmetro `decode_responses` sempre será `True` pois o resultado precisa ser retornado como uma string.	`{}`

AsyncRedisBackend¶

Parâmetro	Tipo	Descrição	Padrão
`url`	`str`	URL do Redis.	`'redis://localhost:6379'`
`kwargs`	`dict[str, Any]`	Recebe os mesmos argumentos do construtor que `redis.asyncio.client.Redis.from_url`. O parâmetro `decode_responses` sempre será `True` pois o resultado precisa ser retornado como uma string.	`{}`

Mongo¶

Mongo também oferece suporte a backend assíncrono e síncrono

MongoBackend¶

Parâmetro	Tipo	Descrição	Padrão
`host`	`str`	URI do MongoDB.	`'localhost'`
`database`	`str`	Nome do banco de dados de cache.	`'.cachetoolz'`
`kwargs`	`dict[str, Any]`	Toma os mesmos argumentos do construtor que `pymongo.mongo_client.MongoClient`.	`{}`

AsyncMongoBackend¶

Parâmetro	Tipo	Descrição	Padrão
`host`	`str`	URI do MongoDB.	`'localhost'`
`database`	`str`	Nome do banco de dados de cache.	`'.cachetoolz'`
`kwargs`	`dict[str, Any]`	Toma os mesmos argumentos do construtor que `pymongo.mongo_client.MongoClient`.	`{}`

Coder¶

O objeto codificador é responsável por codificar e decodificar objetos python para json a serem armazenados em cache. Algumas classes já são suportadas, mas se precisar você pode adicionar novos codificadores e decodificadores.

Tipos suportados por padrão:¶

None
bytes
str
int
float
bool
dict
set
frozenset
list
tuple # é decodificado para uma lista
uuid.UUID
pathlib.Path
collections.deque
re.Pattern
datetime.time
datetime.date
datetime.datetime
datetime.timedelta
decimal.Decimal
ipaddress.IPv4Address
apaddress.IPv4Interface
apaddress.IPv4Network
apaddress.IPv6Address
apaddress.IPv6Interface
apaddress.IPv6Network

Registrar codificador¶

Você pode cadastrar uma classe para decodificação, ela precisa ter os métodos encode e decode onde o método encode deve ter um parâmetro chamado value e deve ter o tipo anotado. Esses métodos podem ser instance, @staticmethod ou @classmethod. A função decodificar receberá o valor exato retornado pela função de codificação.

Métodos de classe

from collections import deque

@coder.register
class DequeCoder:
    @classmethod
    def encode(cls, value: deque):
        return {'iterable': list(value), 'maxlen': value.maxlen}

    @classmethod
    def decode(cls, value):
        return deque(val['iterable'], val['maxlen'])

Métodos estáticos

from collections import deque

@coder.register
class DequeCoder:
    @staticmethod
    def encode(value: deque):
        return {'iterable': list(value), 'maxlen': value.maxlen}

    @staticmethod
    def decode(value):
        return deque(val['iterable'], val['maxlen'])

Métodos de instância

from collections import deque

@coder.register
class DequeCoder:
    def encode(self, value: deque):
        return {'iterable': list(value), 'maxlen': value.maxlen}

    def decode(self, value):
         return deque(val['iterable'], val['maxlen'])

Ao registrar uma turma, ela será instanciada. Portanto, se a classe exigir algum parâmetro de inicialização, você poderá registrar uma instância dela junto com os parâmetros necessários.

from collections import deque

class DequeCoder:
    def __init__(self, foo):
        self.foo = foo

     def encode(self, value: deque):
        return {'iterable': list(value), 'maxlen': value.maxlen}

    def decode(self, value):
        return deque(val['iterable'], val['maxlen'])

coder.register(DequeCoder(foo='bar'))

Registrar codificação¶

Caso não haja necessidade de decodificar o resultado ou prefira adicioná-lo separadamente, você tem a opção de cadastrar um único codificador.

from collections import deque

from cachetoolz.coder import encoder


@encoder.register('deque')
def _(value: deque):
    return {'iterable': list(value), 'maxlen': value.maxlen}

Registrar decodificação¶

Ao registar um descodificador, é essencial garantir que o nome corresponde ao nome do codificador. Não fazer isso resultará na falta de conexão entre eles.

from collections import deque

from cachetoolz.coder import decoder

@decoder.register('deque')
def _(value):
    return deque(value['iterable'], value['maxlen'])