Skip to content

Repository files navigation

MohFlow β€” Structured JSON Logging for Python (Loki, Grafana, Console, File)

CI PyPI version Python Support Code Quality Lint Compliance

Python Versions: 3.8, 3.9, 3.10, 3.11+

MohFlow is a Python structured logging library (JSON-first) that targets console, files, and aggregators (e.g., Loki). It's designed to be easy to use while providing powerful logging capabilities for modern Python applications.

πŸš€ Quickstart

Installation

pip install mohflow

Basic Usage

from mohflow import MohflowLogger

# Create a logger instance
logger = MohflowLogger(service_name="my-app")

# Log structured data
logger.info("User action completed", user_id="123", action="login", success=True)

Output:

{
  "timestamp": "2025-09-18T10:30:00.123456+00:00",
  "level": "INFO",
  "service": "my-app",
  "message": "User action completed",
  "user_id": "123",
  "action": "login",
  "success": true
}

πŸ”’ Quality Gates

Before submitting any changes, ensure all quality gates pass locally:

make format    # Format code with black (must pass with zero errors)
make lint      # Lint code with flake8 (must pass with zero errors)
make test      # Run test suite with pytest (must pass all tests)

Requirements:

  • All commands must execute with zero errors before PR submission
  • GitHub Actions CI enforces the same standards
  • Quality gates are non-negotiable for code acceptance

πŸ§ͺ Comprehensive Test Suite

MohFlow includes a robust test suite with 401 tests organized into three categories:

Test Structure

tests/
β”œβ”€β”€ unit/           # 277 tests - Core functionality
β”œβ”€β”€ integration/    # 17 tests - End-to-end scenarios
└── ui/            # 107 tests - Mohnitor UI & automation

Test Commands

make test              # Run all 401 tests with coverage
make test-unit         # Run 277 unit tests (fast)
make test-integration  # Run 17 integration tests
make test-ui           # Run 107 UI tests (includes automation)

Test Results

  • βœ… 339 tests pass - All core functionality verified
  • ⏭️ 62 tests skip - Graceful handling of optional dependencies (FastAPI, Selenium)
  • 🎯 47% coverage - Comprehensive test coverage across all modules

TDD Workflow

MohFlow follows Test-Driven Development (TDD) for all new features:

3-Step Process:

  1. Write failing test β†’ Create test that validates expected behavior
  2. Implement minimal code β†’ Write just enough code to make test pass
  3. Refactor β†’ Improve code while keeping tests green

Example TDD Cycle:

# tests/test_example.py
import pytest
from mohflow import MohflowLogger

def test_logger_creates_structured_output():
    """Test that logger produces JSON-structured output."""
    # 1. Write failing test (RED)
    logger = MohflowLogger(service_name="test")

    # 2. Implement feature to make test pass (GREEN)
    # Logger should be created successfully and not raise errors
    logger.info("test message", user_id="123")

    # Assert that the logger's service name is set correctly
    assert logger.config.SERVICE_NAME == "test"

Browser Automation Testing

MohFlow includes comprehensive UI automation tests for the Mohnitor log viewer:

# Test UI functionality with Selenium (requires Chrome)
make test-automation

# Mock tests (no external dependencies)
make test-automation-mock

Features tested:

  • Real-time log streaming via WebSocket
  • Log filtering and search functionality
  • Multi-service discovery and display
  • Error log visualization with proper styling
  • High-volume log handling and performance

πŸ“‹ Spec-Kit Workflow

For structured feature development, use the spec-kit process:

  • /specify β†’ Create feature specification with requirements and user stories
  • /plan β†’ Generate implementation plan with technical design and tasks
  • /tasks β†’ Create numbered, actionable tasks for implementation

Example:

# Conceptual workflow steps (not shell commands):
/specify "Add user authentication to logging context"
/plan "Based on specification, create technical design"
/tasks "Generate specific implementation steps"

New specifications go in the specs/ directory.

