colored_logger

colored_logger

Colored logger wrapper that automatically applies colors to log levels.

This module provides a ColoredLogger class that wraps cl and automatically applies appropriate colors to different log levels for better visual distinction in console output.

Quick Start for Users and AI Agents

Standard Usage (Recommended):

from crew_dcs.utils.logging import get_colored_logger

# Get a logger with default settings (INFO level, colored output)
logger = get_colored_logger()

# Use in async functions
await logger.info("Operation completed")
await logger.warning("This might be a problem")
await logger.error("Something went wrong")

Configuration for Scripts/Notebooks:

from crew_dcs.utils.logging import set_domolibrary_logger

# Configure logger with custom settings
logger = set_domolibrary_logger(
    app_name="my_script",
    min_level="DEBUG",  # Show DEBUG, INFO, WARNING, ERROR, CRITICAL
    set_as_global=True,  # Make this the global logger for all modules
)

Adjust Level Dynamically:

# Show more verbose output
logger.set_level("DEBUG")

# Show only warnings and errors
logger.set_level("WARNING")

# Check current level
current_level = logger.get_level()  # Returns 'DEBUG', 'INFO', etc.

For Route Functions (API calls):

from crew_dcs.utils.logging import log_call, LogDecoratorConfig

@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def my_route_function(auth, **kwargs):
    # Function automatically logs entry/exit with colored output
    pass

Level Hierarchy: - DEBUG (10) - Most verbose, shows everything - INFO (20) - Normal operations (default) - WARNING (30) - Warning messages only - ERROR (40) - Error messages only - CRITICAL (50) - Most severe only

Important Notes: - ALWAYS import from crew_dcs.utils.logging, NEVER from cl directly - NEVER import cl in crew_dcs code - Logger methods are async: use await logger.info(), not logger.info() - Use set_domolibrary_logger() at script entry points for global configuration - Use get_colored_logger() within modules for local logger instances

ColoredLogger

ColoredLogger(
    base_logger: CrewLogger,
    debug_color: str = "cyan",
    info_color: str = "green",
    warning_color: str = "yellow",
    error_color: str = "bold_red",
    critical_color: str = "bold_red",
    min_level: str = "INFO",
    exclude_patterns: list[str] | None = None,
)

Bases: CrewLogger

Logger that automatically colorizes messages by log level.

This class wraps cl's Logger and adds automatic coloring and level filtering. It's the standard logger for all crew_dcs code.

Default colors

• DEBUG: cyan
• INFO: green
• WARNING: yellow
• ERROR: red
• CRITICAL: bold_red

Usage

from crew_dcs.utils.logging import get_colored_logger

Basic usage

logger = get_colored_logger() await logger.info("This will be green") await logger.warning("This will be yellow") await logger.error("This will be red")

Control verbosity

logger.set_level("DEBUG") # Show all messages logger.set_level("WARNING") # Show only warnings and errors

Check current level

print(f"Current level: {logger.get_level()}")

Note

• All log methods are ASYNC - must use await
• Level filtering happens at this layer before passing to cl
• Use set_domolibrary_logger() for global configuration
• Use get_colored_logger() for module-level instances

Initialize colored logger wrapper.

Parameters:

Name	Type	Description	Default
base_logger	CrewLogger	The underlying cl Logger instance	required
debug_color	str	Color for debug messages (default: cyan)	'cyan'
info_color	str	Color for info messages (default: green)	'green'
warning_color	str	Color for warning messages (default: yellow)	'yellow'
error_color	str	Color for error messages (default: bold_red)	'bold_red'
critical_color	str	Color for critical messages (default: bold_red)	'bold_red'
min_level	str	Minimum log level to display (default: INFO)	'INFO'
exclude_patterns	list[str] | None	List of message patterns to filter out (default: None)	None

Source code in src/crew_dcs/utils/logging/colored_logger.py

