Ficha 02 · Redis e Elasticsearch · Ficha de trabalho

Aluno(a)

Turma

Nº

Data

Exercício 1 · Setup Redis

Instala Redis localmente (Docker) e testa com redis-cli. Faz SET/GET, INCR e EXPIRE.

Resposta:

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

# Aceder cli
docker exec -it redis redis-cli

Dentro do redis-cli:

SET nome "Ana"
GET nome
# "Ana"

INCR counter
INCR counter
GET counter
# "2"

SETEX session 60 "data"
TTL session
# 60 (decreasing)

DEL nome
EXISTS nome
# 0

Exercício 2 · Tipos de dados Redis

Testa cada tipo: - Hash: guardar utilizador (user:1 com name, age, email). - List: queue com 3 tasks (push, pop). - Set: tags (tags:post1 com python, web, ai). - Sorted set: leaderboard com 3 jogadores.

Resposta:

# Hash
HSET user:1 name "Ana" age 25 email "ana@x.pt"
HGETALL user:1
HGET user:1 name
HINCRBY user:1 age 1
HGET user:1 age   # "26"

# List (queue)
RPUSH queue "task1" "task2" "task3"
LRANGE queue 0 -1
LPOP queue        # "task1"
LLEN queue        # 2

# Set
SADD tags:post1 "python" "web" "ai"
SMEMBERS tags:post1
SISMEMBER tags:post1 "python"   # 1
SADD tags:post1 "python"        # 0 (já existe)
SCARD tags:post1                # 3

# Sorted set (leaderboard)
ZADD leaderboard 100 "Ana" 200 "Bruno" 150 "Carla"
ZREVRANGE leaderboard 0 -1 WITHSCORES
# 1) "Bruno"  2) "200"
# 3) "Carla"  4) "150"
# 5) "Ana"    6) "100"

ZINCRBY leaderboard 50 "Ana"
ZRANK leaderboard "Bruno"       # 0 (top em DESC view)

Exercício 3 · Python — Cache pattern

Cria função Python get_post(id) que primeiro tenta Redis cache (TTL 5min), e só vai à BD se miss. Simula a "BD" com um dict.

Resposta:

pip install redis

import redis
import json
import time

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# "BD" simulada
posts_db = {
    1: {"title": "Post 1", "body": "..."},
    2: {"title": "Post 2", "body": "..."},
}

def fetch_from_db(id):
    print(f"  [DB hit] Fetching post {id}")
    time.sleep(0.5)   # simular latência
    return posts_db.get(id)

def get_post(id):
    key = f"post:{id}"

    # Try cache
    cached = r.get(key)
    if cached:
        print(f"  [CACHE hit] {key}")
        return json.loads(cached)

    # Miss → fetch
    post = fetch_from_db(id)
    if post:
        r.setex(key, 300, json.dumps(post))   # 5min TTL
    return post

# Test
print("Primeira chamada:")
print(get_post(1))

print("\nSegunda chamada:")
print(get_post(1))   # CACHE hit, instantâneo

print("\nOutro post:")
print(get_post(2))

print("\nLimpar cache:")
r.delete("post:1")
print(get_post(1))   # DB hit de novo

Exercício 4 · Rate limiting com Redis

Implementa função is_rate_limited(user_id, limit=10, window=60) que retorna True se utilizador excedeu limite no último minuto.

Resposta:

import redis
import time

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def is_rate_limited(user_id: str, limit: int = 10, window: int = 60) -> bool:
    key = f"rate:{user_id}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)
    return count > limit

# Test — 15 chamadas, limit 10/min
for i in range(15):
    if is_rate_limited("user1", limit=10, window=60):
        print(f"Request {i+1}: ❌ RATE LIMITED")
    else:
        print(f"Request {i+1}: ✅ OK")
    time.sleep(0.1)

Output esperado: 10 OK, depois 5 RATE LIMITED.

Em produção: usar sliding window com sorted set para ser mais preciso.

Exercício 5 · Setup Elasticsearch

Instala Elasticsearch via Docker. Faz curl ao endpoint root e indexa um documento.

Resposta:

docker run -d -p 9200:9200 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
  --name es elasticsearch:8.12.0

Aguardar 30s para iniciar.

# Verificar
curl http://localhost:9200
# {"name": "...", "cluster_name": "...", ...}

# Indexar
curl -X POST "localhost:9200/products/_doc" \
  -H 'Content-Type: application/json' \
  -d '{"name": "Caneta azul Bic", "price": 1.50, "category": "Escolar"}'

curl -X POST "localhost:9200/products/_doc" \
  -H 'Content-Type: application/json' \
  -d '{"name": "Caderno A4 quadriculado", "price": 5.00, "category": "Escolar"}'

# Pesquisar
curl -X GET "localhost:9200/products/_search?q=caneta&pretty"

Exercício 6 · Elasticsearch full-text search (Python)

