📘 Logging Best Practices: Production Apps

🎯 Introduction

Welcome to this essential tutorial on logging best practices for production Python applications! 🎉 If you’ve ever struggled to debug issues in production or wondered why your app crashed at 3 AM, this guide is for you!

Logging is like having a flight recorder for your application ✈️. It captures what happened, when it happened, and why things went wrong (or right!). Whether you’re building web APIs 🌐, data pipelines 📊, or automation scripts 🤖, mastering production logging will save you countless hours of debugging and help you sleep better at night! 😴

By the end of this tutorial, you’ll know how to implement professional-grade logging that will make debugging a breeze and keep your production apps running smoothly! Let’s dive in! 🏊‍♂️

📚 Understanding Production Logging

🤔 What is Production Logging?

Production logging is like having security cameras 📹 throughout your application. Just as cameras record what happens in a building, logs record what happens in your code!

In Python terms, production logging means:

• ✨ Capturing important events without impacting performance
• 🚀 Providing enough detail to debug issues
• 🛡️ Protecting sensitive information
• 📊 Enabling monitoring and alerting
• 🔍 Making problems easy to trace and fix

💡 Why Production Logging Matters

Here’s why professional developers prioritize logging:

1. Debugging Without SSH 🔒: Debug issues without accessing production servers
2. Performance Monitoring 📈: Track response times and bottlenecks
3. Security Auditing 🛡️: Know who did what and when
4. Business Intelligence 💼: Understand user behavior and app usage
5. Compliance 📋: Meet regulatory requirements

Real-world example: Imagine an e-commerce site 🛒. Good logging helps you track orders, debug payment failures, monitor inventory updates, and understand why customers abandon their carts!

🔧 Basic Syntax and Usage

📝 Setting Up Python Logging

Let’s start with a production-ready logging setup:

import logging
import logging.handlers
import os
from datetime import datetime

# 🎨 Create a logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

# 📝 Format for our logs
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# 💾 File handler for persistent logs
file_handler = logging.handlers.RotatingFileHandler(
    'app.log',
    maxBytes=10485760,  # 10MB
    backupCount=5
)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)

# 🖥️ Console handler for development
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(formatter)

# ➕ Add handlers to logger
logger.addHandler(file_handler)
logger.addHandler(console_handler)

# 🎯 Let's test it!
logger.info("Application started! 🚀")
logger.debug("Debug mode is active 🐛")

💡 Explanation: We set up rotating file logs (to prevent disk space issues) and console output for development. The formatter ensures consistent, readable log messages!

🎯 Logging Levels

Understanding logging levels is crucial:

# 🔍 DEBUG - Detailed information for diagnosing problems
logger.debug(f"Processing user {user_id} with data: {data}")

# 📢 INFO - General informational messages
logger.info(f"User {user_id} logged in successfully ✅")

# ⚠️ WARNING - Something unexpected but not critical
logger.warning(f"API rate limit approaching: {current_rate}/1000")

# ❌ ERROR - Something failed but app continues
logger.error(f"Failed to send email to {email}: {error}")

# 💥 CRITICAL - Serious error, app might crash
logger.critical("Database connection lost! 🚨")

💡 Practical Examples

🛒 Example 1: E-Commerce Order Processing

Let’s build a production-ready order processing system:

import logging
import json
from typing import Dict, Optional
from datetime import datetime

