Python Redis: Um guia para iniciantes

Usar Redis com Python é uma combinação poderosa para engenheiros de dados que buscam criar sistemas de alto desempenho, escalonáveis e de fácil manutenção. Se você estiver otimizando camadas de cache, gerenciando fluxos de eventos em tempo real ou implementando o bloqueio distribuído, o Redis oferece uma solução flexível e eficiente.

Neste guia, abordaremos os conceitos básicos do uso do Redis com Python para ajudar você a integrá-lo perfeitamente aos seus fluxos de trabalho de engenharia de dados.

O que é o Python Redis?

O Redis é um armazenamento de estrutura de dados na memória, de código aberto, usado como banco de dados, cache e corretor de mensagens. Significa Remote Dictionary Server e foi originalmente projetado para ser um sistema de armazenamento de valores-chave com tempos de acesso rápidos.

Ele oferece suporte a várias estruturas de dados como stringshashes, listas, conjuntos e conjuntos ordenados com consultas de intervalo. O Redis também oferece recursos avançados, como transações, mensagens pub/sub, scripts Lua e recursos integrados de replicação e clustering para melhorar a disponibilidade e o dimensionamento.

Casos de uso comuns do Redis

Os possíveis casos de uso do Redis são vastos e variados.

Alguns dos casos de uso mais comuns em engenharia de dados incluem: 

• Armazenamento em cache: Acelere as consultas ao banco de dados e as respostas da API.
• Filas de mensagens: Potencialize arquiteturas orientadas a eventos e pipelines de processamento de trabalhos.
• Análise em tempo real: Processe e analise dados de streaming com latência mínima.
• Tabelas de classificação: Classifique e recupere com eficiência as entidades com melhor pontuação.
• Armazenamento de sessões: Mantenha um gerenciamento rápido e dimensionável do estado da sessão.
• Gerenciamento de filas de trabalho: Garanta a programação e a execução eficientes das tarefas.
• Indexação geoespacial: Armazenar e consultar dados baseados em localização.

O Redis também é comumente usado em combinação com outras tecnologias, como servidores da Web (por exemplo, NGINX), bancos de dados (por exemplo, MySQL) e sistemas de mensagens (por exemplo, Kafka) para otimizar o desempenho e a escalabilidade.

Por que usar o Redis com Python?

Python é uma linguagem dominante na engenharia de dados devido à sua flexibilidade, ao extenso ecossistema e à facilidade de integração com tecnologias de Big Data. 

A combinação com o Redis pode trazer muitos benefícios para seus aplicativos:

• Desempenho rápido: Como o Redis armazena dados na memória, ele pode oferecer tempos de acesso significativamente mais rápidos em comparação com os bancos de dados tradicionais. bancos de dados tradicionais, que armazenam dados em disco. Isso o torna ideal para casos de uso que exigem alto desempenho, como aplicativos de cache e em tempo real.
• Escalabilidade: O Redis foi projetado com a escalabilidade em mente, facilitando o manuseio de grandes quantidades de dados sem comprometer o desempenho.
• Facilidade de uso: A sintaxe simples do Python e o rico ecossistema de bibliotecas facilitam o trabalho com o Redis. A biblioteca oficial do cliente Redis-py também fornece uma interface amigável para que você interaja com o Redis a partir do seu código Python.

O que é Redis-py?

Redis-py é a biblioteca oficial do cliente Redis para Python. Ele fornece uma interface fácil de usar para que você se comunique com o Redis a partir do código Python.

Alguns de seus principais recursos incluem:

• Suporte para todos os tipos de dados do Redis
• Suporte a pipeline e transações para operações eficientes e atômicas
• Suporte a mensagens Pub/sub
• Codificação/decodificação automática de dados para/de bytes, strings e outros tipos Python
• Suporte ao modo de cluster para trabalhar com clusters Redis

Configurando o Redis para o desenvolvimento em Python