Features

  • πŸ“‹ Structured JSON logging for better log parsing
  • πŸš€ Simple setup with sensible defaults
  • πŸ”„ Built-in Grafana Loki integration
  • πŸ“ File logging support
  • 🌍 Environment-based configuration
  • πŸ” Rich context logging
  • ⚑ Lightweight and performant
  • πŸ€– Auto-configuration based on environment detection
  • πŸ“Š Pre-built dashboard templates for Grafana and Kibana
  • πŸ”’ Enhanced context awareness with request correlation
  • πŸ›‘οΈ Built-in security with sensitive data filtering
  • βš™οΈ JSON configuration support with schema validation
  • πŸ–₯️ CLI interface for dynamic debugging and management
  • πŸ”— Request correlation for distributed tracing

πŸ† Code Quality & Production Readiness

MohFlow is built with enterprise-grade code quality standards:

  • βœ… 100% Lint Compliance - Zero flake8 violations across entire codebase
  • βœ… Type Safety - Full type hints with mypy compatibility
  • βœ… Security Focused - Built-in PII detection and sensitive data filtering
  • βœ… Performance Optimized - Async handlers and high-throughput batching
  • βœ… Framework Integration - Intelligent auto-detection for Django, FastAPI, Flask, and more
  • βœ… Production Tested - Comprehensive test coverage with real-world scenarios
  • βœ… Clean Architecture - Modular design with clear separation of concerns

Recent Improvements (v1.1.2+)

  • πŸ§ͺ Comprehensive Test Suite: 401 tests organized into unit, integration, and UI automation tests
  • πŸ”§ GitHub Workflow Ready: All CI/CD checks pass - formatting, linting, and testing
  • πŸš€ Browser Automation: Full Selenium-based UI testing for Mohnitor log viewer
  • πŸ” Enhanced Security: Tracing field exemptions preserve distributed tracing context
  • 🎯 Test Organization: Clean test hierarchy with graceful dependency handling
  • πŸ› οΈ Developer Experience: Improved error messages and comprehensive testing framework

Installation

pip install mohflow

Quick Start

Basic usage with console logging:

from mohflow import MohflowLogger

# Initialize logger with minimal configuration
logger = MohflowLogger(service_name="my-app")

# Log messages
logger.info("Application started")
logger.error("An error occurred", error_code=500)

Configuration

Mohflow can be configured in multiple ways:

Basic Configuration

logger = MohflowLogger(
    service_name="my-app",                                    # Required
    environment="production",                                 # Optional (default: "development")
    loki_url="http://localhost:3100/loki/api/v1/push",       # Optional (default: None)
    log_level="INFO",                                        # Optional (default: "INFO")
    console_logging=True,                                    # Optional (default: True)
    file_logging=False,                                      # Optional (default: False)
    log_file_path="logs/app.log",                           # Required if file_logging=True
    enable_auto_config=False,                               # Optional (default: False)
    enable_context_enrichment=True,                         # Optional (default: True)
    enable_sensitive_data_filter=True                       # Optional (default: True)
)

JSON Configuration

Create a mohflow_config.json file for advanced configuration:

{
  "service_name": "my-app",
  "environment": "production",
  "log_level": "INFO",
  "console_logging": true,
  "file_logging": true,
  "log_file_path": "logs/app.log",
  "loki_url": "http://localhost:3100/loki/api/v1/push",
  "context_enrichment": {
    "include_timestamp": true,
    "include_system_info": true,
    "include_request_context": true
  },
  "sensitive_data_filter": {
    "enabled": true,
    "redaction_text": "[REDACTED]",
    "patterns": ["password", "token", "secret"]
  }
}

Use the JSON configuration:

logger = MohflowLogger(config_file="mohflow_config.json")

Configuration Precedence

MohFlow follows a clear configuration precedence order (highest to lowest priority):

  1. Runtime parameters (direct function arguments)
  2. Environment variables (prefixed with MOHFLOW_)
  3. JSON configuration file
  4. Default values
# Example showing precedence
logger = MohflowLogger(
    config_file="config.json",        # Base configuration
    environment="staging",            # Overrides config file
    log_level="DEBUG"                 # Overrides environment variable
)

Environment Variables

Configure MohFlow using environment variables:

export MOHFLOW_SERVICE_NAME="my-app"
export MOHFLOW_LOG_LEVEL="INFO"
export MOHFLOW_ENVIRONMENT="production"
export MOHFLOW_CONSOLE_LOGGING="true"
export MOHFLOW_LOKI_URL="http://loki:3100/loki/api/v1/push"