class OrderProcessor:
    def __init__(self):
        self.logger = logging.getLogger(f"{__name__}.OrderProcessor")
        self.logger.info("OrderProcessor initialized 🛍️")
    
    def process_order(self, order_data: Dict) -> Optional[str]:
        """Process an order with comprehensive logging"""
        order_id = order_data.get('id', 'unknown')
        
        # 🎯 Log important business events
        self.logger.info(
            f"Processing order {order_id}",
            extra={
                'order_id': order_id,
                'customer_id': order_data.get('customer_id'),
                'total_amount': order_data.get('total'),
                'items_count': len(order_data.get('items', []))
            }
        )
        
        try:
            # 📦 Validate inventory
            self._check_inventory(order_data['items'])
            
            # 💳 Process payment
            payment_result = self._process_payment(order_data)
            
            # 🚚 Create shipping
            shipping_id = self._create_shipping(order_data)
            
            # ✅ Success!
            self.logger.info(
                f"Order {order_id} completed successfully! 🎉",
                extra={
                    'order_id': order_id,
                    'shipping_id': shipping_id,
                    'processing_time': datetime.now().isoformat()
                }
            )
            
            return shipping_id
            
        except InventoryError as e:
            self.logger.warning(
                f"Inventory issue for order {order_id}: {e}",
                extra={'order_id': order_id, 'error_type': 'inventory'}
            )
            raise
            
        except PaymentError as e:
            self.logger.error(
                f"Payment failed for order {order_id}: {e}",
                extra={
                    'order_id': order_id,
                    'error_type': 'payment',
                    'payment_method': order_data.get('payment_method')
                },
                exc_info=True  # 🔍 Include stack trace
            )
            raise
            
        except Exception as e:
            self.logger.critical(
                f"Unexpected error processing order {order_id}: {e}",
                extra={'order_id': order_id},
                exc_info=True
            )
            raise
    
    def _check_inventory(self, items):
        """Check if items are in stock"""
        self.logger.debug(f"Checking inventory for {len(items)} items 📦")
        # Inventory logic here
        
    def _process_payment(self, order_data):
        """Process payment securely"""
        # ⚠️ Never log sensitive data!
        self.logger.info(
            "Processing payment",
            extra={
                'amount': order_data['total'],
                'method': order_data['payment_method'],
                # Don't log: credit card numbers, CVV, etc.
            }
        )
        # Payment logic here
        
    def _create_shipping(self, order_data):
        """Create shipping label"""
        self.logger.debug("Creating shipping label 🚚")
        # Shipping logic here
        return f"SHIP-{order_data['id']}"

# 🎮 Custom exceptions
class InventoryError(Exception):
    pass

class PaymentError(Exception):
    pass

🎯 Key Points: Notice how we use structured logging with extra fields, include stack traces for errors, and never log sensitive data!

🎮 Example 2: API Performance Monitoring

Let’s create a logging decorator for API endpoints:

import time
import functools
import logging
from typing import Callable

class APILogger:
    def __init__(self):
        self.logger = logging.getLogger(f"{__name__}.API")
        
    def log_endpoint(self, func: Callable) -> Callable:
        """Decorator to log API endpoint calls with performance metrics"""
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 🕐 Start timing
            start_time = time.time()
            endpoint_name = func.__name__
            
            # 📝 Log request
            self.logger.info(
                f"API call started: {endpoint_name}",
                extra={
                    'endpoint': endpoint_name,
                    'method': kwargs.get('method', 'GET'),
                    'user_id': kwargs.get('user_id', 'anonymous')
                }
            )
            
            try:
                # 🎯 Execute the function
                result = func(*args, **kwargs)
                
                # 📊 Calculate duration
                duration = time.time() - start_time
                
                # ✅ Log success
                self.logger.info(
                    f"API call completed: {endpoint_name}",
                    extra={
                        'endpoint': endpoint_name,
                        'duration_ms': round(duration * 1000, 2),
                        'status': 'success',
                        'response_size': len(str(result))
                    }
                )
                
                # ⚠️ Warn if slow
                if duration > 1.0:
                    self.logger.warning(
                        f"Slow API response: {endpoint_name} took {duration:.2f}s",
                        extra={
                            'endpoint': endpoint_name,
                            'duration_ms': round(duration * 1000, 2)
                        }
                    )
                
                return result
                
            except Exception as e:
                # ❌ Log failure
                duration = time.time() - start_time
                self.logger.error(
                    f"API call failed: {endpoint_name}",
                    extra={
                        'endpoint': endpoint_name,
                        'duration_ms': round(duration * 1000, 2),
                        'status': 'error',
                        'error_type': type(e).__name__
                    },
                    exc_info=True
                )
                raise
        
        return wrapper

# 🎮 Usage example
api_logger = APILogger()

