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.
pip install mohflowfrom 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
}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
MohFlow includes a robust test suite with 401 tests organized into three categories:
tests/
βββ unit/ # 277 tests - Core functionality
βββ integration/ # 17 tests - End-to-end scenarios
βββ ui/ # 107 tests - Mohnitor UI & automation
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)- β 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
MohFlow follows Test-Driven Development (TDD) for all new features:
- Write failing test β Create test that validates expected behavior
- Implement minimal code β Write just enough code to make test pass
- Refactor β Improve code while keeping tests green
# 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"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-mockFeatures 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
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.
- π 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
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
- π§ͺ 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
pip install mohflowBasic 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)Mohflow can be configured in multiple ways:
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)
)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")MohFlow follows a clear configuration precedence order (highest to lowest priority):
- Runtime parameters (direct function arguments)
- Environment variables (prefixed with
MOHFLOW_) - JSON configuration file
- Default values
# Example showing precedence
logger = MohflowLogger(
config_file="config.json", # Base configuration
environment="staging", # Overrides config file
log_level="DEBUG" # Overrides environment variable
)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_NAMEEnable automatic environment detection and configuration:
# Auto-detects AWS, GCP, Azure, Kubernetes, Docker, etc.
logger = MohflowLogger(
service_name="my-app",
enable_auto_config=True
)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()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]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_idtransaction_id,session_id,operation_idparent_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)
)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'}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 fieldsConvenient 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
)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")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)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"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 servicesDeploy 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"
)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
)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"}# 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.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)# 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
)Logs are output in enriched JSON format for comprehensive observability:
{
"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"
}{
"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"
}{
"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"
}MohFlow includes pre-built dashboard templates for instant log visualization:
- 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
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"
)# Clone the repository
git clone https://github.com/parijatmukherjee/mohflow.git
cd mohflow
# Install development dependencies
make install# 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 packageMohFlow 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
# 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.jsonContributions are welcome! Please feel free to submit a Pull Request.
Before submitting a pull request, please ensure:
- All tests pass:
make test - Code is formatted:
make format - Linting passes:
make lintβ (zero violations required) - Type hints are used where appropriate
- Documentation is updated for new features
We maintain 100% lint compliance - your code should pass make lint without any violations.
This project is licensed under the MIT License - see the LICENSE file for details.