snakestack 0.43.0

Uma coleção de utilitários para acelerar o desenvolvimento backend em Python com padrões reutilizáveis e produtivos

🐍 SnakeStack

---

📦 Visão Geral

O snakestack é um pacote modular que oferece uma base robusta para construção de serviços backend com foco em:

• Observabilidade com OpenTelemetry
• Cache assíncrono com Redis
• Integração com Google Pub/Sub
• Acesso assíncrono ao MongoDB
• Client HTTPX com suporte a tracing
• Modelos base com pydantic e pydantic-settings
• Stack de logging estruturado e configurável
• Circuit breaker para proteção de integrações externas

---

⚙️ Instalação

Base

pip install snakestack

Extras disponíveis

Extra	Comando de instalação
Redis	pip install snakestack[redis]
MongoDB	pip install snakestack[mongodb]
Pub/Sub	pip install snakestack[pubsub]
Telemetry	pip install snakestack[telemetry]
Retry	pip install snakestack[retry]
Circuit Breaker	pip install snakestack[circuit-breaker]
Todos	pip install snakestack[all]

---

📚 Módulos disponíveis

Módulo	Descrição	Extra
snakestack.log	Logging estruturado com filtros, formatadores e propagação de request ID	—
snakestack.cache	Cliente Redis assíncrono com TTL, prefixo e serialização via orjson	[redis]
snakestack.pubsub	Publisher e subscriber para Google Pub/Sub com presets e tracing	[pubsub]
snakestack.telemetry	Setup do OpenTelemetry com auto-instrumentação de HTTPX e logging	[telemetry]
snakestack.mongodb	Wrapper assíncrono para PyMongo com singleton de conexão	[mongodb]
snakestack.healthz	Agregador de health checks com medição de latência por check	—
snakestack.http_client	AsyncClient do HTTPX com composição de URL e normalização de erros	—
snakestack.model	Modelos base Pydantic v2: StrictModel, LenientModel, ORMModel	—
snakestack.task	Pool de workers assíncrono sobre asyncio.Queue com shutdown graceful	—
snakestack.retry	Decorator de retry para APIs externas com backoff exponencial, logging e tracing	[retry]
snakestack.circuit_breaker	Circuit breaker com backends em memória ou Redis, presets para HTTP e banco de dados, logging e tracing	[circuit-breaker]
snakestack.config	Gerenciamento de settings com pydantic-settings	—

---

🧪 Exemplos de Uso

Os exemplos completos estão no diretório examples/.

Cache

Arquivo	Descrição
cache/sample_async_service.py	Uso básico do AsyncRedisService com diferentes tipos de valor

Healthz

Arquivo	Descrição
healthz/sample_healthz.py	Health check com callbacks síncronos e assíncronos
healthz/sample_healthz_with_deps.py	Health check verificando Redis e MongoDB

Circuit Breaker

Arquivo	Descrição
circuit_breaker/sample_http.py	Circuit breaker em chamadas HTTP com preset circuit_breaker_http
circuit_breaker/sample_http_with_telemetry.py	Circuit breaker com tracing via OpenTelemetry

HTTPX

Arquivo	Descrição
http_client/sample_default.py	Client HTTPX básico com composição de URL
http_client/sample_with_telemetry.py	Client HTTPX com tracing via OpenTelemetry

Logging

Arquivo	Descrição
log/sample_default.py	Formatter padrão
log/sample_with_request_id.py	Formatter com propagação de request ID
log/sample_custom_json.py	Formatter JSON estruturado
log/sample_filter_excluded_name.py	Filtro por nome do logger
log/sample_filter_path.py	Filtro por path da requisição

MongoDB

Arquivo	Descrição
mongodb/sample_mongodb.py	Conexão, consulta e encerramento com PyMongo assíncrono

Pub/Sub

Arquivo	Descrição
pubsub/sample_publisher.py	Publicação básica de mensagens
pubsub/sample_publisher_with_open_telemetry.py	Publicação com tracing
pubsub/sample_publisher_with_callback.py	Publicação com callback
pubsub/sample_subscriber.py	Subscriber com SimpleProcessor
pubsub/sample_subscriber_with_batch.py	Subscriber com BatchProcessor
pubsub/sample_subscriber_with_open_telemetry.py	Subscriber com tracing
pubsub/sample_flow_control.py	Configuração de flow control

Task

Arquivo	Descrição
task/sample_task.py	Worker pool básico
task/sample_task_with_sentinel.py	Worker pool com sentinel de encerramento
task/sample_task_with_callback.py	Worker pool com callback customizado

Retry

Arquivo	Descrição
retry/sample_with_logging.py	Retry com logging estruturado em chamadas a APIs externas
retry/sample_with_telemetry.py	Retry com eventos de tracing via OpenTelemetry

Telemetry

Arquivo	Descrição
telemetry/sample_telemetry.py	Setup do OpenTelemetry com instrument_app
telemetry/sample_decorator_telemetry.py	Tracing via decorator @instrumented_span

---

🛠️ Desenvolvimento Local

1. Clone o repositório

git clone https://github.com/BrunoSegato/snakestack.git
cd snakestack

2. Instale as dependências

uv sync --all-groups --all-extras

3. Instale os hooks de pre-commit

make hooks

4. Suba os serviços locais

docker-compose up -d

---

🧾 Comandos

Comando	Descrição
make install	Instala apenas dependências de runtime
make lint	Executa ruff check
make format	Executa ruff check --fix
make mypy	Executa mypy sem cache incremental
make check	Executa lint + mypy
make pre-commit	Executa todos os hooks de pre-commit
make test	Executa todos os testes
make test-unit	Executa apenas testes unitários
make test-matching test=nome	Executa testes por nome com -vv -s --pdb
make release	Bump de versão + changelog via commitizen
make publish	Publica no PyPI

---

🧪 Testes

O projeto possui cobertura completa de testes unitários organizados por domínio.

# Todos os testes
make test

# Apenas testes unitários
make test-unit

# Testes de um domínio específico
uv run pytest -m cache
uv run pytest -m pubsub
uv run pytest -m telemetry
uv run pytest -m mongodb
uv run pytest -m retry
uv run pytest -m circuit_breaker

# Teste por nome
make test-matching test='nome_do_teste'