robin-commons 0.1.0

Common infrastructure components for Robin microservices

Common Infrastructure Library

A comprehensive Python library providing battle-tested infrastructure components for building resilient, observable microservices. Designed for teams building distributed systems with Python.

Overview

This library extracts common patterns and utilities used in production microservices, enabling:

• Consistent infrastructure patterns across services
• Single source of truth for resilience, observability, and messaging
• Rapid onboarding of new services
• Independent versioning and upgrade paths

Components

🛡️ Resilience: Circuit Breaker

Prevent cascading failures with automatic circuit breaking. Includes three states (CLOSED, HALF_OPEN, OPEN) and configurable failure thresholds.

from robin_commons.resilience import CircuitBreaker, CircuitBreakerConfig

config = CircuitBreakerConfig(failure_threshold=5, recovery_timeout=60)
breaker = CircuitBreaker(config)

async with breaker:
    await external_service.call()

📝 Logging: Structured JSON Logging

Production-ready logging with JSON output and context propagation. Built on Loguru with environment-aware formatting.

from robin_commons.log import logger, configure_logging

configure_logging(environment="production")
logger.info("Application started", service="my-service", version="1.0.0")

💾 Cache: Redis Client

Redis client with automatic cluster detection and connection pooling. Handles both single-node and cluster deployments.

from robin_commons.cache import get_redis_client

redis = get_redis_client(host="localhost", port=6379)
await redis.set("key", "value")

📊 Telemetry: Observability Suite

Complete observability stack with distributed tracing, metrics collection, and request correlation using OpenTelemetry.

Tracing & Metrics:

from robin_commons.telemetry import TracingService, MetricsCollector, TelemetryConfig

config = TelemetryConfig(
    service_name="my-service",
    otlp_endpoint="http://localhost:4317"
)
tracing = TracingService(config)
metrics = MetricsCollector(config)

# Traces and metrics are automatically collected
with tracing.span("process_request") as span:
    metrics.record_http_request(method="GET", status=200)

Request Correlation:

from robin_commons.telemetry import CorrelationContext

context = CorrelationContext()
correlation_id = context.get_or_create_correlation_id()

FastAPI Integration:

from fastapi import FastAPI
from robin_commons.telemetry import ObservabilityMiddleware

app = FastAPI()
app.add_middleware(ObservabilityMiddleware)

📨 Messaging: NATS Client

Production-grade NATS client with JetStream support for durable pub/sub messaging.

from robin_commons.messaging import NATSClient, MessagingConfig

config = MessagingConfig(nats_url="nats://localhost:4222")
client = NATSClient(config)

await client.connect()
await client.publish("events.user.created", {"user_id": 123})

# Subscribe with durability
async def handle_message(msg):
    print(f"Received: {msg.data}")

await client.subscribe("events.user.*", handle_message)

Typed Event Publishing:

from robin_commons.messaging import EventPublisher
from typing import TypedDict

class UserCreatedEvent(TypedDict):
    user_id: int
    email: str

publisher = EventPublisher(client)
await publisher.publish("user.created", UserCreatedEvent(user_id=1, email="[e-mail]"))

Installation

From PyPI (when available)

pip install robin-commons

From Source

git clone https://github.com/neeve-ai/robin-commons.git
cd robin-commons
pip install -e .

Optional Dependencies

Install additional instrumentation for specific frameworks:

# FastAPI and SQLAlchemy observability
pip install robin-commons[fastapi,sqlalchemy]

# All optional dependencies
pip install robin-commons[all]

Quick Start

1. Set Up Logging

from robin_commons.log import configure_logging, logger

configure_logging(environment="development")
logger.info("Application initialized")

2. Initialize Observability

from robin_commons.telemetry import TelemetryConfig, TracingService, MetricsCollector

config = TelemetryConfig(
    service_name="my-service",
    environment="production",
    otlp_endpoint="http://localhost:4317",
    sample_rate=0.1
)