Vamos dar uma olhada em como podemos configurar o Redis para nós mesmos usando Python.

Há três maneiras principais de colocar o Redis em funcionamento:

1. Docker (Recomendado para engenheiros de dados)
2. Subsistema Windows para Linux (WSL)
3. Compilações de terceiros (por exemplo, Memurai, tporadowski/redis)

Método A: Usando o Docker

Em primeiro lugar, ao usar o Docker, você precisará ter o Docker Desktop instalado em seu computador. Você pode fazer o download no site do site do DockerHub.

Quando o Docker estiver instalado, abra o prompt de comando e digite o seguinte comando.

docker pull redis

Em seguida, iniciaremos um novo contêiner.

docker run --name my-redis -p 6379:6379 -d redis

docker pull redis baixa a imagem oficial mais recente do Redis. docker run --name my-redis -p 6379:6379 -d redis inicia um novo contêiner chamado my-redis. O parâmetro -p 6379:6379 mapeia a porta 6379 da sua máquina local para a porta 6379 do contêiner, para que você possa se conectar ao Redis em localhost:6379. O sinalizador -d executa o contêiner em segundo plano (modo desanexado).

Para verificar, execute:

docker exec -it my-redis redis-cli ping

Ele deve responder com PONG.

Método B: Usando o Windows Subsystem for Linux (WSL)

Para máquinas Windows, outro método possível é usar o Subsistema Windows para a instalação. Isso usará o Linux para fazer a instalação.

wsl --install

Esse comando instala e configura a distribuição padrão do Ubuntu dentro da WSL. Reinicialize se necessário.

Em seguida, abra o terminal do Ubuntu (na Microsoft Store ou digitando wsl no PowerShell) e execute:

sudo apt-get update
sudo apt-get install redis-server

Em seguida, inicie o servidor Redis.

sudo service redis-server start

Vamos fazer um teste para ver se o servidor está funcionando.

redis-cli ping

Aqui está uma descrição detalhada dos comandos que usamos acima:

• sudo apt-get update obtém as definições de pacotes mais recentes.
• sudo apt-get install redis-server instala o Redis no ambiente do Ubuntu.
• sudo service redis-server start inicia imediatamente o serviço Redis.
• redis-cli ping testa a conectividade com o servidor Redis local, retornando PONG se for bem-sucedido.

Método C: Compilações de terceiros

• Memurai: Memurai.com
• tporadowski/redis: Lançamentos do GitHub

Essas soluções fornecem binários nativos do Windows. Depois de fazer o download e instalar ou descompactar, execute:

redis-server.exe

Uma vez iniciado, o redis-server.exe deve imprimir registros indicando que está escutando em 127.0.0.1:6379.

Você pode confirmar usando a CLI incluída (se fornecida):

redis-cli.exe ping

Você deve esperar um resultado PONG se for bem-sucedido.

Instalando a biblioteca redis-py

Em seguida, você precisará instalar a biblioteca redis-py para acessar o banco de dados usando Python. Antes de instalar a biblioteca, certifique-se de que você tenha o Python e o Redis instalados separadamente.

pip install redis

Com esse comando, você baixa e instala a biblioteca cliente Redis do Python (redis-py) a partir do PyPI. Se você estiver usando um ambiente virtual, certifique-se de ativá-lo antes de executar esse comando.

Verifique usando:

python -c "import redis; print(redis.__version__)"

Isso deve imprimir um número de versão (por exemplo, 5.2.1).

Configuração de um servidor Redis

Se você seguiu qualquer método acima (Docker, WSL ou uma compilação de terceiros), agora deve ter um servidor Redis em execução em localhost:6379. Isso é suficiente para o desenvolvimento local no Windows.

Trabalho com estruturas de dados do Redis

Abaixo estão exemplos em Python que demonstram como você pode se conectar e manipular diferentes tipos de dados do Redis. Cada bloco de código inclui comentários explicativos que detalham como os métodos ou comandos funcionam.