# For nested configurations
export MOHFLOW_CONTEXT_ENRICHMENT_INCLUDE_TIMESTAMP="true"
export MOHFLOW_SENSITIVE_DATA_FILTER_ENABLED="true"
# Will automatically pick up environment variables
logger = MohflowLogger()  # service_name from MOHFLOW_SERVICE_NAME

Auto-Configuration

Enable automatic environment detection and configuration:

# Auto-detects AWS, GCP, Azure, Kubernetes, Docker, etc.
logger = MohflowLogger(
    service_name="my-app",
    enable_auto_config=True
)

Enhanced Features

Thread-Safe Context Management

MohFlow provides thread-safe context management for microservices and async applications:

from mohflow.context.enrichment import RequestContextManager
from mohflow.context.correlation import get_correlation_id, set_correlation_id
import threading

def handle_request(request_id, user_id):
    # Each thread gets its own context
    with RequestContextManager(request_id=request_id, user_id=user_id):
        logger.info("Processing request")
        
        # Correlation ID is automatically generated and thread-local
        correlation_id = get_correlation_id()
        logger.info("Generated correlation", correlation_id=correlation_id)

# Multiple threads with independent contexts
for i in range(3):
    thread = threading.Thread(target=handle_request, args=(f"req-{i}", f"user-{i}"))
    thread.start()

Sensitive Data Protection

Automatic detection and redaction of sensitive information:

# Built-in patterns detect and redact sensitive data
logger.info("User registration", 
    username="john_doe",
    password="secret123",           # [REDACTED]
    email="john@example.com",       # [REDACTED] 
    credit_card="4111-1111-1111-1111",  # [REDACTED]
    api_key="sk-abc123def456"       # [REDACTED]
)

# Add custom sensitive patterns
logger.add_sensitive_field("internal_id")
logger.info("Processing", internal_id="12345")  # [REDACTED]

Tracing Field Exemptions

MohFlow v1.1.1+ automatically preserves distributed tracing fields while redacting sensitive data:

# Tracing fields are preserved during sensitive data filtering
logger.info("Processing payment request",
    correlation_id="req-abc-123",      # βœ… Preserved (tracing)
    request_id="req-456",              # βœ… Preserved (tracing)
    trace_id="trace-789",              # βœ… Preserved (tracing)
    span_id="span-101112",             # βœ… Preserved (tracing)
    user_id="user-789",                # βœ… Untouched (neutral)
    api_key="sk-secret123",            # ❌ [REDACTED] (sensitive)
    credit_card="4111-1111-1111-1111"  # ❌ [REDACTED] (sensitive)
)

Output with tracing preservation:

{
  "timestamp": "2025-09-18T10:30:00.123456+00:00",
  "level": "INFO",
  "service": "payment-service",
  "message": "Processing payment request",
  "correlation_id": "req-abc-123",
  "request_id": "req-456",
  "trace_id": "trace-789",
  "span_id": "span-101112",
  "user_id": "user-789",
  "api_key": "[REDACTED]",
  "credit_card": "[REDACTED]"
}

Default preserved tracing fields:

  • correlation_id, request_id, trace_id, span_id
  • transaction_id, session_id, operation_id
  • parent_id, root_id, trace_context
  • Pattern-based: trace_*, span_*

Custom tracing patterns:

from mohflow.context.filters import SensitiveDataFilter

# Add custom tracing field patterns
filter_obj = SensitiveDataFilter(
    exclude_tracing_fields=True,
    tracing_field_patterns=[r"^x_trace_.*", r".*_correlation_.*"]
)

# Custom patterns will preserve fields like:
# x_trace_id, request_correlation_key, etc.
logger.info("Custom tracing",
    x_trace_custom="trace-123",        # βœ… Preserved (custom pattern)
    req_correlation_key="corr-456",    # βœ… Preserved (custom pattern)
    password="secret"                  # ❌ [REDACTED] (sensitive)
)

Auto-Configuration with Environment Detection

Automatically configure logging based on your deployment environment:

# Detects AWS, GCP, Azure, Kubernetes, Docker, etc.
logger = MohflowLogger(
    service_name="my-app",
    enable_auto_config=True  # Automatically configures based on environment
)

# Get detected environment information
env_info = logger.get_environment_info()
print(f"Running on: {env_info}")
# Output: {'cloud_provider': 'aws', 'region': 'us-east-1', 'environment_type': 'production'}

Custom Context Enrichers

Add custom metadata to all log messages:

import os

# Add custom enrichers
logger.add_custom_enricher("version", lambda: os.getenv("APP_VERSION", "unknown"))
logger.add_custom_enricher("build", lambda: os.getenv("BUILD_NUMBER", "dev"))

# All logs will now include version and build information
logger.info("Application started")  # Includes version and build fields

Factory Methods for Common Use Cases

Convenient factory methods for quick setup:

# Create logger with auto-configuration
logger = MohflowLogger.with_auto_config(
    service_name="my-app"
)

# Create logger from configuration file
logger = MohflowLogger.from_config_file(
    "config.json",
    log_level="DEBUG"  # Override specific settings
)

Practical Usage Examples

Complete Microservice Setup

from mohflow import MohflowLogger
from mohflow.context.enrichment import RequestContextManager
from mohflow.context.correlation import get_correlation_id
import os
import uuid

# Initialize with full feature set
logger = MohflowLogger(
    service_name="payment-service",
    environment=os.getenv("ENVIRONMENT", "development"),
    enable_auto_config=True,           # Auto-detect cloud environment
    enable_context_enrichment=True,    # Add system metadata
    enable_sensitive_data_filter=True, # Protect sensitive data
    loki_url=os.getenv("LOKI_URL"),    # Optional Loki integration
)

def process_payment(user_id: str, amount: float, card_number: str):
    """Process payment with full observability"""
    request_id = str(uuid.uuid4())
    
    with RequestContextManager(
        request_id=request_id, 
        user_id=user_id,
        operation_name="process_payment"
    ):
        logger.info("Payment processing started", 
            amount=amount,
            currency="USD"
        )
        
        try:
            # Sensitive data is automatically redacted
            logger.debug("Payment details", 
                card_number=card_number,  # [REDACTED]
                amount=amount
            )
            
            # Simulate payment processing
            if amount > 0:
                # Get correlation ID for external service calls
                correlation_id = get_correlation_id()
                
                # Call external payment gateway
                # headers = {"X-Correlation-ID": correlation_id}
                
                logger.info("Payment processed successfully", 
                    transaction_id=f"txn_{request_id}",
                    status="completed"
                )
                return {"status": "success", "transaction_id": f"txn_{request_id}"}
            else:
                raise ValueError("Invalid amount")
                
        except Exception as e:
            logger.error("Payment processing failed", 
                error=str(e),
                error_type=type(e).__name__,
                exc_info=True
            )
            raise

# Usage
result = process_payment("user_123", 99.99, "4111-1111-1111-1111")

Flask Application with Request Tracking

from flask import Flask, request, g
from mohflow import MohflowLogger
from mohflow.context.enrichment import RequestContextManager
import uuid
import time

app = Flask(__name__)

# Initialize logger with auto-configuration
logger = MohflowLogger(
    service_name="flask-api",
    enable_auto_config=True,
    enable_context_enrichment=True,
    enable_sensitive_data_filter=True
)

@app.before_request
def before_request():
    """Set up request context for each request"""
    g.request_id = str(uuid.uuid4())
    g.start_time = time.time()

@app.after_request
def after_request(response):
    """Log request completion"""
    duration = time.time() - g.start_time
    logger.info("Request completed",
        method=request.method,
        path=request.path,
        status_code=response.status_code,
        duration_ms=round(duration * 1000, 2),
        user_agent=request.headers.get('User-Agent')
    )
    return response

@app.route('/api/users/<user_id>')
def get_user(user_id):
    with RequestContextManager(
        request_id=g.request_id,
        user_id=user_id,
        operation_name="get_user"
    ):
        logger.info("Fetching user data")
        
        # Simulate database query
        user_data = {"id": user_id, "name": "John Doe"}
        
        logger.info("User data retrieved", user_found=True)
        return user_data