class UserAPI:
    @api_logger.log_endpoint
    def get_user_profile(self, user_id: str, method='GET'):
        """Get user profile with automatic logging"""
        # 🎯 Your API logic here
        time.sleep(0.1)  # Simulate work
        return {'user_id': user_id, 'name': 'Alice', 'level': 42}
    
    @api_logger.log_endpoint
    def update_user_score(self, user_id: str, score: int, method='POST'):
        """Update user score with automatic logging"""
        # 🎯 Your API logic here
        if score < 0:
            raise ValueError("Score cannot be negative! 😱")
        return {'success': True, 'new_score': score}

# 🚀 Test it!
api = UserAPI()
api.get_user_profile(user_id='123')
api.update_user_score(user_id='123', score=100)

🚀 Advanced Concepts

🧙‍♂️ Structured Logging with JSON

For production apps, JSON logs are easier to parse and analyze:

import logging
import json
from pythonjsonlogger import jsonlogger

# 🎨 Setup JSON logging
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# 🚀 Log structured data
logger.info(
    "User action",
    extra={
        "user_id": "123",
        "action": "purchase",
        "item_id": "ABC",
        "amount": 29.99,
        "timestamp": datetime.now().isoformat()
    }
)

# Output: {"message": "User action", "user_id": "123", "action": "purchase", ...}

🏗️ Centralized Logging Configuration

Create a reusable logging configuration:

import logging.config
import os

def setup_logging(app_name: str, environment: str = 'production'):
    """Setup logging configuration for production apps"""
    
    config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'detailed': {
                'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
            },
            'json': {
                '()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
                'format': '%(asctime)s %(name)s %(levelname)s %(message)s'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
                'formatter': 'detailed' if environment == 'development' else 'json',
                'stream': 'ext://sys.stdout'
            },
            'file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'INFO',
                'formatter': 'json',
                'filename': f'/var/log/{app_name}/{app_name}.log',
                'maxBytes': 10485760,  # 10MB
                'backupCount': 5
            },
            'error_file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'ERROR',
                'formatter': 'json',
                'filename': f'/var/log/{app_name}/{app_name}_errors.log',
                'maxBytes': 10485760,  # 10MB
                'backupCount': 5
            }
        },
        'loggers': {
            '': {  # Root logger
                'level': 'INFO',
                'handlers': ['console', 'file', 'error_file']
            },
            'uvicorn': {  # Example: Configure third-party loggers
                'level': 'WARNING'
            }
        }
    }
    
    # 🏗️ Create log directory if needed
    log_dir = f'/var/log/{app_name}'
    os.makedirs(log_dir, exist_ok=True)
    
    # 🎯 Apply configuration
    logging.config.dictConfig(config)
    
    # ✨ Log startup
    logger = logging.getLogger(__name__)
    logger.info(
        f"{app_name} logging initialized! 🚀",
        extra={
            'app_name': app_name,
            'environment': environment,
            'log_directory': log_dir
        }
    )

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Logging Sensitive Data

# ❌ Wrong - Never log passwords or credit cards!
logger.info(f"User login: username={username}, password={password}")
logger.info(f"Payment: card_number={card_number}, cvv={cvv}")

# ✅ Correct - Log safely!
logger.info(f"User login attempt: username={username}")
logger.info(f"Payment processed: last_four={card_number[-4:]}, amount={amount}")

# 🛡️ Even better - Use a sanitizer
def sanitize_sensitive_data(data: dict) -> dict:
    """Remove sensitive fields from log data"""
    sensitive_fields = ['password', 'token', 'api_key', 'secret']
    sanitized = data.copy()
    
    for field in sensitive_fields:
        if field in sanitized:
            sanitized[field] = '***REDACTED***'
    
    return sanitized

# Usage
logger.info("User data", extra=sanitize_sensitive_data(user_data))

🤯 Pitfall 2: Excessive Debug Logging

# ❌ Wrong - This will flood your logs!
for item in huge_list:  # 1 million items
    logger.debug(f"Processing item: {item}")