Strings no Redis

import redis

# Instantiate a Redis client, connecting to localhost on port 6379
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0  # The default Redis database index
)

# 1. SET command: store a string under 'mykey'
r.set("mykey", "hello from Windows")

# 2. GET command: retrieve the value stored at 'mykey'
value = r.get("mykey")
print(value)   # Output is b'hello from Windows', since redis-py returns bytes.

# 3. Convert bytes to string
print(value.decode())  # prints "hello from Windows"

# 4. DEL command: remove 'mykey' from Redis
r.delete("mykey")

Criamos um objeto Redis que se conecta ao servidor Redis local.

• r.set("mykey", "hello from Windows") armazena a string "hello from Windows" na chave "mykey".
• r.get("mykey") obtém o valor armazenado em mykey. O Redis retorna bytes, portanto, nós o decodificamos para obter uma saída legível para humanos.
• r.delete("mykey") exclui completamente a chave mykey.

Exemplo: Armazenamento e recuperação de sessões de usuário

Aqui está um exemplo de código de gerenciamento de sessão de usuário:

import uuid

def store_user_session(user_id):
    """
    Generates a unique session token (UUID) for a user and stores it in Redis.
    The session is stored under the key pattern: user:{user_id}:session
    """
    session_key = f"user:{user_id}:session"
    token = str(uuid.uuid4())   # Generate a random UUID as a session token
    r.set(session_key, token)   # Store token in Redis
    return token

def get_user_session(user_id):
    """
    Retrieves the stored session token for the given user_id.
    Returns None if the session does not exist or is expired.
    """
    session_key = f"user:{user_id}:session"
    token = r.get(session_key)
    return token.decode('utf-8') if token else None

def delete_user_session(user_id):
    """
    Deletes the session entry from Redis for the specified user_id.
    """
    session_key = f"user:{user_id}:session"
    r.delete(session_key)

# Usage demonstration
session_token = store_user_session(1001)
print(f"Stored session token: {session_token}")

retrieved_token = get_user_session(1001)
print(f"Retrieved session token: {retrieved_token}")

delete_user_session(1001)
print(f"Session after delete: {get_user_session(1001)}")  # Should be None or empty

store_user_session(user_id): Cria um ID globalmente exclusivo por meio de uuid.uuid4() e o armazena em uma chave específica do usuário. get_user_session(user_id): Lê de volta o token de sessão. Se None, a sessão está ausente ou expirou. delete_user_session(user_id): Remove a chave de sessão do Redis, efetivamente desconectando o usuário.

Listas no Redis

As listas do Redis são sequências ordenadas de strings. Eles permitem operações em ambas as extremidades, o que os torna úteis para filas, pilhas ou registros.

# LPUSH: Push an element to the head (left) of the list
r.lpush("task_queue", "task1")

# RPUSH: Push an element to the tail (right) of the list
r.rpush("task_queue", "task2")
r.rpush("task_queue", "task3")

# LPOP: Pop (remove and return) the element at the head
task = r.lpop("task_queue")
print(task)  # b'task1'

# Optional: RPOP removes and returns the element at the tail
task = r.rpop("task_queue")
print(task)  # b'task3'

lpush("task_queue", "task1") adiciona "task1" ao lado esquerdo de uma lista chamada "task_queue". rpush("task_queue", "task2") e rpush("task_queue", "task3") adicionam tarefas ao lado direito. lpop("task_queue") and rpop("task_queue") remove e retorna itens das respectivas extremidades da lista.

Exemplo: Implementação de uma fila simples apoiada no Redis

def enqueue_task(queue_name, task):
    """
    Appends a task to the end (right) of the Redis list named queue_name.
    """
    r.rpush(queue_name, task)

def dequeue_task(queue_name):
    """
    Removes a task from the front (left) of the Redis list named queue_name.
    Returns the task as a string, or None if the queue is empty.
    """
    task = r.lpop(queue_name)
    return task.decode('utf-8') if task else None