def __init__(
    self,
    base_logger: CrewLogger,
    debug_color: str = "cyan",
    info_color: str = "green",
    warning_color: str = "yellow",
    error_color: str = "bold_red",
    critical_color: str = "bold_red",
    min_level: str = "INFO",
    exclude_patterns: list[str] | None = None,
):
    """
    Initialize colored logger wrapper.

    Args:
        base_logger: The underlying cl Logger instance
        debug_color: Color for debug messages (default: cyan)
        info_color: Color for info messages (default: green)
        warning_color: Color for warning messages (default: yellow)
        error_color: Color for error messages (default: bold_red)
        critical_color: Color for critical messages (default: bold_red)
        min_level: Minimum log level to display (default: INFO)
        exclude_patterns: List of message patterns to filter out (default: None)
    """
    # Don't call super().__init__() - we're wrapping, not inheriting data
    self._logger = base_logger
    self.debug_color = debug_color
    self.info_color = info_color
    self.warning_color = warning_color
    self.error_color = error_color
    self.critical_color = critical_color
    self._min_level = self._LEVEL_HIERARCHY[min_level.upper()]
    self.exclude_patterns = exclude_patterns or []

    # Register this instance for level synchronization
    _register_colored_logger(self)

critical async

critical(
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any
) -> bool

Log CRITICAL level message with automatic coloring.

Source code in src/crew_dcs/utils/logging/colored_logger.py

async def critical(
    self,
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any,
) -> bool:
    """Log CRITICAL level message with automatic coloring."""
    if not self._should_log("CRITICAL", message):
        return False
    colored_msg = colorize(message, color or self.critical_color)
    return await self._logger.critical(
        colored_msg, method=method, level_name=level_name, **context
    )

debug async

debug(
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any
) -> bool

Log DEBUG level message with automatic coloring.

Source code in src/crew_dcs/utils/logging/colored_logger.py

async def debug(
    self,
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any,
) -> bool:
    """Log DEBUG level message with automatic coloring."""
    should_log = self._should_log("DEBUG", message)

    if not should_log:
        return False
    colored_msg = colorize(message, color or self.debug_color)
    return await self._logger.debug(
        colored_msg, method=method, level_name=level_name, **context
    )

error async

error(
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any
) -> bool

Log ERROR level message with automatic coloring.

Source code in src/crew_dcs/utils/logging/colored_logger.py

async def error(
    self,
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any,
) -> bool:
    """Log ERROR level message with automatic coloring."""
    if not self._should_log("ERROR", message):
        return False
    colored_msg = colorize(message, color or self.error_color)
    return await self._logger.error(
        colored_msg, method=method, level_name=level_name, **context
    )

get_level

get_level() -> str

Get the current minimum log level.

Returns:

Type	Description
str	Current log level name (e.g., 'INFO', 'WARNING', 'ERROR')

Example

logger = get_colored_logger() logger.get_level() 'INFO'

Source code in src/crew_dcs/utils/logging/colored_logger.py

def get_level(self) -> str:
    """
    Get the current minimum log level.

    Returns:
        Current log level name (e.g., 'INFO', 'WARNING', 'ERROR')

    Example:
        >>> logger = get_colored_logger()
        >>> logger.get_level()
        'INFO'
    """
    for name, value in self._LEVEL_HIERARCHY.items():
        if value == self._min_level:
            return name
    return "INFO"  # Default

info async

info(
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any
) -> bool

Log INFO level message with automatic coloring.

Source code in src/crew_dcs/utils/logging/colored_logger.py

async def info(
    self,
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any,
) -> bool:
    """Log INFO level message with automatic coloring."""
    if not self._should_log("INFO", message):
        return False
    colored_msg = colorize(message, color or self.info_color)
    return await self._logger.info(
        colored_msg, method=method, level_name=level_name, **context
    )

log async

log(level: Any, message: str, **context: Any) -> bool

Log with automatic coloring for cl log() usage.

Source code in src/crew_dcs/utils/logging/colored_logger.py