# ✅ Correct - Log summaries and samples!
logger.debug(f"Processing {len(huge_list)} items")
if len(huge_list) > 0:
    logger.debug(f"First item sample: {huge_list[0]}")

# 📊 Or use sampling
import random
if random.random() < 0.01:  # Log 1% of items
    logger.debug(f"Sample item: {item}")

🐌 Pitfall 3: Synchronous Logging Blocking Performance

# ❌ Wrong - Blocks your app!
logger.info(f"Slow operation: {expensive_calculation()}")

# ✅ Correct - Calculate first, then log!
result = expensive_calculation()
logger.info(f"Operation completed", extra={'result_size': len(result)})

# 🚀 Even better - Use async logging
import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncLogger:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=1)
        self.logger = logging.getLogger(__name__)
    
    async def log_async(self, level, message, **kwargs):
        """Log without blocking the event loop"""
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(
            self.executor,
            lambda: self.logger.log(level, message, **kwargs)
        )

🛠️ Best Practices

1. 🎯 Use Structured Logging: JSON format for easy parsing
2. 📊 Include Context: Add request IDs, user IDs, etc.
3. 🛡️ Protect Sensitive Data: Never log passwords, tokens, or PII
4. 📈 Monitor Performance: Log response times and slow queries
5. 🔄 Rotate Log Files: Prevent disk space issues
6. 🏷️ Use Log Levels Correctly: DEBUG for development, INFO for production
7. 🔍 Include Stack Traces: Use exc_info=True for exceptions
8. 📝 Log Business Events: Not just technical errors
9. 🚀 Async When Possible: Don’t block your application
10. 📋 Follow Standards: Use consistent formats and fields

🧪 Hands-On Exercise

🎯 Challenge: Build a Production-Ready API Logger

Create a comprehensive logging system for a REST API:

📋 Requirements:

• ✅ Log all API requests with timing
• 🛡️ Sanitize sensitive data automatically
• 📊 Track error rates and slow endpoints
• 🔄 Implement request correlation IDs
• 📈 Add performance metrics
• 🎨 Support both JSON and human-readable formats

🚀 Bonus Points:

• Add request/response body logging (with size limits)
• Implement log sampling for high-traffic endpoints
• Create alerts for critical errors
• Add distributed tracing support

💡 Solution

import logging
import time
import uuid
import json
from typing import Dict, Any, Optional
from functools import wraps
from datetime import datetime