# Example usage:
enqueue_task("my_queue", "send_email")
enqueue_task("my_queue", "generate_report")

while True:
    task = dequeue_task("my_queue")
    if not task:
        print("No more tasks in queue.")
        break
    print(f"Processing task: {task}")

enqueue_task sempre adiciona novas tarefas ao final da lista, o que simula uma fila FIFO tradicional. dequeue_task remove tarefas da frente. Se a lista estiver vazia, você retornará None.

O loop while processa as tarefas até que a fila esteja vazia e, em seguida, para.

Hashes no Redis

Os hashes são semelhantes aos dicionários do Python, mas são armazenados no Redis. Eles são melhores para agrupar campos relacionados (por exemplo, detalhes do usuário).

# HSET: Store 'name' and 'email' fields for a user hash key
r.hset("user:1001", "name", "Alice")
r.hset("user:1001", "email", "alice@example.com")

# HGET: Retrieve a single field from the hash
email = r.hget("user:1001", "email")
print(email.decode('utf-8'))  # alice@example.com

# HDEL: Remove a field from the hash
r.hdel("user:1001", "email")

hset("user:1001", "name", "Alice") define o campo "name" como o valor "Alice" dentro do hash na chave "user:1001". hget("user:1001", "email") busca o campo "email". hdel("user:1001", "email") exclui o campo "email" desse hash.

Exemplo: Armazenamento e acesso a perfis de usuário estruturados

def create_user_profile(user_id, name, email):
    """
    Creates a user profile in Redis under the key 'user:{user_id}'.
    'name' and 'email' are stored as separate fields in the hash.
    """
    user_key = f"user:{user_id}"
    r.hset(user_key, mapping={"name": name, "email": email})

def get_user_profile(user_id):
    """
    Retrieves and returns all fields in the user profile hash
    as a Python dictionary. Keys and values are decoded from bytes.
    """
    user_key = f"user:{user_id}"
    profile_data = r.hgetall(user_key)
    return {k.decode('utf-8'): v.decode('utf-8') for k, v in profile_data.items()}

def delete_user_profile(user_id):
    """
    Deletes the entire user profile key from Redis.
    """
    user_key = f"user:{user_id}"
    r.delete(user_key)

# Usage demonstration
create_user_profile(1002, "Bob", "bob@example.com")
print(get_user_profile(1002))  # e.g. {'name': 'Bob', 'email': 'bob@example.com'}
delete_user_profile(1002)

Armazenamos o nome e o e-mail de um usuário em um hash. mapping={"name": name, "email": email} usa a capacidade de redis-pyde definir vários campos de uma só vez. r.hgetall(user_key) recupera todos os campos e valores, retornados como um dicionário de bytes brutos que decodificamos para strings. delete_user_profile remove toda a chave do hash.

Conjuntos e conjuntos ordenados

Os dados no Redis podem ser gerenciados usando conjuntos ou conjuntos ordenados. 

Os conjuntos são coleções de valores exclusivos, enquanto os conjuntos ordenados são coleções de pares de valores-chave com uma pontuação associada a cada valor. Essas estruturas de dados oferecem maneiras eficientes de armazenar e recuperar dados, o que as torna úteis para aplicativos que exigem pesquisas rápidas.

Conjuntos

Um conjunto no Redis é semelhante a um conjunto matemático - ele contém uma coleção de elementos exclusivos, e não são permitidos elementos duplicados. Os conjuntos podem ser usados para modelar relacionamentos entre diferentes entidades, como usuários que curtiram uma publicação nas mídias sociais.

Aqui está uma demonstração de conjuntos:

# SADD: Add multiple members to a set
r.sadd("tags:python", "redis", "windows", "backend")

# SMEMBERS: Retrieve all unique members in the set
tags = r.smembers("tags:python")
print(tags)  # {b'redis', b'windows', b'backend'}