if __name__ == '__main__':
    app.run(debug=True)

Advanced Features

CLI Interface

MohFlow includes a powerful CLI for debugging and management:

# Basic usage
python -m mohflow.cli --service-name "my-app" --log-level DEBUG

# Validate configuration
python -m mohflow.cli --validate-config --config-file config.json

# Interactive debugging session
python -m mohflow.cli --interactive --service-name "my-app"

# Test logging functionality
python -m mohflow.cli --test --service-name "my-app" --loki-url "http://localhost:3100"

Context Enrichment and Request Correlation

Automatically enrich logs with system metadata and request correlation:

from mohflow.context.enrichment import RequestContextManager
from mohflow.context.correlation import get_correlation_id

# Set request context for distributed tracing
with RequestContextManager(request_id="req-123", user_id="user-456"):
    logger.info("Processing request")  # Automatically includes request context
    
    # Get correlation ID for external service calls
    correlation_id = get_correlation_id()
    # Pass correlation_id to external services

Dashboard Templates

Deploy pre-built dashboards for instant log visualization:

from mohflow.templates import deploy_grafana_dashboard, deploy_kibana_dashboard

# Deploy Grafana dashboard
deploy_grafana_dashboard(
    template_name="application_logs",
    grafana_url="http://localhost:3000",
    api_key="your-api-key"
)

# Deploy Kibana dashboard
deploy_kibana_dashboard(
    template_name="error_tracking",
    kibana_url="http://localhost:5601"
)

Security Features

Built-in sensitive data filtering:

# Sensitive data is automatically redacted
logger.info("User login", password="secret123", token="abc123")
# Output: {"message": "User login", "password": "[REDACTED]", "token": "[REDACTED]"}

# Customize sensitive patterns
logger = MohflowLogger(
    service_name="my-app",
    enable_sensitive_data_filter=True
)

Examples

FastAPI Integration with Enhanced Features

from fastapi import FastAPI, Request
from mohflow import MohflowLogger
from mohflow.context.enrichment import RequestContextManager
import uuid

app = FastAPI()

# Initialize with auto-configuration and enhanced features
logger = MohflowLogger(
    service_name="fastapi-app",
    environment="production",
    enable_auto_config=True,  # Auto-detect cloud environment
    enable_context_enrichment=True,
    enable_sensitive_data_filter=True
)

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    
    # Set request context for correlation
    with RequestContextManager(request_id=request_id, path=str(request.url.path)):
        logger.info("Request started", method=request.method)
        response = await call_next(request)
        logger.info("Request completed", status_code=response.status_code)
        return response

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

@app.post("/login")
async def login(username: str, password: str):
    # Password automatically redacted in logs
    logger.info("Login attempt", username=username, password=password)
    return {"status": "success"}

Cloud-Native Deployment

# Auto-configure for cloud environments (AWS, GCP, Azure, K8s)
logger = MohflowLogger(
    service_name="cloud-app",
    enable_auto_config=True,  # Detects cloud provider automatically
    config_file="config.json"  # Load additional config from file
)

# Enhanced logging with automatic context enrichment
logger.info("Service started")  # Includes hostname, process_id, thread_id, etc.

Microservices with Request Correlation

import requests
from mohflow import MohflowLogger
from mohflow.context.enrichment import RequestContextManager, get_correlation_id

logger = MohflowLogger(service_name="user-service", enable_auto_config=True)

def process_user_request(user_id: str):
    with RequestContextManager(request_id=f"user-{user_id}", user_id=user_id):
        logger.info("Processing user request")
        
        # Get correlation ID for downstream services
        correlation_id = get_correlation_id()
        
        # Call another service with correlation
        response = requests.post(
            "http://payment-service/process",
            headers={"X-Correlation-ID": correlation_id},
            json={"user_id": user_id}
        )
        
        logger.info("Payment processed", payment_status=response.status_code)

Configuration Management

# Use JSON configuration for complex setups
logger = MohflowLogger(config_file="production_config.json")

# Override specific settings at runtime
logger = MohflowLogger(
    config_file="base_config.json",
    environment="staging",  # Override environment
    log_level="DEBUG"       # Override log level
)

Log Output Format

