Logging Integration

class CorrelationIdFilter(logging.Filter):
    """
    Logging filter to attach correlation IDs to log records.
    
    Parameters:
    - name: Filter name (default: '')
    - uuid_length: Optional length to trim UUID to (default: None)
    - default_value: Default value when no correlation ID exists (default: None)
    """
    def __init__(
        self,
        name: str = '',
        uuid_length: Optional[int] = None,
        default_value: Optional[str] = None
    ): ...
    
    def filter(self, record: LogRecord) -> bool:
        """
        Attach a correlation ID to the log record.
        
        Adds 'correlation_id' attribute to the log record containing:
        - Current correlation ID from context variable
        - Trimmed to uuid_length if specified
        - default_value if no correlation ID exists
        
        Parameters:
        - record: Log record to modify
        
        Returns:
        - bool: Always True (never filters out records)
        """

# In log format string
'%(levelname)s [%(correlation_id)s] %(name)s %(message)s'

# Results in output like:
# INFO [a1b2c3d4] myapp.views Processing user request
# ERROR [a1b2c3d4] myapp.models Database connection failed

class CeleryTracingIdsFilter(logging.Filter):
    """
    Logging filter to attach Celery tracing IDs to log records.
    
    Parameters:
    - name: Filter name (default: '')
    - uuid_length: Optional length to trim UUIDs to (default: None)  
    - default_value: Default value when no IDs exist (default: None)
    """
    def __init__(
        self,
        name: str = '',
        uuid_length: Optional[int] = None,
        default_value: Optional[str] = None
    ): ...
    
    def filter(self, record: LogRecord) -> bool:
        """
        Append parent and current ID to the log record.
        
        Adds two attributes to the log record:
        - 'celery_parent_id': ID of the process that spawned current task
        - 'celery_current_id': Unique ID for the current task process
        
        Both IDs are trimmed to uuid_length if specified, and use
        default_value when the respective context variable is None.
        
        Parameters:
        - record: Log record to modify
        
        Returns:
        - bool: Always True (never filters out records)
        """

# In log format string  
'%(levelname)s [%(celery_current_id)s|%(celery_parent_id)s] %(name)s %(message)s'

# Results in output like:
# INFO [task-abc123|req-xyz789] tasks.email Sending notification email
# ERROR [task-def456|task-abc123] tasks.data Task failed processing batch

import logging
from asgi_correlation_id import CorrelationIdFilter

# Configure logging with correlation ID
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s [%(correlation_id)s] %(name)s %(message)s'
)

# Add filter to root logger
correlation_filter = CorrelationIdFilter()
logging.getLogger().addFilter(correlation_filter)

# Use logging normally - correlation ID is automatically included
logger = logging.getLogger(__name__)
logger.info("This will include the correlation ID")

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s %(levelname)s [%(correlation_id)s] %(name)s:%(lineno)d %(message)s',
            'datefmt': '%Y-%m-%d %H:%M:%S'
        },
        'simple': {
            'format': '%(levelname)s [%(correlation_id)s] %(message)s'
        },
    },
    'filters': {
        'correlation_id': {
            '()': 'asgi_correlation_id.CorrelationIdFilter',
            'uuid_length': 8,  # Trim to 8 characters
            'default_value': 'no-id'
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'detailed',
            'filters': ['correlation_id'],
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': 'app.log',
            'formatter': 'simple',
            'filters': ['correlation_id'],
        },
    },
    'loggers': {
        'myapp': {
            'handlers': ['console', 'file'],
            'level': 'INFO',
            'propagate': False,
        },
    },
}

import logging.config
logging.config.dictConfig(LOGGING)

import logging
from asgi_correlation_id import CeleryTracingIdsFilter

# Configure Celery task logging
celery_filter = CeleryTracingIdsFilter(uuid_length=8, default_value='none')

# Add to Celery logger
celery_logger = logging.getLogger('celery')
celery_logger.addFilter(celery_filter)

# Configure format to include both IDs
logging.basicConfig(
    format='%(asctime)s [%(celery_current_id)s|%(celery_parent_id)s] %(name)s %(message)s'
)

import logging
from asgi_correlation_id import CorrelationIdFilter, CeleryTracingIdsFilter

# Create filters
http_filter = CorrelationIdFilter(uuid_length=8, default_value='no-req')
celery_filter = CeleryTracingIdsFilter(uuid_length=8, default_value='no-task')

# Configure different loggers for different contexts
http_logger = logging.getLogger('webapp')
http_logger.addFilter(http_filter)

celery_logger = logging.getLogger('tasks')  
celery_logger.addFilter(celery_filter)

# Set up formatters for each context
logging.basicConfig(
    format='%(asctime)s %(name)s %(message)s'
)

# HTTP logs will show: 2023-01-01 12:00:00 webapp [a1b2c3d4] Processing request
# Celery logs will show: 2023-01-01 12:00:00 tasks [task-123|req-456] Running background job

# With uuid_length=8
filter = CorrelationIdFilter(uuid_length=8)

# "a1b2c3d4-e5f6-7890-abcd-ef1234567890" becomes "a1b2c3d4"
# "short" remains "short"
# None remains None (or uses default_value)

filter = CorrelationIdFilter(default_value='no-correlation-id')

# When correlation_id.get() returns None:
# Log shows: INFO [no-correlation-id] myapp Processing request

from logging import Filter, LogRecord
from typing import TYPE_CHECKING, Optional

if TYPE_CHECKING:
    from logging import LogRecord

# Internal utility function
def _trim_string(string: Optional[str], string_length: Optional[int]) -> Optional[str]:
    """Trim string to specified length if provided."""