Os Redis Sets garantem a exclusividade dos membros. A tentativa de adicionar um membro duplicado não terá efeito.

O resultado de smembers é um conjunto de bytes Python.

Conjuntos ordenados

Os conjuntos ordenados no Redis são semelhantes aos conjuntos regulares, mas cada membro também tem uma pontuação correspondente. Isso permite operações eficientes de classificação e ordenação no conjunto.

Os conjuntos ordenados são usados com frequência em tabelas de classificação, em que a pontuação representa a classificação ou os pontos de um jogador.

Aqui está uma demonstração de conjuntos ordenados:

# ZADD: Add members with scores
r.zadd("leaderboard", {"player1": 10, "player2": 20})

# ZRANGE: Retrieve members in ascending order of score
leaders = r.zrange("leaderboard", 0, -1, withscores=True)
print(leaders)  # [(b'player1', 10.0), (b'player2', 20.0)]

Os conjuntos ordenados armazenam os membros em uma ordem específica determinada por suas pontuações numéricas. zrange("leaderboard", 0, -1, withscores=True) retorna todos os membros da classificação 0 até o final, incluindo suas pontuações.

Exemplo: Gerenciar tags ou placares de líderes

O uso de conjuntos ordenados pode ser útil para gerenciar tags ou criar tabelas de classificação. 

def add_tag(post_id, tag):
    """
    Adds a 'tag' to the set of tags belonging to a specific post.
    Each post has its own set under 'post:{post_id}:tags'.
    """
    r.sadd(f"post:{post_id}:tags", tag)

def get_tags(post_id):
    """
    Retrieves all tags for a specific post, decoding the bytes into strings.
    """
    raw_tags = r.smembers(f"post:{post_id}:tags")
    return {tag.decode('utf-8') for tag in raw_tags}

def update_leaderboard(player, score):
    """
    Updates or inserts a player's score in the 'game:leaderboard' sorted set.
    A higher score indicates a better position if sorting descending.
    """
    r.zadd("game:leaderboard", {player: score})

def get_leaderboard():
    """
    Returns an ascending list of (player, score) tuples from the leaderboard.
    To invert the sorting (highest first), you'd use ZREVRANGE.
    """
    entries = r.zrange("game:leaderboard", 0, -1, withscores=True)
    return [(player.decode('utf-8'), score) for player, score in entries]

# Usage demonstration
add_tag(123, "python")
add_tag(123, "redis")
print(get_tags(123))

update_leaderboard("Alice", 300)
update_leaderboard("Bob", 450)
print(get_leaderboard())

add_tag e get_tags demonstram um sistema de marcação de um para muitos, armazenando tags em um conjunto. update_leaderboard e get_leaderboard mostram como você pode implementar uma tabela de classificação de jogos.

Recursos do Redis em aplicativos Python

O Redis tem vários recursos que o tornam uma escolha popular para o armazenamento de dados em aplicativos Python: enfileiramento, bloqueio e armazenamento em cache. 

A seguir, você encontrará suas implementações em Python:

Redis como uma fila (com bloqueio)

Vamos explorar o uso do Redis como uma fila com bloqueio usando BLPOP.

BLPOP é uma operação de bloqueio que aguarda até que um elemento esteja disponível em uma lista.

def blocking_consumer(queue_name):
    """
    Continuously listens to the specified queue (Redis list) using BLPOP,
    which blocks until new items are pushed. Once an item arrives,
    it is removed from the queue and processed.
    """
    print(f"Waiting on queue: {queue_name}")
    while True:
        result = r.blpop(queue_name)
        if result:
            list_name, task_bytes = result
            task = task_bytes.decode('utf-8')
            print(f"Received task: {task}")
        else:
            print("Queue is empty or an error occurred.")
            break

def enqueue_task(queue_name, task):
    """
    Pushes a task to the end of a Redis list (queue).
    """
    r.rpush(queue_name, task)

