kamihi.base.utils

Utility functions and variables for the Kamihi project.

License

MIT

Functions:

Name	Description
cron_regex	Get the compiled regex pattern for validating cron expressions.
is_valid_cron_expression	Validate if a given string is a valid cron expression.
requires	Check if required optional dependencies are available.
timer	Context manager to log the time taken for a block of code.

cron_regex cached

cron_regex() -> re.Pattern

Get the compiled regex pattern for validating cron expressions.

Returns:

Type	Description
Pattern	re.Pattern: The compiled regex pattern.

Source code in src/kamihi/base/utils.py

@functools.lru_cache(maxsize=1)
def cron_regex() -> re.Pattern:
    """
    Get the compiled regex pattern for validating cron expressions.

    Returns:
        re.Pattern: The compiled regex pattern.

    """
    # Month and weekday names
    month_name = r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)"
    weekday_name = r"(?:mon|tue|wed|thu|fri|sat|sun)"
    name = rf"(?:{month_name}|{weekday_name})"

    # Basic building blocks
    number = r"\d+"
    star = r"\*"
    step = rf"(?:\*/{number})"
    rng = rf"(?:{number}|{name})-(?:{number}|{name})"
    rng_step = rf"{rng}/{number}"
    xth_y = rf"{number}th\s+(?:{number}|{weekday_name})"
    last_x = rf"last\s+(?:{number}|{weekday_name})"
    last = r"last"

    # Single field expression
    single_expr = rf"(?:{star}|{step}|{rng_step}|{rng}|{number}|{name}|{xth_y}|{last_x}|{last})"

    # Comma-separated list
    field = rf"{single_expr}(?:,{single_expr})*"

    # Full crontab line (5–7 fields)
    return re.compile(
        rf"^(?:{field}\s+){{4,6}}{field}$",  # 5–7 fields
        re.IGNORECASE,
    )

is_valid_cron_expression

is_valid_cron_expression(expression: str) -> bool

Validate if a given string is a valid cron expression.

Parameters:

Name	Type	Description	Default
expression	str	The cron expression to validate.	required

Returns:

Name	Type	Description
bool	bool	True if the expression is valid, False otherwise.

Source code in src/kamihi/base/utils.py

def is_valid_cron_expression(expression: str) -> bool:
    """
    Validate if a given string is a valid cron expression.

    Args:
        expression (str): The cron expression to validate.

    Returns:
        bool: True if the expression is valid, False otherwise.

    """
    return bool(cron_regex().match(expression.strip()))

requires

requires(group: str) -> Callable

Check if required optional dependencies are available.

Parameters:

Name	Type	Description	Default
group	str	The name of the extra group to check for dependencies.	required

Source code in src/kamihi/base/utils.py

def requires(group: str) -> Callable:
    """
    Check if required optional dependencies are available.

    Args:
        group (str): The name of the extra group to check for dependencies.

    """

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            _check_extra_installed(group)
            return func(*args, **kwargs)

        return wrapper

    return decorator

timer

timer(
    logger: Logger, message: str, level: str = "DEBUG"
) -> Generator[None, Any, None]

Context manager to log the time taken for a block of code.

Parameters:

Name	Type	Description	Default
logger	Logger	The logger instance to use for logging.	required
message	str	The message to log with the elapsed time.	required
level	str	The logging level to use (default is "DEBUG").	'DEBUG'

Returns:

Type	Description
None	Generator[None, Any, None]: A generator that yields control to the block of code being timed.

Source code in src/kamihi/base/utils.py

@contextmanager
def timer(logger: "Logger", message: str, level: str = "DEBUG") -> Generator[None, Any, None]:  # noqa: ANN001
    """
    Context manager to log the time taken for a block of code.

    Args:
        logger (Logger): The logger instance to use for logging.
        message (str): The message to log with the elapsed time.
        level (str): The logging level to use (default is "DEBUG").

    Returns:
        Generator[None, Any, None]: A generator that yields control to the block of code being timed.

    """
    start_time = time.perf_counter()
    yield
    end_time = time.perf_counter()
    logger.bind(ms=round((end_time - start_time) * 1000)).log(level, message)