Logs are output in enriched JSON format for comprehensive observability:

Basic Log Format

{
    "timestamp": "2025-09-11T18:30:00.123456+00:00",
    "level": "INFO",
    "service_name": "my-app",
    "message": "User logged in",
    "environment": "production",
    "user_id": 123,
    "process_id": 12345,
    "thread_id": 67890,
    "hostname": "app-server-01"
}

Enhanced Log with Request Context

{
    "timestamp": "2025-09-11T18:30:00.123456+00:00",
    "level": "INFO",
    "service_name": "user-service",
    "message": "Processing payment",
    "environment": "production",
    "request_id": "req-uuid-123",
    "correlation_id": "corr-uuid-456",
    "user_id": "user-789",
    "process_id": 12345,
    "thread_id": 67890,
    "hostname": "k8s-pod-abc123",
    "cloud_provider": "aws",
    "region": "us-east-1"
}

Security-Filtered Log

{
    "timestamp": "2025-09-11T18:30:00.123456+00:00",
    "level": "INFO",
    "service_name": "auth-service",
    "message": "Login attempt",
    "username": "john_doe",
    "password": "[REDACTED]",
    "api_key": "[REDACTED]",
    "ip_address": "192.168.1.100"
}

Dashboard Templates

MohFlow includes pre-built dashboard templates for instant log visualization:

Available Templates

  • application_logs: General application logging dashboard
  • error_tracking: Error monitoring and alerting dashboard
  • performance_metrics: Performance and latency tracking
  • security_audit: Security events and audit trail
  • request_correlation: Distributed tracing visualization

Quick Dashboard Deployment

from mohflow.templates import list_available_templates, deploy_grafana_dashboard

# List all available templates
templates = list_available_templates()
print(f"Available templates: {templates}")

# Deploy to Grafana
deploy_grafana_dashboard(
    template_name="application_logs",
    grafana_url="http://localhost:3000",
    api_key="your-grafana-api-key",
    datasource_name="Loki"
)

Development

Setup

# Clone the repository
git clone https://github.com/parijatmukherjee/mohflow.git
cd mohflow

# Install development dependencies
make install

Running Tests

# Run complete test suite (401 tests)
make test

# Run test categories individually
make test-unit         # 277 unit tests (fast feedback)
make test-integration  # 17 integration tests (scenarios)
make test-ui           # 107 UI tests (includes automation)

# Run automation tests specifically
make test-automation      # Full browser automation (requires Chrome)
make test-automation-mock # Mock tests (no dependencies)

# Code quality
make format            # Format code with black
make lint             # Lint with flake8 (100% compliant)
make build            # Build package

Code Quality Standards

MohFlow maintains strict code quality standards:

# All code passes flake8 linting with zero violations
make lint  # βœ… 0 violations

# Type checking (when available)
mypy src/mohflow/

# Security scanning
bandit -r src/mohflow/

Recent Quality Improvements:

  • 🎯 Resolved all 142 initial lint violations
  • πŸ”§ Fixed syntax errors and undefined variables
  • πŸ“ Enforced 79-character line limits across entire codebase
  • 🧹 Removed unused imports and variables
  • πŸ—οΈ Refactored complex functions for better maintainability
  • πŸ” Enhanced f-string usage and blank line formatting

CLI Development and Testing

# Test CLI functionality
python -m mohflow.cli --help

# Run interactive debugging session
python -m mohflow.cli --interactive --service-name "dev-app"

# Validate configuration files
python -m mohflow.cli --validate-config --config-file examples/config.json

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Code Quality Requirements

Before submitting a pull request, please ensure:

  1. All tests pass: make test
  2. Code is formatted: make format
  3. Linting passes: make lint βœ… (zero violations required)
  4. Type hints are used where appropriate
  5. Documentation is updated for new features

We maintain 100% lint compliance - your code should pass make lint without any violations.

License

This project is licensed under the MIT License - see the LICENSE file for details.

About

Python structured logging library with JSON output, Loki integration, and console/file handlers. Zero-config observability for modern Python apps.

Topics

Resources

Contributing

Stars

4 stars

Watchers

1 watching

Forks

Releases

Packages

Used by

Contributors

Languages