# Example usage:
enqueue_task("blocking_queue", "task_block_1")
enqueue_task("blocking_queue", "task_block_2")

# In a real application, the consumer might run in a separate thread or process
blocking_consumer("blocking_queue")

blocking_consumer usa blpop em um loop. Se a lista estiver vazia, o blpop aguardará até que outro item seja enviado.

Quando um item é recebido, ele é removido da lista e impresso.

Essa abordagem é ideal para padrões de produtor-consumidor em que os funcionários consomem tarefas à medida que elas aparecem.

Implementação de bloqueios no Redis

Os bloqueios distribuídos impedem que vários clientes ou processos modifiquem simultaneamente o mesmo recurso. O Redis pode ajudar a evitar condições de corrida em um ambiente distribuído.

import time
from redis.exceptions import LockError

def process_critical_section():
    """
    Acquires a lock named 'resource_lock' with a timeout of 10 seconds.
    The lock automatically expires after 10 seconds to prevent deadlocks.
    """
    lock = r.lock("resource_lock", timeout=10)
    try:
        # Attempt to acquire the lock, wait for up to 5 seconds if another process holds it
        acquired = lock.acquire(blocking=True, blocking_timeout=5)
        if acquired:
            print("Lock acquired; performing critical operation...")
            time.sleep(3)  # Simulate some operation
        else:
            print("Failed to acquire lock within 5 seconds.")
    except LockError:
        print("A LockError occurred, possibly releasing already released lock.")
    finally:
        # Always release the lock in a finally block to ensure cleanup
        lock.release()
        print("Lock released.")

# Usage demonstration
process_critical_section()

Criamos um bloqueio com timeout=10, o que significa que, se o bloqueio não for liberado manualmente, o Redis o removerá automaticamente em 10 segundos para evitar o bloqueio indefinido (deadlock). lock.acquire(blocking=True, blocking_timeout=5) tenta adquirir o bloqueio por 5 segundos antes de desistir. Após a conclusão, o site lock.release() libera o recurso para outros processos.

Armazenamento em cache com o Redis

O armazenamento em cache é um cenário de uso comum: armazenar dados acessados com frequência na memória, reduzindo a carga em bancos de dados ou APIs externas.

import requests
import json

def get_user_data(user_id):
    """
    Retrieves user data from a hypothetical API endpoint.
    If the data is found in Redis (cache), use that. Otherwise, call the API,
    store the response in Redis with a 60-second expiration, and return it.
    """
    cache_key = f"user_data:{user_id}"
    cached_data = r.get(cache_key)
    if cached_data:
        print("Cache hit!")
        return json.loads(cached_data)

    print("Cache miss. Fetching from API...")
    response = requests.get(f"https://api.example.com/users/{user_id}")
    user_info = response.json()

    # Store in Redis for 60 seconds
    r.setex(cache_key, 60, json.dumps(user_info))
    return user_info

# Usage
user = get_user_data(42)  # First call => cache miss
user_again = get_user_data(42)  # Subsequent call => cache hit

Verificamos se user_data:{user_id} existe no Redis. Se isso acontecer, você será atingido pelo cache e ignoraremos a chamada à API. Caso contrário, buscamos a API remota e, em seguida, setex (set + expiração) por 60 segundos. As chamadas subsequentes dentro desse período de tempo recuperarão os dados armazenados em cache, reduzindo a latência.

Tópicos avançados em Redis

Você acha que já está familiarizado com o Redis? Vamos examinar alguns tópicos mais avançados.

Usando o Pub/Sub com o Redis

Pub/Sub (Publish/Subscribe) é um padrão de mensagens popular para comunicação em tempo real. Os editores enviam mensagens para um canal, e todos os assinantes desse canal recebem as mensagens.

import threading

def subscriber(r, channel_name):
    """
    Subscribes to the given Redis channel and listens for messages.
    When a new message is published on that channel, it is printed.
    """
    pubsub = r.pubsub()
    pubsub.subscribe(channel_name)
    print(f"Subscribed to {channel_name}")

    # pubsub.listen() yields messages from the subscribed channel(s) in real time
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"Received message: {message['data'].decode('utf-8')}")

