Structured Logging & Observability Best Practices (2025 Edition)¶

# Log level hierarchy
DEBUG = 10    # Detailed information for debugging
INFO = 20     # General information
WARNING = 30   # Something unexpected happened
ERROR = 40    # Serious problem occurred
CRITICAL = 50 # Very serious error occurred

# Context fields for data workflows
context_fields = {
    "service": "geospatial-etl",
    "flow_id": "flow_abc123",
    "task_id": "task_xyz789",
    "correlation_id": "req_456def",
    "user_id": "user_789",
    "tenant_id": "tenant_456",
    "environment": "production",
    "version": "1.2.3"
}

# requirements.txt
structlog==23.2.0
structlog-sentry==23.1.0

# logging_config.py
import structlog
import logging
import sys
from typing import Any, Dict

def configure_logging(service_name: str, environment: str = "development"):
    """Configure structured logging for Python applications"""

    # Configure structlog
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            structlog.processors.JSONRenderer()
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    # Configure standard library logging
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=logging.INFO,
    )

    # Create logger with context
    logger = structlog.get_logger(service_name)
    logger = logger.bind(
        service=service_name,
        environment=environment
    )

    return logger

# Usage in Prefect tasks
from prefect import task, flow
import structlog

logger = structlog.get_logger("geospatial-etl")

@task
def process_geospatial_data(data: dict) -> dict:
    """Process geospatial data with structured logging"""

    # Bind task context
    task_logger = logger.bind(
        task_id="process_geospatial_data",
        input_size=len(data),
        data_type=type(data).__name__
    )

    task_logger.info("Starting geospatial data processing")

    try:
        # Process data
        result = process_data(data)

        task_logger.info(
            "Geospatial data processing completed",
            output_size=len(result),
            processing_time_ms=1500
        )

        return result

    except Exception as e:
        task_logger.error(
            "Geospatial data processing failed",
            error=str(e),
            error_type=type(e).__name__,
            exc_info=True
        )
        raise

@flow
def geospatial_etl_flow():
    """Main ETL flow with structured logging"""

    flow_logger = logger.bind(flow_id="geospatial_etl_flow")
    flow_logger.info("Starting geospatial ETL flow")

    # Process data
    data = extract_data()
    result = process_geospatial_data(data)
    load_data(result)

    flow_logger.info("Geospatial ETL flow completed successfully")

# requirements.txt
loguru==0.7.2

# loguru_config.py
from loguru import logger
import sys
import json
from typing import Dict, Any

def serialize_record(record: Dict[str, Any]) -> str:
    """Serialize log record to JSON"""
    subset = {
        "timestamp": record["time"].isoformat(),
        "level": record["level"].name,
        "message": record["message"],
        "module": record["module"],
        "function": record["function"],
        "line": record["line"],
        **record["extra"]
    }
    return json.dumps(subset)

def configure_loguru(service_name: str, environment: str = "development"):
    """Configure loguru for structured logging"""

    # Remove default handler
    logger.remove()

    # Add JSON handler
    logger.add(
        sys.stdout,
        format=serialize_record,
        level="INFO",
        serialize=True
    )

    # Add file handler for errors
    logger.add(
        "logs/error.log",
        format=serialize_record,
        level="ERROR",
        serialize=True,
        rotation="100 MB",
        retention="30 days"
    )

    # Bind service context
    logger = logger.bind(service=service_name, environment=environment)

    return logger

# Usage in FastAPI
from fastapi import FastAPI, Request
from loguru import logger
import uuid

app = FastAPI()

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    """FastAPI middleware for request logging"""

    # Generate correlation ID
    correlation_id = str(uuid.uuid4())

    # Log request
    logger.info(
        "Request started",
        method=request.method,
        url=str(request.url),
        correlation_id=correlation_id,
        client_ip=request.client.host
    )

    # Process request
    response = await call_next(request)

    # Log response
    logger.info(
        "Request completed",
        status_code=response.status_code,
        correlation_id=correlation_id
    )

    return response

# prefect_logging.py
from prefect import task, flow, get_run_logger
import structlog
from typing import Dict, Any

