ganicas-package 0.3.0

Ganicas internal Python package for structured logging and utilities.

Ganicas Utils

Ganicas Utils is an internal Python package providing structured logging utilities and middleware for Flask and FastAPI applications. Built on top of structlog, it enables production-ready, context-aware logging with minimal configuration.

---

📋 Table of Contents

• Features
• Installation
• Quick Start
• Structured Logging
  • Basic Configuration
  • Production Configuration
• Middleware
  • Flask Middleware
  • FastAPI Middleware
  • Request Logging Middleware
• Why Structured Logging?
• Development
• License

---

✨ Features

• Structured Logging: JSON-formatted logs for easy parsing by log aggregation tools (ELK, Datadog, Grafana Loki)
• Context Management: Automatic request context binding (request_id, IP, user_agent, etc.)
• Flask & FastAPI Support: Ready-to-use middleware for both frameworks
• Advanced Request Logging: Comprehensive ASGI middleware with:
  • Automatic request/response logging
  • Sensitive header sanitization
  • Slow request detection
  • Sampling for high-traffic endpoints
  • Exception tracking with full context
  • Distributed tracing support (traceparent header)
• Production Ready: Battle-tested with 99% code coverage

---

📦 Installation

pip install ganicas-package

Or with Poetry:

poetry add ganicas-package

---

🚀 Quick Start

Basic Configuration

Replace logger = logging.getLogger(__name__) with logger = structlog.get_logger(__name__):

from ganicas_utils.logging import LoggingConfigurator
from ganicas_utils.config import Config
import structlog

config = Config()

LoggingConfigurator(
    service_name=config.APP_NAME,
    log_level='INFO',
    setup_logging_dict=True
).configure_structlog(
    formatter='plain_console',
    formatter_std_lib='plain_console'
)

logger = structlog.get_logger(__name__)
logger.info("Application started", version="1.0.0", environment="production")

---

📊 Structured Logging

Production Configuration

For production environments, use JSON formatting for machine-readable logs:

from ganicas_utils.logging import LoggingConfigurator
from ganicas_utils.config import Config
import structlog

config = Config()

LoggingConfigurator(
    service_name=config.APP_NAME,
    log_level='INFO',
    setup_logging_dict=True
).configure_structlog(
    formatter='json_formatter',
    formatter_std_lib='json_formatter'
)

logger = structlog.get_logger(__name__)
logger.info("User login", user_id=12345, ip_address="192.168.1.1")
logger.warning("High memory usage", memory_percent=85.5, threshold=80)
logger.error("Database connection failed", db_host="localhost", error_code="CONN_REFUSED")

try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception("Division by zero error", operation="calculate_ratio")

---

🔧 Middleware

Flask Middleware

The FlaskRequestContextMiddleware automatically adds request context to all logs:

import uuid
from flask import Flask
from ganicas_utils.logging import LoggingConfigurator
from ganicas_utils.logging.middlewares import FlaskRequestContextMiddleware
from ganicas_utils.config import Config
import structlog

config = Config()

LoggingConfigurator(
    service_name=config.APP_NAME,
    log_level="INFO",
    setup_logging_dict=True,
).configure_structlog(formatter='json_formatter', formatter_std_lib='json_formatter')

logger = structlog.get_logger(__name__)

app = Flask(__name__)
app.wsgi_app = FlaskRequestContextMiddleware(app.wsgi_app)

@app.route("/")
def home():
    logger.info("Processing request")  # Automatically includes request_id, method, path
    return "Hello, World!"

if __name__ == "__main__":
    app.run()

Automatic context injection:

• request_id - Unique identifier for each request
• request_method - HTTP method (GET, POST, etc.)
• request_path - Request URL path

---

FastAPI Middleware

Basic Context Middleware

For simple request context binding, use FastAPIRequestContextMiddleware:

from fastapi import FastAPI
from ganicas_utils.logging import LoggingConfigurator
from ganicas_utils.logging.middlewares import FastAPIRequestContextMiddleware
from ganicas_utils.config import Config
import structlog

config = Config()

LoggingConfigurator(
    service_name=config.APP_NAME,
    log_level="INFO",
    setup_logging_dict=True,
).configure_structlog(formatter='json_formatter', formatter_std_lib='json_formatter')

logger = structlog.get_logger(__name__)
app = FastAPI()
app.add_middleware(FastAPIRequestContextMiddleware)

@app.get("/")
async def root():
    logger.info("Processing request")  # Automatically includes request context
    return {"message": "Hello World"}

---

Request Logging Middleware

For production-grade request/response logging with advanced features, use RequestLoggingMiddleware:

from fastapi import FastAPI
from ganicas_utils.logging import LoggingConfigurator
from ganicas_utils.logging.middlewares import RequestLoggingMiddleware
import structlog

LoggingConfigurator(
    service_name="my-api",
    log_level="INFO",
    setup_logging_dict=True,
).configure_structlog(formatter='json_formatter', formatter_std_lib='json_formatter')