async def log(self, level: Any, message: str, **context: Any) -> bool:
    """Log with automatic coloring for cl log() usage."""
    from cl.client.enums import LogLevel

    if isinstance(level, LogLevel):
        level_name = level.name
        level_enum = level
    else:
        level_name = str(level).upper()
        try:
            level_enum = LogLevel[level_name]
        except (KeyError, AttributeError):
            level_enum = LogLevel.INFO
            level_name = "INFO"

    if not self._should_log(level_name, message):
        return False

    color = {
        "DEBUG": self.debug_color,
        "INFO": self.info_color,
        "WARNING": self.warning_color,
        "ERROR": self.error_color,
        "CRITICAL": self.critical_color,
    }.get(level_name)

    colored_msg = colorize(message, color)
    return await self._logger.log(level=level_enum, message=colored_msg, **context)

set_level

set_level(level: str)

Set the minimum log level for this logger.

Parameters:

Name	Type	Description	Default
level	str	Log level name - 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'	required

Example

logger = get_colored_logger() logger.set_level('WARNING') # Only show WARNING and above logger.set_level('ERROR') # Only show ERROR and CRITICAL logger.set_level('INFO') # Show INFO and above (default)

Source code in src/crew_dcs/utils/logging/colored_logger.py

def set_level(self, level: str):
    """
    Set the minimum log level for this logger.

    Args:
        level: Log level name - 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'

    Example:
        >>> logger = get_colored_logger()
        >>> logger.set_level('WARNING')  # Only show WARNING and above
        >>> logger.set_level('ERROR')    # Only show ERROR and CRITICAL
        >>> logger.set_level('INFO')     # Show INFO and above (default)
    """
    level_upper = level.upper()
    if level_upper not in self._LEVEL_HIERARCHY:
        valid_levels = ", ".join(self._LEVEL_HIERARCHY.keys())
        raise ValueError(
            f"Invalid log level '{level}'. Valid levels: {valid_levels}"
        )

    self._min_level = self._LEVEL_HIERARCHY[level_upper]
    return self

warning async

warning(
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any
) -> bool

Log WARNING level message with automatic coloring.

Source code in src/crew_dcs/utils/logging/colored_logger.py

async def warning(
    self,
    message: str,
    method: str = "COMMENT",
    level_name: str | None = None,
    color: str | None = None,
    **context: Any,
) -> bool:
    """Log WARNING level message with automatic coloring."""
    if not self._should_log("WARNING", message):
        return False
    colored_msg = colorize(message, color or self.warning_color)
    return await self._logger.warning(
        colored_msg, method=method, level_name=level_name, **context
    )

LogDecoratorConfig dataclass

LogDecoratorConfig(
    entity_extractor: Any = None,
    result_processor: Any = None,
)

Compatibility shim for LogDecoratorConfig.

colorize

colorize(message: str, color: str | None = None) -> str

Apply ANSI color codes to a message.

Parameters:

Name	Type	Description	Default
message	str	The message to colorize	required
color	str | None	Color name (e.g., 'red', 'green', 'cyan', 'bold_red')	None

Returns:

Type	Description
str	Colorized message string

Source code in src/crew_dcs/utils/logging/colored_logger.py

def colorize(message: str, color: str | None = None) -> str:
    """
    Apply ANSI color codes to a message.

    Args:
        message: The message to colorize
        color: Color name (e.g., 'red', 'green', 'cyan', 'bold_red')

    Returns:
        Colorized message string
    """
    if not color:
        return message

    color_code = _COLOR_CODES.get(color.lower())
    if color_code:
        return f"{color_code}{message}{_COLOR_CODES['reset']}"

    return message

get_colored_logger

get_colored_logger(
    app_name: str | None = None,
    debug_color: str = "cyan",
    info_color: str = "green",
    warning_color: str = "yellow",
    error_color: str = "bold_red",
    critical_color: str = "bold_red",
    env: str = "development",
    exclude_patterns: list[str] | None = None,
    min_level: str | None = None,
    enable_datadog: bool | None = None,
    set_as_global: bool = False,
    force: bool = False,
) -> ColoredLogger

Get or create a colored logger instance with advanced features.

Standard Usage (Module-Level): Call with no parameters to get the existing global logger. This is the recommended pattern for module-level logger variables: >>> logger = get_colored_logger() # Gets existing global logger