@task
def geospatial_etl_task(data: Dict[str, Any]) -> Dict[str, Any]:
    """Prefect task with structured logging"""

    # Get Prefect logger
    prefect_logger = get_run_logger()

    # Create structured logger
    logger = structlog.get_logger("geospatial-etl")

    # Bind task context
    task_logger = logger.bind(
        task_id=prefect_logger.task_run_id,
        flow_id=prefect_logger.flow_run_id,
        task_name="geospatial_etl_task"
    )

    task_logger.info(
        "Starting geospatial ETL task",
        input_data_size=len(data),
        input_keys=list(data.keys())
    )

    try:
        # Process data
        result = process_geospatial_data(data)

        task_logger.info(
            "Geospatial ETL task completed",
            output_data_size=len(result),
            processing_time_ms=1500
        )

        return result

    except Exception as e:
        task_logger.error(
        "Geospatial ETL task failed",
        error=str(e),
        error_type=type(e).__name__,
        exc_info=True
    )
        raise

@flow
def geospatial_etl_flow():
    """Main ETL flow with structured logging"""

    logger = structlog.get_logger("geospatial-etl")
    flow_logger = logger.bind(flow_id="geospatial_etl_flow")

    flow_logger.info("Starting geospatial ETL flow")

    # Extract data
    data = extract_geospatial_data()

    # Transform data
    result = geospatial_etl_task(data)

    # Load data
    load_geospatial_data(result)

    flow_logger.info("Geospatial ETL flow completed successfully")

# geospatial_logging.py
import structlog
from sqlalchemy import text
from sqlalchemy.orm import sessionmaker

logger = structlog.get_logger("geospatial-db")

class GeospatialDatabaseLogger:
    def __init__(self, engine):
        self.engine = engine
        self.Session = sessionmaker(bind=engine)

    def execute_geospatial_query(self, query: str, params: dict = None):
        """Execute geospatial query with structured logging"""

        query_logger = logger.bind(
            query_type="geospatial",
            query_hash=hash(query),
            params=params
        )

        query_logger.info("Executing geospatial query", query=query)

        try:
            with self.Session() as session:
                result = session.execute(text(query), params or {})

                query_logger.info(
                    "Geospatial query executed successfully",
                    row_count=result.rowcount,
                    execution_time_ms=1500
                )

                return result.fetchall()

        except Exception as e:
            query_logger.error(
                "Geospatial query failed",
                error=str(e),
                error_type=type(e).__name__,
                exc_info=True
            )
            raise

# log_redaction.py
import re
import json
from typing import Dict, Any

class LogRedactor:
    def __init__(self):
        self.patterns = [
            (r'password["\']?\s*[:=]\s*["\']?([^"\']+)["\']?', r'password="***"'),
            (r'token["\']?\s*[:=]\s*["\']?([^"\']+)["\']?', r'token="***"'),
            (r'api_key["\']?\s*[:=]\s*["\']?([^"\']+)["\']?', r'api_key="***"'),
            (r'secret["\']?\s*[:=]\s*["\']?([^"\']+)["\']?', r'secret="***"'),
        ]

    def redact(self, log_message: str) -> str:
        """Redact sensitive information from log message"""
        for pattern, replacement in self.patterns:
            log_message = re.sub(pattern, replacement, log_message, flags=re.IGNORECASE)
        return log_message

    def redact_dict(self, log_dict: Dict[str, Any]) -> Dict[str, Any]:
        """Redact sensitive information from log dictionary"""
        redacted = {}
        for key, value in log_dict.items():
            if key.lower() in ['password', 'token', 'api_key', 'secret']:
                redacted[key] = "***"
            elif isinstance(value, dict):
                redacted[key] = self.redact_dict(value)
            elif isinstance(value, str):
                redacted[key] = self.redact(value)
            else:
                redacted[key] = value
        return redacted

# Essential logging operations
# 1. Configure structured logging
logger = structlog.get_logger("service-name")

# 2. Bind context
logger = logger.bind(
    service="service-name",
    flow_id="flow_abc123",
    task_id="task_xyz789"
)

# 3. Log with context
logger.info("Processing data", input_size=1000, processing_time_ms=1500)

# 4. Handle errors
logger.error("Processing failed", error=str(e), exc_info=True)

# 5. Query logs
# {job="service-name"} |= "ERROR"
# {job="service-name"} |= "flow_id" |= "abc123"