🧩 Exercícios: Comunicação em Tempo Real com WebSockets

Spec Backend Com Python E Fastapi • Trilha Progressiva em 4 Níveis

---

🎯 Nível 1: Fundamentos

Problema 17.1 — Endpoint WebSocket Simples com Echo

Contexto: Endpoint WebSocket Simples com Echo no contexto de Comunicação em Tempo Real com WebSockets.

Requisitos de Execução:

1. Criar endpoint @app.websocket('/ws') que receba mensagens de texto e responda com eco.

Resultado Esperado

Conexão aceita -> Mensagem recebida -> Resposta 'Eco: mensagem' enviada.

📤 Instruções de Entrega (Microsoft Teams)

1. Salve o arquivo como: Atividade_17_1_SeuNome
2. Envie na tarefa: Atividade Cap 17 - Comunicação em Tempo Real com WebSockets

🔑 Gabarito de Código & Solução Comentada

# ws_echo.py
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_echo_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Eco: {data}")

---

🔍 Nível 2: Prática

Problema 17.2 — Gerenciador de Conexões WebSocket (ConnectionManager)

Contexto: Gerenciador de Conexões WebSocket (ConnectionManager) no contexto de Comunicação em Tempo Real com WebSockets.

Requisitos de Execução:

1. Implementar classe ConnectionManager com métodos connect, disconnect e broadcast para múltiplos clientes.

Resultado Esperado

Mensagem transmitida (broadcast) para 3 clientes conectados.

📤 Instruções de Entrega (Microsoft Teams)

1. Salve o arquivo como: Atividade_17_2_SeuNome
2. Envie na tarefa: Atividade Cap 17 - Comunicação em Tempo Real com WebSockets

🔑 Gabarito de Código & Solução Comentada

# ws_manager.py
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[str] = []

    def connect(self, client_id: str):
        self.active_connections.append(client_id)

    def disconnect(self, client_id: str):
        self.active_connections.remove(client_id)

    def broadcast(self, message: str):
        for client in self.active_connections:
            print(f"Enviando '{message}' para cliente {client}")

manager = ConnectionManager()
manager.connect("c1"); manager.connect("c2")
manager.broadcast("Atualização de Cotações")

---

⚡ Nível 3: Integração

Problema 17.3 — Autenticação de Conexão WebSocket via Query Token

Contexto: Autenticação de Conexão WebSocket via Query Token no contexto de Comunicação em Tempo Real com WebSockets.

Requisitos de Execução:

1. Validar token JWT passado como query parameter na inicialização da conexão WebSocket, rejeitando com código 1008 se inválido.

Resultado Esperado

WebSocket: Token validado antes de aceitar handshake.

📤 Instruções de Entrega (Microsoft Teams)

1. Salve o arquivo como: Atividade_17_3_SeuNome
2. Envie na tarefa: Atividade Cap 17 - Comunicação em Tempo Real com WebSockets

🔑 Gabarito de Código & Solução Comentada

# ws_auth.py
def autenticar_websocket_token(token: str):
    if token != "token_valido_123":
        return False, 1008  # Policy Violation
    return True, 1000  # Normal Closure

print("Autenticação WS:", autenticar_websocket_token("token_valido_123"))

---

🏆 Nível 4: Desafio Corporativo

Problema 17.4 — Pub/Sub Distribuído de WebSockets com Redis

Contexto: Pub/Sub Distribuído de WebSockets com Redis no contexto de Comunicação em Tempo Real com WebSockets.

Requisitos de Execução:

1. Integrar o ConnectionManager com Redis Pub/Sub para propagar mensagens entre múltiplos servidores FastAPI.

Resultado Esperado

Mensagem publicada no Redis propagada para nós do cluster.

📤 Instruções de Entrega (Microsoft Teams)

1. Salve o arquivo como: Atividade_17_4_SeuNome
2. Envie na tarefa: Atividade Cap 17 - Comunicação em Tempo Real com WebSockets

🔑 Gabarito de Código & Solução Comentada

# ws_redis_pubsub.py
def propagar_mensagem_cluster(canal: str, payload: dict):
    print(f"Publicando no Redis canal '{canal}': {payload}")
    return True

propagar_mensagem_cluster("chat_geral", {"user": "Alice", "msg": "Olá Cluster"})
logger.info('Etapa de execução concluída')
assert len(str(resultado if 'resultado' in locals() else 'ok')) > 0