radical.asyncflow.logging¶

def init_default_logger(
    log_level: Union[int, str] = logging.INFO,
    *,
    output_file: Optional[Union[str, pathlib.Path]] = None,
    file_log_level: Optional[Union[int, str]] = None,
    use_colors: bool = True,
    show_details: bool = False,
    style: str = "modern",
    clear_handlers: bool = False,
    logger_name: Optional[str] = None,
    structured_logging: bool = False,
    structured_file: Optional[Union[str, pathlib.Path]] = None,
) -> logging.Logger:
    """Setup and configure logging with enhanced features.

    Args:
        log_level: Base logging level for console output.
        output_file: Path to log file. If provided, file logging is enabled.
        file_log_level: Logging level for file output. Defaults to log_level.
        use_colors: Enable colored console output.
        show_details: Include thread/process info in log messages.
        style: Format style - 'modern', 'core', 'execution_backend', or 'minimal'.
        clear_handlers: Remove existing handlers from root logger.
        logger_name: Name for the logger. If None, returns root logger.
        structured_logging: Enable JSON structured logging to file.
        structured_file: Custom path for structured JSON log file.

    Returns:
        Configured logger instance.
    """
    # Get or create logger
    logger = logging.getLogger(logger_name) if logger_name else logging.getLogger()

    # Clear existing handlers if requested
    if clear_handlers:
        logger.handlers.clear()

    # Console always uses colored formatter (never JSON)
    console_formatter = _ColoredFormatter(
        use_colors=use_colors, show_details=show_details, style=style
    )

    # Console handler
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(console_formatter)
    console_handler.setLevel(log_level)

    if show_details:
        console_handler.addFilter(_thread_info_filter)

    logger.addHandler(console_handler)

    # Regular file handler (if requested) - uses colored formatter without colors
    if output_file is not None:
        file_level = log_level if file_log_level is None else file_log_level
        file_path = pathlib.Path(output_file)
        file_path.parent.mkdir(parents=True, exist_ok=True)

        file_handler = logging.FileHandler(file_path)
        file_formatter = _ColoredFormatter(
            use_colors=False,  # No colors in file
            show_details=show_details,
            style=style,
        )
        file_handler.setFormatter(file_formatter)
        file_handler.setLevel(file_level)

        if show_details:
            file_handler.addFilter(_thread_info_filter)

        logger.addHandler(file_handler)

    # Structured JSON file handler (if requested)
    if structured_logging:
        # Auto-generate structured log file name if not provided
        if structured_file is None:
            if output_file is not None:
                # Use same directory as regular log file
                structured_file = pathlib.Path(output_file).with_suffix(".json")
            else:
                # Default to structured.log in current directory
                structured_file = "radical.asyncflow.logs.json"

        struct_level = log_level if file_log_level is None else file_log_level
        struct_path = pathlib.Path(structured_file)
        struct_path.parent.mkdir(parents=True, exist_ok=True)

        struct_handler = logging.FileHandler(struct_path)
        struct_formatter = _StructuredFormatter()
        struct_handler.setFormatter(struct_formatter)
        struct_handler.setLevel(struct_level)

        if show_details:
            struct_handler.addFilter(_thread_info_filter)

        logger.addHandler(struct_handler)

    # Set logger level
    logger.setLevel(logging.NOTSET)

    # Capture warnings
    logging.captureWarnings(True)

    # Log configuration info
    level_name = (
        logging.getLevelName(log_level) if isinstance(log_level, int) else log_level
    )
    file_level_name = (
        logging.getLevelName(file_log_level)
        if isinstance(file_log_level, int)
        else file_log_level
    )

    structured_info = str(structured_file) if structured_logging else "disabled"

    logger.info(
        "Logger configured successfully - "
        "Console: %s, File: %s (%s), Structured: %s, Style: %s",
        level_name,
        output_file or "disabled",
        file_level_name or "N/A",
        structured_info,
        style,
    )

    return logger

def get_structured_logger(
    name: str = None, level: Union[int, str] = logging.INFO
) -> StructuredLogger:
    """Get a structured logger that supports both old and new usage patterns."""
    # Check if logger already configured, if not configure it with structured logging
    base_logger = logging.getLogger(name)
    if not base_logger.handlers:
        init_default_logger(level, logger_name=name, structured_logging=True)
        base_logger = logging.getLogger(name)

    return StructuredLogger(base_logger)

def get_logger(
    name: str = None, level: Union[int, str] = logging.INFO
) -> logging.Logger:
    """Quick logger setup for simple use cases."""
    if not logging.getLogger().handlers:
        init_default_logger(level, logger_name=name, style="modern")
    return logging.getLogger(name)