tracing = TracingService(config)
metrics = MetricsCollector(config)

3. Add Circuit Breaker

from robin_commons.resilience import CircuitBreaker, CircuitBreakerConfig

breaker_config = CircuitBreakerConfig(
    failure_threshold=5,
    recovery_timeout=30,
    expected_exception=ConnectionError
)
breaker = CircuitBreaker(breaker_config)

try:
    async with breaker:
        await call_external_service()
except CircuitBreakerError:
    logger.warning("Circuit breaker is open")

4. Set Up NATS Messaging

from robin_commons.messaging import NATSClient, MessagingConfig

config = MessagingConfig(
    nats_url="nats://localhost:4222",
    max_reconnect_attempts=10
)
client = NATSClient(config)
await client.connect()

Configuration

All components support configuration via environment variables:

# Logging
ENVIRONMENT=production

# Redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_MAX_CONNECTIONS=100

# Observability
SERVICE_NAME=my-service
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
OTEL_TRACES_SAMPLER_ARG=0.1

# NATS
NATS_URL=nats://localhost:4222

Architecture

robin_commons/
├── resilience/          # Circuit breaker and fault tolerance
├── log/                 # Structured logging
├── cache/               # Redis client utilities
├── telemetry/           # OpenTelemetry observability
│   ├── tracing.py       # Distributed tracing
│   ├── metrics.py       # Metrics collection
│   ├── correlation.py   # Request correlation
│   ├── middleware.py    # Framework integration
│   └── instrumentation/ # Auto-instrumentation
└── messaging/           # NATS client and event publishing
    ├── client.py        # NATS connection management
    ├── streaming.py     # JetStream utilities
    └── event_publisher/ # Typed event publishing

Versioning

This library follows Semantic Versioning:

• MAJOR: Breaking API changes
• MINOR: New features (backward-compatible)
• PATCH: Bug fixes

See CHANGELOG.md for detailed version history.

Testing

Run tests locally:

pytest tests/ -v

# With coverage
pytest tests/ --cov=robin_commons --cov-report=html

Run integration tests (requires Docker):

docker-compose -f docker-compose.test.yml up
pytest tests/integration/ -v

Troubleshooting

Circuit Breaker Always Open

Check your failure_threshold and recovery_timeout settings. Ensure the external service is actually recovering.

Missing Traces in OTLP Collector

Verify OTEL_EXPORTER_OTLP_ENDPOINT is correct and the collector is running.

NATS Connection Timeout

Check firewall rules and ensure NATS server is accessible at the configured URL.

See docs/ for detailed troubleshooting guides.

Contributing

Contributions are welcome! Please follow these guidelines:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Add tests for new functionality
4. Ensure all tests pass (pytest)
5. Commit with clear messages (git commit -m 'Add amazing feature')
6. Push to your branch (git push origin feature/amazing-feature)
7. Open a Pull Request

Code Style

• Use Black for formatting
• Follow PEP 8 conventions
• Add type hints to all functions
• Maintain > 90% test coverage

License

This project is licensed under the Apache License, Version 2.0.

You are free to use, modify, and distribute this library in accordance with the terms of the license. A copy of the license is available in the LICENSE file.

Scope Clarification

This repository contains open-source shared libraries used within the Robin ecosystem, such as common utilities, logging infrastructure, and foundational components.

It does not include:

• The Robin core engine
• Agent orchestration logic
• Proprietary AI models or workflows
• Commercial SaaS infrastructure

Those components are part of Neeve’s proprietary systems and are distributed separately under commercial terms.

Contributions

By contributing to this repository, you agree that your contributions will be licensed under the Apache License, Version 2.0.

Support

• 📚 Documentation
• 🐛 Issue Tracker
• 💬 Discussions

Related Reading

• OpenTelemetry Documentation
• NATS Documentation
• Circuit Breaker Pattern