def publisher(r, channel_name, message):
    """
    Publishes a message to the specified Redis channel.
    All subscribers to this channel immediately receive the message.
    """
    r.publish(channel_name, message)

# Example usage
channel = "updates"

# Start subscriber in a separate thread to avoid blocking the main thread
sub_thread = threading.Thread(target=subscriber, args=(r, channel))
sub_thread.start()

# Publish messages
publisher(r, channel, "Hello from Windows!")
publisher(r, channel, "Another update!")

pubsub = r.pubsub() cria um objeto PubSub. pubsub.subscribe(channel_name) instrui o Redis a enviar mensagens desse canal para o objeto PubSub. pubsub.listen() é um iterador infinito, produzindo mensagens à medida que elas chegam. r.publish(channel_name, message) envia uma mensagem para todos os assinantes de channel_name.

Expiração de chaves no Redis

O Redis oferece suporte a TTL (time-to-live). Isso é essencial para dados temporários, como sessões ou caches de curta duração.

# EXPIRE: Set a 30-second expiration on a key
r.set("temp_key", "some value")
r.expire("temp_key", 30)

# SETEX: Combined set + expire in one command
r.setex("temp_key2", 60, "another value")

r.expire("temp_key", 30) define um TTL de 30 segundos. Após 30 segundos, temp_key é excluído automaticamente. r.setex("temp_key2", 60, "another value") é um atalho que define a chave e a expiração ao mesmo tempo.

Exemplo: Expiração da sessão

Aqui está um exemplo de como você pode usar chaves de expiração para definir uma sessão para expirar após um período de tempo especificado.

def store_session_with_expiry(user_id, token, ttl=3600):
    """
    Stores a session token for a specific user with a time-to-live (TTL).
    By default, the session expires after 1 hour (3600 seconds).
    """
    session_key = f"user:{user_id}:session"
    r.setex(session_key, ttl, token)

def get_session_with_expiry(user_id):
    """
    Retrieves the session token for the user. Returns None if the key doesn't exist
    or if it has expired.
    """
    session_key = f"user:{user_id}:session"
    token = r.get(session_key)
    return token.decode('utf-8') if token else None

# Usage
store_session_with_expiry(2001, "session_token_abc", 3600)
retrieved_token = get_session_with_expiry(2001)
print(f"Retrieved token: {retrieved_token}")

store_session_with_expiry usa o setex para armazenar o user:{user_id}:session com um TTL especificado. Se você não passar um TTL, a chave poderá permanecer indefinidamente.

Quando o TTL expira, o Redis remove a chave automaticamente, impossibilitando a recuperação da sessão.

Práticas recomendadas para usar o Redis com Python

O manuseio do Redis requer algumas considerações e práticas recomendadas para garantir o desempenho ideal e evitar armadilhas comuns.

Algumas dicas que você deve ter em mente incluem:

• Configure adequadamente as opções de persistência para a segurança dos dados.
• Monitore o uso da memória para evitar erros de falta de memória.
• Use pipelining para operações em lote eficientes.
• Fique atento à latência da rede, pois ela pode afetar o desempenho ao trabalhar com um servidor Redis remoto.
• Proteja as conexões do Redis com senhas e firewalls.

Conclusão

O Redis é uma ferramenta poderosa e versátil para gerenciar dados em aplicativos Python. Ao usar o Python com o Redis, você pode criar muitos aplicativos e casos de uso, desde o armazenamento de dados em cache até o gerenciamento de filas. 

Você quer saber mais sobre o Redis? Nossa Introdução ao NoSQL pode ajudar você a aprender como lidar com dados de valor-chave. Você está interessado em Python? Então, nosso Fundamentos de dados Python pode ser o melhor lugar para você começar.