rcommerz-logger-python 1.0.1

Production-ready structured logging for Python/FastAPI microservices with OpenTelemetry support

rcommerz-logger-python

Production-ready structured logging package for Python/FastAPI microservices with OpenTelemetry support and LGTM stack compatibility.

Table of Contents

• Features
• Installation
• Quick Start
• Usage Examples
• API Reference
• Best Practices
• Testing
• Development
• Contributing
• CI/CD & Publishing
• License

Features

• ✅ Singleton Pattern - Initialize once, use everywhere
• ✅ Structured JSON Logging - Always outputs valid JSON
• ✅ OpenTelemetry Integration - Automatic trace_id and span_id extraction
• ✅ FastAPI Middleware - Automatic HTTP request/response logging
• ✅ Performance - Built on structlog (high-performance structured logging)
• ✅ Type-Safe - Full type hints support
• ✅ Container-Friendly - Outputs to stdout
• ✅ Production-Ready - Minimal dependencies, battle-tested

Quick Start

Installation

# Install from PyPI
pip install rcommerz-logger-python

# Install with FastAPI support
pip install rcommerz-logger-python[fastapi]

# Install for development
pip install -e ".[dev,fastapi]"

Basic Usage

from rcommerz_logger import Logger, LoggerConfig

# Step 1: Initialize logger once at application startup
Logger.initialize(LoggerConfig(
    service_name="product-service",
    service_version="1.2.0",
    env="production",
    level="INFO"  # DEBUG, INFO, WARN, ERROR
))

Usage Examples

Logging Methods

from rcommerz_logger import Logger

logger = Logger.get_instance()

# Basic logging
logger.info("Application started")
logger.error("Database connection failed", {"error": err})
logger.warn("Cache miss", {"cache_key": "user:123"})
logger.debug("Processing request", {"request_id": "abc-123"})

# With dynamic context
logger.info("Order created", {
    "order_id": "ORD-12345",
    "user_id": "USR-999",
    "total_amount": 199.99,
    "items_count": 3
})

# Security logging
logger.security("Failed login attempt", {
    "user_id": "USR-456",
    "ip": "192.168.1.100",
    "reason": "Invalid credentials"
})

# Audit logging
logger.audit("User permission changed", {
    "user_id": "USR-789",
    "admin_id": "ADM-001",
    "old_role": "user",
    "new_role": "admin"
})

# Error with exception
try:
    result = divide_by_zero()
except Exception as e:
    logger.error("Calculation failed", {"error": e})

FastAPI Integration

from fastapi import FastAPI
from rcommerz_logger import Logger, LoggerConfig, create_fastapi_middleware

app = FastAPI()

# Initialize logger
Logger.initialize(LoggerConfig(
    service_name="api-gateway",
    service_version="1.0.0",
    env="production",
    level="INFO"
))

# Add HTTP logging middleware
create_fastapi_middleware(
    app,
    exclude_paths=["/health", "/metrics"],
    include_headers=False,
    include_body=False  # Be careful with sensitive data
)

@app.get("/api/products")
async def get_products():
    logger = Logger.get_instance()
    logger.info("Fetching products", {"category": "electronics"})
    return {"products": []}