Advanced Usage (Script Entry Points): Pass app_name to create a new logger with custom handlers. Use set_as_global=True to make it the global logger. For most cases, prefer set_domolibrary_logger() instead.

Parameters:

Name	Type	Description	Default
app_name	str | None	Name of the application for logging context (default: None) If provided, creates a NEW logger with handlers. If None, returns existing global logger.	None
debug_color	str	Color for debug messages (default: cyan)	'cyan'
info_color	str	Color for info messages (default: green)	'green'
warning_color	str	Color for warning messages (default: yellow)	'yellow'
error_color	str	Color for error messages (default: bold_red)	'bold_red'
critical_color	str	Color for critical messages (default: bold_red)	'bold_red'
env	str	Environment name ('development', 'production', 'staging', etc.)	'development'
exclude_patterns	list[str] | None	List of message patterns to filter out (default: None)	None
min_level	str | None	Minimum log level to display (default: inherit from global logger or INFO)	None
enable_datadog	bool | None	Enable Datadog handler (auto-detect from env if None)	None
set_as_global	bool	Set this as cl's global logger (default: False) Only set to True when configuring at entry points. Modules should leave as False.	False
force	bool	Force setting as global even if user has configured logger (default: False)	False

Returns:

Type	Description
ColoredLogger	ColoredLogger instance with automatic color application

Example

Standard usage - get existing global logger (recommended for modules)

logger = get_colored_logger() await logger.info("Success!") # Will be green

Advanced usage - create new logger at entry point