Cria script Python com elasticsearch client: - Indexa 5 documentos (livros: title, author, year). - Pesquisa por palavra no title. - Filtro por year range.

Resposta:

pip install elasticsearch

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")

# Index
livros = [
    {"title": "Os Maias", "author": "Eça de Queirós", "year": 1888},
    {"title": "Memorial do Convento", "author": "José Saramago", "year": 1982},
    {"title": "Mensagem", "author": "Fernando Pessoa", "year": 1934},
    {"title": "O Ano da Morte de Ricardo Reis", "author": "Saramago", "year": 1984},
    {"title": "A Cidade e as Serras", "author": "Eça de Queirós", "year": 1901},
]

for l in livros:
    es.index(index="livros", document=l)

# Refresh (force indexing)
es.indices.refresh(index="livros")

# 1. Full-text search por title
print("Busca por 'morte':")
results = es.search(index="livros", query={"match": {"title": "morte"}})
for h in results["hits"]["hits"]:
    print(f"  {h['_source']['title']} (score {h['_score']:.2f})")

# 2. Filter por year range
print("\nLivros do século XX (1900-2000):")
results = es.search(
    index="livros",
    query={"range": {"year": {"gte": 1900, "lt": 2000}}}
)
for h in results["hits"]["hits"]:
    print(f"  {h['_source']['title']} ({h['_source']['year']})")

# 3. Combinado (bool)
print("\nSaramago do anos 80:")
results = es.search(
    index="livros",
    query={
        "bool": {
            "must": [{"match": {"author": "Saramago"}}],
            "filter": [{"range": {"year": {"gte": 1980, "lt": 1990}}}]
        }
    }
)
for h in results["hits"]["hits"]:
    print(f"  {h['_source']['title']} ({h['_source']['year']})")

Exercício 7 · Aggregations Elasticsearch

Usando o índice de livros do exercício 6, calcula: - Quantos livros por autor. - Média de anos por autor.

Resposta:

results = es.search(
    index="livros",
    size=0,    # não queremos os docs, só aggs
    aggs={
        "por_autor": {
            "terms": {"field": "author.keyword"},
            "aggs": {
                "media_ano": {"avg": {"field": "year"}}
            }
        }
    }
)

print("Estatísticas por autor:")
for bucket in results["aggregations"]["por_autor"]["buckets"]:
    print(f"  {bucket['key']}: {bucket['doc_count']} livros, "
          f"ano médio {bucket['media_ano']['value']:.0f}")

Output:

  Eça de Queirós: 2 livros, ano médio 1894
  Saramago: 2 livros, ano médio 1983
  José Saramago: 1 livros, ano médio 1982
  Fernando Pessoa: 1 livros, ano médio 1934

Note: Elasticsearch trata "Saramago" e "José Saramago" como diferentes — em prod normalizarias.

Exercício 8 · Stack integrada — MongoDB + Redis cache

Sistema: - MongoDB tem 100 produtos. - Função get_product(id): - Primeiro tenta Redis (key product:id). - Se miss, vai a MongoDB. - Guarda em Redis com TTL 60s.

Demonstra speedup do cache.

Resposta:

from pymongo import MongoClient
import redis
import json
import time

# Connect
mongo = MongoClient("mongodb://localhost:27017/").loja
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# Seed (uma vez)
if mongo.products.count_documents({}) == 0:
    for i in range(1, 101):
        mongo.products.insert_one({
            "_id": f"p{i}",
            "name": f"Product {i}",
            "price": round(i * 1.5, 2),
            "stock": 100 - i
        })
    print("Seeded 100 products")

def get_product(id: str):
    key = f"product:{id}"

    # Cache try
    cached = r.get(key)
    if cached:
        return ("CACHE", json.loads(cached))

    # Mongo fetch
    p = mongo.products.find_one({"_id": id})
    if p:
        # Convert ObjectId/etc to JSON-serializable
        r.setex(key, 60, json.dumps(p, default=str))
    return ("DB", p)

# Test
ids = ["p1", "p5", "p10", "p1", "p5", "p20", "p1"]

for id in ids:
    start = time.perf_counter()
    source, p = get_product(id)
    elapsed_ms = (time.perf_counter() - start) * 1000
    print(f"  {id}: [{source}] {p['name']} ({elapsed_ms:.1f} ms)")

Output típico:

p1: [DB]    Product 1 (3.2 ms)
p5: [DB]    Product 5 (2.8 ms)
p10: [DB]   Product 10 (2.5 ms)
p1: [CACHE] Product 1 (0.4 ms)   ← 8x mais rápido
p5: [CACHE] Product 5 (0.3 ms)
p20: [DB]   Product 20 (2.7 ms)
p1: [CACHE] Product 1 (0.3 ms)

Cache hits ~10x mais rápidos. Em produção com network, diferença é ainda maior (Redis em RAM vs Mongo em SSD).