app = FastAPI()

# Add comprehensive request logging
app.add_middleware(
    RequestLoggingMiddleware,
    slow_request_threshold_ms=1000,      # Warn on requests > 1s
    propagate_request_id=True,           # Add request_id to response headers
    skip_paths={"/healthz", "/metrics"}, # Don't log health checks
    sample_2xx_rate=0.1,                 # Sample 10% of successful requests
)

@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "John Doe"}

Features

Automatic Logging:

• request.start - Logs when request begins
• request.end - Logs when request completes (with status, duration, size)
• request.exception - Logs unhandled exceptions with full traceback

Logged Information:

• Request: method, path, query params, client IP, user agent, content type/length
• Response: status code, size, content type, duration
• Headers: Sanitized request/response headers (for 4xx/5xx errors)
• Performance: Request duration, slow request detection

Security:

• Automatic sanitization of sensitive headers (Authorization, Cookie, X-API-Key)
• Authorization header preserves scheme: Bearer *** instead of exposing tokens
• No request/response body logging (only sizes)

Performance Optimization:

• Skip logging for health checks and metrics endpoints
• Sample successful requests to reduce log volume
• Skip OPTIONS requests
• Configurable path prefixes to skip

Distributed Tracing:

• Supports W3C traceparent header
• Falls back to x-request-id or x-amzn-trace-id
• Propagates request_id to response headers

Configuration Options

Parameter	Type	Default	Description
logger	structlog.BoundLoggerBase	structlog.get_logger("http")	Custom logger instance
slow_request_threshold_ms	int	None	Threshold in ms to flag slow requests
propagate_request_id	bool	True	Add x-request-id to response headers
skip_paths	set[str]	{"/healthz", "/metrics"}	Exact paths to skip logging
skip_prefixes	tuple[str, ...]	("/metrics",)	Path prefixes to skip logging
sample_2xx_rate	float	None	Sample rate for 2xx/3xx responses (0.0-1.0)

Example Logs

Successful Request:

{
  "event": "request.end",
  "request_id": "550e8400-e29b-41d4-a716-446655440000",
  "method": "GET",
  "path": "/api/users/123",
  "status_code": 200,
  "duration_ms": 45,
  "response_size": 256,
  "client_ip": "192.168.1.100",
  "user_agent": "Mozilla/5.0...",
  "level": "info"
}

Slow Request Warning:

{
  "event": "request.end",
  "request_id": "550e8400-e29b-41d4-a716-446655440001",
  "method": "POST",
  "path": "/api/process",
  "status_code": 200,
  "duration_ms": 1523,
  "slow_request": true,
  "slow_threshold_ms": 1000,
  "level": "warning"
}

Error with Sanitized Headers:

{
  "event": "request.end",
  "request_id": "550e8400-e29b-41d4-a716-446655440002",
  "method": "POST",
  "path": "/api/login",
  "status_code": 401,
  "duration_ms": 12,
  "request_headers": {
    "authorization": "Bearer ***",
    "content-type": "application/json"
  },
  "level": "warning"
}

---

🎯 Why Structured Logging?

Traditional logging challenges:

• Plain text logs are hard to parse programmatically
• Difficult to filter and search in log aggregation tools
• Missing context makes debugging distributed systems challenging

Structured logging benefits:

• Machine-readable: JSON format for easy parsing by ELK, Datadog, Grafana Loki
• Rich context: Automatic correlation with request_id, user_id, transaction_id
• Better filtering: Query logs by any field (status_code, duration, user_id, etc.)
• Observability: Enhanced monitoring and alerting capabilities
• Debugging: Trace requests across microservices with distributed tracing support

This package uses structlog - a powerful library that enhances Python's standard logging with better context management and flexible log formatting.

---

🛠️ Development

Prerequisites

Install Poetry for dependency management:

curl -sSL https://install.python-poetry.org | python3 -

Setup

# Install dependencies
poetry install --with dev

# Run tests with coverage
poetry run pytest -v --cov=ganicas_utils

# Run tests with detailed output
poetry run pytest -rs --cov=ganicas_utils -s

# Run pre-commit hooks
poetry run pre-commit run --all-files

Running Tests

# Run all tests
poetry run pytest

# Run specific test file
poetry run pytest tests/test_request_logging_middleware.py

# Run with coverage report
poetry run pytest --cov=ganicas_utils --cov-report=html

Code Quality

This project uses:

• pytest for testing (99% coverage)
• ruff for linting and formatting
• pre-commit for automated checks

---

📄 License

Proprietary - Internal use only for Ganicas projects.

---

🤝 Contributing

This is an internal package. For questions or contributions, please contact the Ganicas development team.

---

📚 Additional Resources

• structlog Documentation
• FastAPI Middleware Guide
• Flask Middleware Guide

---

Made with ❤️ by Ganicas Team