@app.post("/api/orders")
async def create_order(order: dict):
    logger = Logger.get_instance()
    logger.info("Creating order", {"order_id": order.get("id")})
    return {"status": "created"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Request Context with Authentication

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract user from token/session
        user_id = extract_user_id(request)
        
        # Store in request state for logger to pick up
        request.state.user_id = user_id
        
        response = await call_next(request)
        return response

app = FastAPI()
app.add_middleware(AuthMiddleware)
create_fastapi_middleware(app, exclude_paths=["/health"])

OpenTelemetry Trace Integration

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from rcommerz_logger import Logger, LoggerConfig

# Setup OpenTelemetry
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Initialize logger
Logger.initialize(LoggerConfig(
    service_name="payment-service",
    service_version="1.0.0",
    env="production"
))

logger = Logger.get_instance()

# Logger automatically includes trace_id and span_id
def process_payment(payment_id: str):
    with tracer.start_as_current_span("process-payment") as span:
        span.set_attribute("payment_id", payment_id)
        
        # Logs will include trace_id and span_id automatically
        logger.info("Processing payment", {
            "payment_id": payment_id,
            "amount": 199.99
        })
        
        # ... payment logic ...
        
        logger.info("Payment completed", {"payment_id": payment_id})

Example Output

Standard Log

{
  "@timestamp": "2026-02-22T10:30:45.123Z",
  "log.level": "INFO",
  "log_type": "normal",
  "service.name": "product-service",
  "service.version": "1.2.0",
  "env": "production",
  "host.name": "pod-product-abc123",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "message": "Order created",
  "order_id": "ORD-12345",
  "user_id": "USR-999",
  "total_amount": 199.99,
  "items_count": 3
}

HTTP Request Log

{
  "@timestamp": "2026-02-22T10:30:46.456Z",
  "log.level": "INFO",
  "log_type": "http",
  "service.name": "api-gateway",
  "service.version": "1.0.0",
  "env": "production",
  "host.name": "pod-gateway-xyz789",
  "trace_id": "3ad45f8b21c34e2a8d41ba6e3f9c0412",
  "span_id": "91f23ab4cd5e6789",
  "message": "GET /api/products 200",
  "method": "GET",
  "path": "/api/products",
  "status_code": 200,
  "duration_ms": 45.23,
  "client_ip": "10.0.1.25",
  "user_agent": "python-requests/2.31.0"
}

Error Log

{
  "@timestamp": "2026-02-22T10:30:47.789Z",
  "log.level": "ERROR",
  "log_type": "error",
  "service.name": "payment-service",
  "service.version": "2.1.0",
  "env": "production",
  "host.name": "pod-payment-def456",
  "trace_id": "7cd89e12f43b4a9f8e21dc7a5b6f3028",
  "span_id": "12a34b56c78d90ef",
  "message": "Payment processing failed",
  "error_message": "Insufficient funds",
  "error_type": "PaymentError",
  "payment_id": "PAY-67890",
  "user_id": "USR-111",
  "amount": 500.00
}

API Reference

Logger Methods

info(message: str, context: Optional[Dict[str, Any]] = None)

Log informational messages.

error(message: str, context: Optional[Dict[str, Any]] = None)

Log error messages. Automatically extracts exception details.

warn(message: str, context: Optional[Dict[str, Any]] = None)

Log warning messages.

debug(message: str, context: Optional[Dict[str, Any]] = None)

Log debug messages (only if level is DEBUG).

security(message: str, context: Optional[Dict[str, Any]] = None)

Log security-related events (log_type = "security").

audit(message: str, context: Optional[Dict[str, Any]] = None)

Log audit trail events (log_type = "audit").

http(message: str, context: Optional[Dict[str, Any]] = None)

Log HTTP-specific events (log_type = "http").

Middleware Function

create_fastapi_middleware(app, exclude_paths=None, include_headers=False, include_body=False)

• app: FastAPI application instance
• exclude_paths: List of paths to exclude (e.g., ["/health", "/metrics"])
• include_headers: Include request headers in logs (default: False)
• include_body: Include request body in logs (default: False, use carefully)

Best Practices

✅ DO

• Initialize logger once at application startup
• Use appropriate log levels (info, error, warn, debug)
• Include relevant context (user_id, order_id, etc.)
• Use logger.security() for authentication/authorization events
• Use logger.audit() for compliance-tracked actions
• Filter sensitive data before logging
• Use exclude_paths for health checks and metrics

❌ DON'T

• Log sensitive information (passwords, credit cards, tokens)
• Use debug level in production
• Initialize logger multiple times
• Use include_body=True without sanitizing
• Create multiple logger instances

Environment Variables

# Set via LoggerConfig.level
export LOG_LEVEL=INFO  # DEBUG, INFO, WARN, ERROR

LGTM Stack Compatibility

Designed for:

• Loki - Structured JSON with consistent labels
• Grafana - Standardized field names
• Tempo - Automatic trace correlation
• Prometheus - Metrics from log patterns

Example Loki Query

{service_name="product-service"} | json | log_type="http" | status_code >= 500

Type Hints

Full type hints support for better IDE experience:

from rcommerz_logger import Logger, LoggerConfig, LogContext
from typing import Dict, Any

context: LogContext = {"user_id": "123", "action": "login"}
logger.info("User logged in", context)

Testing

The package has 100% test coverage with 69 comprehensive tests.

# Install dev dependencies
pip install -e ".[dev]"

# Run all tests
pytest

# Run with coverage
pytest --cov=rcommerz_logger --cov-report=term-missing

# Run specific test file
pytest tests/test_logger.py -v

# Run with verbose output
pytest -v

Test Structure

• Unit Tests (test_logger.py): 26 tests for core logger functionality
• Middleware Tests (test_middleware.py): 12 tests for FastAPI integration
• Integration Tests (test_integration.py): 12 end-to-end tests
• Edge Cases (test_edge_cases.py): 7 tests for error handling
• OpenTelemetry (test_otel_coverage.py): 5 tests for trace extraction
• Convenience (test_convenience.py): 7 tests for helper functions

Development

Setup Development Environment

# Clone the repository
git clone https://github.com/rcommerz/logger-python.git
cd logger-python

# Create virtual environment
python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install in editable mode with dev dependencies
pip install -e ".[dev,fastapi]"

# Run tests
pytest

# Format code
black src/ tests/

# Type checking
mypy src/

Project Structure

rcommerz-logger-python/
├── src/                      # Source code (maps to rcommerz_logger)
│   ├── __init__.py          # Package exports
│   ├── logger.py            # Core Logger class
│   ├── middleware.py        # FastAPI middleware
│   └── types.py             # Type definitions
├── tests/                   # Test suite (100% coverage)
│   ├── test_logger.py
│   ├── test_middleware.py
│   ├── test_integration.py
│   ├── test_edge_cases.py
│   ├── test_otel_coverage.py
│   └── test_convenience.py
├── setup.py                 # Package configuration
├── pyproject.toml          # Build system configuration
├── requirements.txt        # Production dependencies
├── pytest.ini             # Pytest configuration
├── README.md              # This file
└── LICENSE                # MIT License

Contributing

Contributions are welcome! Please follow these guidelines:

How to Contribute

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Add tests for new functionality
5. Ensure tests pass (pytest)
6. Maintain 100% coverage (pytest --cov=rcommerz_logger)
7. Format code (black src/ tests/)
8. Commit changes (git commit -m 'Add amazing feature')
9. Push to branch (git push origin feature/amazing-feature)
10. Open a Pull Request

Code Standards

• Python 3.9+ compatibility
• Type hints for all functions
• 100% test coverage requirement
• Black code formatting
• Clear commit messages
• Documentation for new features

Running Quality Checks

# Run all tests
pytest -v

# Check coverage
pytest --cov=rcommerz_logger --cov-report=html
open htmlcov/index.html

# Format code
black src/ tests/

# Type checking
mypy src/

# Lint
flake8 src/ tests/

CI/CD & Publishing

This package uses GitHub Actions for automated testing and publishing.

Automated Workflows

1. Continuous Integration (test.yml)

Runs on every push and pull request:

• ✅ Tests across Python 3.9, 3.10, 3.11, 3.12
• ✅ Code quality checks (Black, flake8, mypy)
• ✅ Security scanning (safety, bandit)
• ✅ 100% coverage requirement

2. Automated Publishing (publish.yml)

Triggers on version tags (v*):

• ✅ Runs full test suite
• ✅ Builds package (wheel + source)
• ✅ Publishes to PyPI automatically
• ✅ Creates GitHub release with artifacts

3. Release Drafter (release-drafter.yml)

• ✅ Auto-generates release notes from PRs
• ✅ Categorizes changes (features, fixes, etc.)

4. Dependabot (dependabot.yml)

• ✅ Weekly dependency updates
• ✅ Automatic security patches

Publishing a New Version

Automated Release (Recommended)

# 1. Update version in three places:
#    - setup.py (line 8)
#    - pyproject.toml (line 6)
#    - src/__init__.py (line 9)

# 2. Update CHANGELOG.md

# 3. Commit and push
git add setup.py pyproject.toml src/__init__.py CHANGELOG.md
git commit -m "chore: Bump version to 1.1.0"
git push

# 4. Create and push tag
git tag -a v1.1.0 -m "Release version 1.1.0"
git push origin v1.1.0

# 5. GitHub Actions automatically:
#    ✓ Runs tests (69 tests must pass)
#    ✓ Builds package
#    ✓ Publishes to PyPI
#    ✓ Creates GitHub release

That's it! No manual publishing needed.

Manual Publishing (Fallback)

If you need to publish manually:

# Clean and build
rm -rf dist/
python -m build

# Check package
twine check dist/*

# Upload to PyPI
twine upload dist/*

Setup GitHub Actions

See GITHUB_ACTIONS_SETUP.md for detailed setup instructions:

1. Add PyPI token to GitHub Secrets as PYPI_API_TOKEN
2. Enable GitHub Actions in repository settings
3. Push a version tag to trigger automatic publishing

Monitoring Releases

• GitHub Actions: https://github.com/rcommerz/logger-python/actions
• PyPI Releases: https://pypi.org/project/rcommerz-logger-python/
• GitHub Releases: https://github.com/rcommerz/logger-python/releases

Manual Publishing Guide

For detailed manual publishing instructions, see:

• PUBLISHING.md - Complete manual publishing workflow
• QUICKSTART_PUBLISHING.md - Quick reference guide

Changelog

[1.0.0] - 2026-02-22

Added

• Initial production release
• Singleton Logger with structured JSON output
• FastAPI middleware for HTTP request logging
• OpenTelemetry trace context extraction
• Support for multiple log types (info, error, warn, debug, security, audit, http)
• Comprehensive test suite (69 tests, 100% coverage)
• Python 3.9+ support
• Full type hints
• LGTM stack compatibility

Performance

• Fast: Built on structlog (high-performance logging)
• Low overhead: <1ms per log entry
• Memory efficient: Minimal memory footprint
• Thread-safe: Safe for concurrent use
• Async-compatible: Works with FastAPI/asyncio

Security

• No sensitive data logging by default
• Configurable field exclusion
• Sanitization support
• Audit trail capabilities
• Security event tracking

Roadmap

• Structured log sampling for high-volume services
• Log rotation and archival support
• Custom formatters
• Additional integrations (Datadog, New Relic, etc.)
• CLI tool for log analysis
• Performance profiling integration

Support & Community

• 📖 Documentation: GitHub README
• 🐛 Bug Reports: GitHub Issues
• 💡 Feature Requests: GitHub Discussions
• 💬 Community: [Slack/Discord] (Coming soon)

Authors

RCOMMERZ DevOps Team

• Email: [e-mail]
• GitHub: @rcommerz

Acknowledgments

• Built with structlog
• Inspired by the Twelve-Factor App methodology
• Designed for the LGTM stack

License

MIT