class ProductionAPILogger:
    def __init__(self, app_name: str):
        self.app_name = app_name
        self.logger = self._setup_logger()
        self.metrics = {'total_requests': 0, 'errors': 0}
        
    def _setup_logger(self):
        """Setup production-ready logger"""
        logger = logging.getLogger(self.app_name)
        logger.setLevel(logging.INFO)
        
        # 🎨 JSON formatter for production
        json_formatter = logging.Formatter(
            '{"time": "%(asctime)s", "app": "%(name)s", '
            '"level": "%(levelname)s", "message": "%(message)s"}'
        )
        
        # 📝 Handlers
        handler = logging.StreamHandler()
        handler.setFormatter(json_formatter)
        logger.addHandler(handler)
        
        return logger
    
    def _sanitize_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Remove sensitive information"""
        if not isinstance(data, dict):
            return data
            
        sensitive_keys = {
            'password', 'token', 'api_key', 'secret',
            'credit_card', 'ssn', 'authorization'
        }
        
        sanitized = {}
        for key, value in data.items():
            if key.lower() in sensitive_keys:
                sanitized[key] = '***REDACTED***'
            elif isinstance(value, dict):
                sanitized[key] = self._sanitize_data(value)
            else:
                sanitized[key] = value
                
        return sanitized
    
    def log_request(self, method='GET', path='/', user_id=None):
        """Decorator for logging API requests"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 🎯 Generate correlation ID
                correlation_id = str(uuid.uuid4())
                start_time = time.time()
                
                # 📝 Log request
                request_data = {
                    'correlation_id': correlation_id,
                    'method': method,
                    'path': path,
                    'user_id': user_id or 'anonymous',
                    'timestamp': datetime.utcnow().isoformat()
                }
                
                self.logger.info(
                    f"API Request: {method} {path}",
                    extra=self._sanitize_data(request_data)
                )
                
                try:
                    # 🚀 Execute function
                    result = func(*args, **kwargs)
                    
                    # 📊 Calculate metrics
                    duration = (time.time() - start_time) * 1000
                    self.metrics['total_requests'] += 1
                    
                    # ✅ Log success
                    response_data = {
                        'correlation_id': correlation_id,
                        'duration_ms': round(duration, 2),
                        'status': 'success',
                        'path': path
                    }
                    
                    self.logger.info(
                        f"API Response: {method} {path}",
                        extra=response_data
                    )
                    
                    # ⚠️ Warn if slow
                    if duration > 1000:
                        self.logger.warning(
                            f"Slow endpoint detected: {path}",
                            extra={
                                'correlation_id': correlation_id,
                                'duration_ms': duration
                            }
                        )
                    
                    return result
                    
                except Exception as e:
                    # ❌ Log error
                    duration = (time.time() - start_time) * 1000
                    self.metrics['errors'] += 1
                    
                    error_data = {
                        'correlation_id': correlation_id,
                        'duration_ms': round(duration, 2),
                        'status': 'error',
                        'error_type': type(e).__name__,
                        'error_message': str(e),
                        'path': path
                    }
                    
                    self.logger.error(
                        f"API Error: {method} {path}",
                        extra=error_data,
                        exc_info=True
                    )
                    
                    # 🚨 Alert on critical errors
                    error_rate = self.metrics['errors'] / max(self.metrics['total_requests'], 1)
                    if error_rate > 0.1:  # 10% error rate
                        self.logger.critical(
                            "High error rate detected!",
                            extra={
                                'error_rate': round(error_rate * 100, 2),
                                'total_errors': self.metrics['errors']
                            }
                        )
                    
                    raise
            
            return wrapper
        return decorator
    
    def get_metrics(self) -> Dict[str, Any]:
        """Get current metrics"""
        return {
            'total_requests': self.metrics['total_requests'],
            'total_errors': self.metrics['errors'],
            'error_rate': round(
                self.metrics['errors'] / max(self.metrics['total_requests'], 1) * 100, 
                2
            ),
            'timestamp': datetime.utcnow().isoformat()
        }

# 🎮 Example usage
logger = ProductionAPILogger('MyAPI')

class UserAPI:
    @logger.log_request(method='GET', path='/api/users/{id}')
    def get_user(self, user_id: str):
        """Get user with automatic logging"""
        # Simulate work
        time.sleep(0.1)
        return {'id': user_id, 'name': 'Alice', 'email': '[email protected]'}
    
    @logger.log_request(method='POST', path='/api/users/{id}/score')
    def update_score(self, user_id: str, score: int, api_key: str):
        """Update score with automatic sanitization"""
        if score < 0:
            raise ValueError("Invalid score!")
        return {'success': True, 'new_score': score}

# 🚀 Test it!
api = UserAPI()
api.get_user('123')
api.update_score('123', 100, api_key='secret-key-123')

# 📊 Check metrics
print(f"Metrics: {logger.get_metrics()}")

🎓 Key Takeaways

You’ve mastered production logging! Here’s what you can now do:

• ✅ Set up professional logging with proper levels and formatting 💪
• ✅ Protect sensitive data from appearing in logs 🛡️
• ✅ Monitor performance with timing and metrics 📊
• ✅ Debug production issues without SSH access 🔍
• ✅ Build scalable logging systems for any Python app! 🚀

Remember: Good logging is like insurance - you hope you never need it, but when you do, you’ll be incredibly grateful it’s there! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve leveled up your Python logging skills!

Here’s what to do next:

1. 💻 Implement the logging system in your current project
2. 🏗️ Set up centralized logging with ELK or CloudWatch
3. 📚 Learn about distributed tracing with OpenTelemetry
4. 🌟 Share your logging best practices with your team!

Remember: Every production issue you quickly resolve with good logging is a victory. Keep logging smartly, and your future self will thank you! 🚀

Happy logging! 🎉🚀✨