logger = get_colored_logger( ... app_name="MyApp", ... min_level="DEBUG", ... set_as_global=True, # Set as global for all modules ... )

For entry points, prefer set_domolibrary_logger() instead:

from crew_dcs.utils.logging import set_domolibrary_logger logger = set_domolibrary_logger(app_name="MyApp", min_level="DEBUG")

Source code in src/crew_dcs/utils/logging/colored_logger.py

def get_colored_logger(  # noqa: C901
    app_name: str | None = None,
    debug_color: str = "cyan",
    info_color: str = "green",
    warning_color: str = "yellow",
    error_color: str = "bold_red",
    critical_color: str = "bold_red",
    env: str = "development",
    exclude_patterns: list[str] | None = None,
    min_level: str | None = None,
    enable_datadog: bool | None = None,
    set_as_global: bool = False,
    force: bool = False,
) -> ColoredLogger:
    """
    Get or create a colored logger instance with advanced features.

    **Standard Usage (Module-Level):**
    Call with no parameters to get the existing global logger. This is the recommended
    pattern for module-level logger variables:
        >>> logger = get_colored_logger()  # Gets existing global logger

    **Advanced Usage (Script Entry Points):**
    Pass app_name to create a new logger with custom handlers. Use set_as_global=True
    to make it the global logger. For most cases, prefer set_domolibrary_logger() instead.

    Args:
        app_name: Name of the application for logging context (default: None)
            If provided, creates a NEW logger with handlers. If None, returns existing global logger.
        debug_color: Color for debug messages (default: cyan)
        info_color: Color for info messages (default: green)
        warning_color: Color for warning messages (default: yellow)
        error_color: Color for error messages (default: bold_red)
        critical_color: Color for critical messages (default: bold_red)
        env: Environment name ('development', 'production', 'staging', etc.)
        exclude_patterns: List of message patterns to filter out (default: None)
        min_level: Minimum log level to display (default: inherit from global logger or INFO)
        enable_datadog: Enable Datadog handler (auto-detect from env if None)
        set_as_global: Set this as cl's global logger (default: False)
            Only set to True when configuring at entry points. Modules should leave as False.
        force: Force setting as global even if user has configured logger (default: False)

    Returns:
        ColoredLogger instance with automatic color application

    Example:
        >>> # Standard usage - get existing global logger (recommended for modules)
        >>> logger = get_colored_logger()
        >>> await logger.info("Success!")  # Will be green
        >>>
        >>> # Advanced usage - create new logger at entry point
        >>> logger = get_colored_logger(
        ...     app_name="MyApp",
        ...     min_level="DEBUG",
        ...     set_as_global=True,  # Set as global for all modules
        ... )
        >>>
        >>> # For entry points, prefer set_domolibrary_logger() instead:
        >>> from crew_dcs.utils.logging import set_domolibrary_logger
        >>> logger = set_domolibrary_logger(app_name="MyApp", min_level="DEBUG")
    """
    # Always get the current global logger
    current_global = _get_global_logger(app_name=app_name or "crew_dcs")

    # If the global logger is already a ColoredLogger, return it directly
    # This ensures all modules share the same logger instance
    if (
        isinstance(current_global, ColoredLogger)
        and not app_name
        and not enable_datadog
    ):
        return current_global

    # If min_level not specified, inherit from existing ColoredLogger or use INFO as default
    if min_level is None:
        if isinstance(current_global, ColoredLogger):
            # Inherit level from existing global ColoredLogger
            min_level = current_global.get_level()
        else:
            min_level = "INFO"

    # Auto-detect Datadog enablement from environment
    if enable_datadog is None:
        enable_datadog = env.lower() in ("production", "prod", "staging")

    # If advanced features are requested, create a new logger with config
    if app_name or enable_datadog:
        from cl import ConsoleLogConfig, DatadogLogConfig, MultiHandlerLogConfig
        from cl.client.enums import LogLevel

        # Convert min_level string to LogLevel enum
        log_level_enum = LogLevel.INFO
        try:  # noqa: SIM105
            log_level_enum = LogLevel[min_level.upper()]
        except (KeyError, AttributeError):
            pass  # Keep INFO default

        console_cfg = ConsoleLogConfig(level=log_level_enum)
        handlers_cfg = [HandlerConfig(type="console", config=console_cfg)]

        # Datadog handler (if enabled)
        if enable_datadog:
            datadog_api_key = os.getenv("DATADOG_API_KEY")
            if not datadog_api_key:
                print(
                    "[WARNING] Datadog requested but DATADOG_API_KEY not set. Datadog logging disabled."
                )
            else:
                try:
                    dd_cfg = DatadogLogConfig(
                        api_key=datadog_api_key,
                        site=os.getenv("DATADOG_SITE", "datadoghq.com"),
                        service=app_name or "crew_dcs",
                        env=env,
                        level=log_level_enum,
                    )
                    handlers_cfg.append(
                        HandlerConfig(
                            type="cloud",
                            config=dd_cfg,
                            platform_config=dd_cfg.to_platform_config(),
                        )
                    )
                    print(f"[INFO] Datadog logging enabled for environment: {env}")
                except (ImportError, AttributeError):
                    print(
                        "[WARNING] Datadog handler not available. Install cl with Datadog support."
                    )

        if len(handlers_cfg) == 1:
            base_config = console_cfg
        else:
            base_config = MultiHandlerLogConfig(
                handlers=handlers_cfg,
                level=log_level_enum,
            )

        # Create new logger with config
        # Note: Filtering is done by ColoredLogger wrapper, not the base CrewLogger
        base_logger = CrewLogger(
            config=base_config,
            app_name=app_name or "crew_dcs",
        )

        # Create colored wrapper
        new_logger = ColoredLogger(
            base_logger=base_logger,
            debug_color=debug_color,
            info_color=info_color,
            warning_color=warning_color,
            error_color=error_color,
            critical_color=critical_color,
            min_level=min_level,
            exclude_patterns=exclude_patterns,
        )

        # Set as global logger
        if set_as_global:
            set_global_logger(new_logger)

        return new_logger

    # Simple path: wrap existing global logger (no caching)
    # Create a temporary wrapper - caller should store if needed
    colored = ColoredLogger(
        base_logger=current_global,
        debug_color=debug_color,
        info_color=info_color,
        warning_color=warning_color,
        error_color=error_color,
        critical_color=critical_color,
        min_level=min_level,
        exclude_patterns=exclude_patterns,
    )

    # Enable DEBUG output on underlying cl if min_level is DEBUG
    if min_level.upper() == "DEBUG":
        current_global.show_debugging = True
        colored.show_debugging = True
        # Also update handler log levels to allow DEBUG messages through
        from cl.client.enums import LogLevel

        if hasattr(current_global, "handlers"):
            for handler in current_global.handlers:
                if hasattr(handler, "log_level"):
                    handler.log_level = LogLevel.DEBUG

    return colored

log_call

log_call(
    action_name: str | None = None,
    level_name: str | None = None,
    config: Any = None,
    color: str | None = None,
    logger: CrewLogger | None = None,
    log_level: str | None = None,
    **kwargs: Any
) -> Callable

Compatibility wrapper for log_call decorator.

Maps old API parameters to new log_function_call API. Also filters out 'logger' parameter that log_function_call injects.

Note: Route functions (level_name='route') default to DEBUG level for API operations. Business logic decisions should use INFO level explicitly in method implementations.

Source code in src/crew_dcs/utils/logging/colored_logger.py

def log_call(  # noqa: C901
    action_name: str | None = None,
    level_name: str | None = None,
    config: Any = None,
    color: str | None = None,
    logger: CrewLogger | None = None,
    log_level: str | None = None,
    **kwargs: Any,
) -> Callable:
    """
    Compatibility wrapper for log_call decorator.

    Maps old API parameters to new log_function_call API.
    Also filters out 'logger' parameter that log_function_call injects.

    Note: Route functions (level_name='route') default to DEBUG level for API operations.
    Business logic decisions should use INFO level explicitly in method implementations.
    """
    import inspect
    from functools import wraps

    # Extract parameters from config if provided
    if config and hasattr(config, "entity_extractor"):
        pass
    if config and hasattr(config, "result_processor"):
        pass

    # Map old parameters to new API
    # Note: Some old parameters may not be supported in new API
    decorator_kwargs = {
        "action_name": action_name,
        "logger": logger,
    }

    # Map log_level if provided, defaulting route/auth functions to DEBUG
    if log_level:
        from cl.client.enums import LogLevel

        with contextlib.suppress(KeyError, AttributeError):
            decorator_kwargs["log_level"] = LogLevel[log_level.upper()]
    elif level_name in {"route", "auth"}:
        # Default route/auth functions to DEBUG level for API operations
        from cl.client.enums import LogLevel

        with contextlib.suppress(KeyError, AttributeError):
            decorator_kwargs["log_level"] = LogLevel.DEBUG

    # Remove None values
    decorator_kwargs = {k: v for k, v in decorator_kwargs.items() if v is not None}

    # Apply the decorator
    decorated_func = log_function_call(**decorator_kwargs)

    # Wrap to filter out 'logger' parameter and handle dynamic log_level from context
    def wrapper(func: Callable) -> Callable:  # noqa: C901
        # Get the function signature to check if it accepts 'logger' or 'context'
        sig = inspect.signature(func)
        accepts_logger = "logger" in sig.parameters
        accepts_context = "context" in sig.parameters or any(
            param.kind == inspect.Parameter.VAR_KEYWORD
            for param in sig.parameters.values()
        )

        # If function doesn't accept logger, wrap it first to filter out logger
        # Also handle dynamic log_level from context
        if not accepts_logger or accepts_context:
            if inspect.iscoroutinefunction(func):

                @wraps(func)
                async def func_wrapper(*args, **kwargs):  # noqa: C901
                    # Extract log_level from context if available
                    context = kwargs.get("context")
                    dynamic_log_level = None
                    if context and hasattr(context, "log_level") and context.log_level:
                        dynamic_log_level = context.log_level

                    # If we have a dynamic log_level, temporarily set it on the logger
                    logger = kwargs.get("logger")
                    original_level = None
                    if logger and dynamic_log_level:
                        try:
                            # Convert to string if it's a LogLevel enum
                            if hasattr(dynamic_log_level, "name"):
                                log_level_str = dynamic_log_level.name
                            elif isinstance(dynamic_log_level, str):
                                log_level_str = dynamic_log_level.upper()
                            else:
                                log_level_str = str(dynamic_log_level).upper()

                            # Store original level and set new level
                            if hasattr(logger, "set_level"):
                                # Get current level before changing
                                if hasattr(logger, "get_level"):
                                    original_level = logger.get_level()
                                elif hasattr(logger, "_min_level"):
                                    # For ColoredLogger, map numeric level back to string
                                    level_map = {
                                        10: "DEBUG",
                                        20: "INFO",
                                        30: "WARNING",
                                        40: "ERROR",
                                        50: "CRITICAL",
                                    }
                                    original_level = level_map.get(
                                        logger._min_level, "INFO"
                                    )

                                # Set the new level
                                logger.set_level(log_level_str)
                            elif hasattr(logger, "_logger") and hasattr(
                                logger._logger, "set_level"
                            ):
                                # For ColoredLogger wrapper, set on underlying logger
                                if hasattr(logger._logger, "get_level"):
                                    original_level = logger._logger.get_level()
                                logger._logger.set_level(log_level_str)
                        except (KeyError, AttributeError, TypeError, ValueError):
                            # If log_level conversion fails, continue without changing level
                            pass

                    # Remove logger if function doesn't accept it
                    if not accepts_logger:
                        kwargs.pop("logger", None)

                    try:
                        result = await func(*args, **kwargs)
                    finally:
                        # Restore original log level if we changed it
                        if logger and original_level is not None:
                            try:
                                if hasattr(logger, "set_level"):
                                    logger.set_level(original_level)
                                elif hasattr(logger, "_logger") and hasattr(
                                    logger._logger, "set_level"
                                ):
                                    logger._logger.set_level(original_level)
                            except (AttributeError, TypeError, ValueError):
                                pass

                    return result

                func_to_decorate = func_wrapper
            else:

                @wraps(func)
                def func_wrapper(*args, **kwargs):  # noqa: C901
                    # Extract log_level from context if available
                    context = kwargs.get("context")
                    dynamic_log_level = None
                    if context and hasattr(context, "log_level") and context.log_level:
                        dynamic_log_level = context.log_level

                    # If we have a dynamic log_level, temporarily set it on the logger
                    logger = kwargs.get("logger")
                    original_level = None
                    if logger and dynamic_log_level:
                        try:
                            # Convert to string if it's a LogLevel enum
                            if hasattr(dynamic_log_level, "name"):
                                log_level_str = dynamic_log_level.name
                            elif isinstance(dynamic_log_level, str):
                                log_level_str = dynamic_log_level.upper()
                            else:
                                log_level_str = str(dynamic_log_level).upper()

                            # Store original level and set new level
                            if hasattr(logger, "set_level"):
                                # Get current level before changing
                                if hasattr(logger, "get_level"):
                                    original_level = logger.get_level()
                                elif hasattr(logger, "_min_level"):
                                    # For ColoredLogger, map numeric level back to string
                                    level_map = {
                                        10: "DEBUG",
                                        20: "INFO",
                                        30: "WARNING",
                                        40: "ERROR",
                                        50: "CRITICAL",
                                    }
                                    original_level = level_map.get(
                                        logger._min_level, "INFO"
                                    )

                                # Set the new level
                                logger.set_level(log_level_str)
                            elif hasattr(logger, "_logger") and hasattr(
                                logger._logger, "set_level"
                            ):
                                # For ColoredLogger wrapper, set on underlying logger
                                if hasattr(logger._logger, "get_level"):
                                    original_level = logger._logger.get_level()
                                logger._logger.set_level(log_level_str)
                        except (KeyError, AttributeError, TypeError, ValueError):
                            # If log_level conversion fails, continue without changing level
                            pass

                    # Remove logger if function doesn't accept it
                    if not accepts_logger:
                        kwargs.pop("logger", None)

                    try:
                        result = func(*args, **kwargs)
                    finally:
                        # Restore original log level if we changed it
                        if logger and original_level is not None:
                            try:
                                if hasattr(logger, "set_level"):
                                    logger.set_level(original_level)
                                elif hasattr(logger, "_logger") and hasattr(
                                    logger._logger, "set_level"
                                ):
                                    logger._logger.set_level(original_level)
                            except (AttributeError, TypeError, ValueError):
                                pass

                    return result

                func_to_decorate = func_wrapper
        else:
            func_to_decorate = func

        # Apply the log_function_call decorator to the (possibly wrapped) function
        inner = decorated_func(func_to_decorate)

        # Per-layer logging: when context.is_log_<layer> is False, skip decorator logging
        _level_name = level_name
        _LAYER_ATTR = {
            "client": "is_log_client",
            "route": "is_log_route",
            "class": "is_log_class",
        }

        def _should_skip_log(context: Any) -> bool:
            if context is None or not _level_name:
                return False
            attr = _LAYER_ATTR.get(_level_name)
            return bool(attr and getattr(context, attr, None) is False)

        if inspect.iscoroutinefunction(func):

            @wraps(func)
            async def outer(*args: Any, **kwargs: Any) -> Any:
                if _should_skip_log(kwargs.get("context")):
                    return await func_to_decorate(*args, **kwargs)
                return await inner(*args, **kwargs)

            return outer

        @wraps(func)
        def outer(*args: Any, **kwargs: Any) -> Any:
            if _should_skip_log(kwargs.get("context")):
                return func_to_decorate(*args, **kwargs)
            return inner(*args, **kwargs)

        return outer

    return wrapper

set_domolibrary_logger

set_domolibrary_logger(
    logger: ColoredLogger | None = None,
    set_as_global: bool = True,
    **kwargs
) -> ColoredLogger

Explicitly set the domolibrary logger, optionally making it the global logger.

This is the recommended way to configure logging for domolibrary if you want colored output throughout the library.

Parameters:

Name	Type	Description	Default
logger	ColoredLogger | None	Existing ColoredLogger instance, or None to create a new one	None
set_as_global	bool	Set this as cl's global logger (default: True)	True
**kwargs		Arguments to pass to get_colored_logger if creating a new instance	{}

Returns:

Type	Description
ColoredLogger	The ColoredLogger instance

Example

Option 1: Use default colored logger

from crew_dcs.utils.logging import set_domolibrary_logger logger = set_domolibrary_logger()

Option 2: Create with custom colors

logger = set_domolibrary_logger( ... info_color="blue", ... error_color="magenta" ... )

Option 3: Use your own ColoredLogger

my_logger = ColoredLogger(base_logger=my_base_logger) set_domolibrary_logger(logger=my_logger)

Source code in src/crew_dcs/utils/logging/colored_logger.py

def set_domolibrary_logger(
    logger: ColoredLogger | None = None,
    set_as_global: bool = True,
    **kwargs,
) -> ColoredLogger:
    """
    Explicitly set the domolibrary logger, optionally making it the global logger.

    This is the recommended way to configure logging for domolibrary if you want
    colored output throughout the library.

    Args:
        logger: Existing ColoredLogger instance, or None to create a new one
        set_as_global: Set this as cl's global logger (default: True)
        **kwargs: Arguments to pass to get_colored_logger if creating a new instance

    Returns:
        The ColoredLogger instance

    Example:
        >>> # Option 1: Use default colored logger
        >>> from crew_dcs.utils.logging import set_domolibrary_logger
        >>> logger = set_domolibrary_logger()
        >>>
        >>> # Option 2: Create with custom colors
        >>> logger = set_domolibrary_logger(
        ...     info_color="blue",
        ...     error_color="magenta"
        ... )
        >>>
        >>> # Option 3: Use your own ColoredLogger
        >>> my_logger = ColoredLogger(base_logger=my_base_logger)
        >>> set_domolibrary_logger(logger=my_logger)
    """
    if logger is None:
        logger = get_colored_logger(set_as_global=True, force=True, **kwargs)

    if set_as_global:
        set_global_logger(logger)
        # Update all existing loggers to use the same level
        # This ensures module-level loggers created before this call
        # get their levels synchronized
        _update_all_loggers_level(logger.get_level())

    return logger