utils

This module provides general purpose utilities that are not specific to any subsystem.

execute_command_sync(*, command, logger_name=None, allow_char=None)

Execute arbitrary command

If the command returns a return code different from zero, a RuntimeError is raised.

PARAMETER	DESCRIPTION
command	Command to be executed. TYPE: str
logger_name	Name of the logger. TYPE: str | None DEFAULT: None
allow_char	Argument propagated to validate_cmd. TYPE: str | None DEFAULT: None

Source code in fractal_server/utils.py

def execute_command_sync(
    *,
    command: str,
    logger_name: str | None = None,
    allow_char: str | None = None,
) -> str:
    """
    Execute arbitrary command

    If the command returns a return code different from zero, a `RuntimeError`
    is raised.

    Args:
        command: Command to be executed.
        logger_name: Name of the logger.
        allow_char: Argument propagated to `validate_cmd`.
    """
    logger = get_logger(logger_name)
    logger.debug(f"START subprocess call to '{command}'")
    validate_cmd(command=command, allow_char=allow_char)
    res = subprocess.run(  # nosec
        shlex.split(command),
        capture_output=True,
        encoding="utf-8",
    )
    returncode = res.returncode
    stdout = res.stdout
    stderr = res.stderr
    if res.returncode != 0:
        logger.debug(f"ERROR in subprocess call to '{command}'")
        raise RuntimeError(
            f"Command {command} failed.\n"
            f"returncode={res.returncode}\n"
            "STDOUT:\n"
            f"{stdout}\n"
            "STDERR:\n"
            f"{stderr}\n"
        )
    logger.debug(f"{returncode=}")
    logger.debug("STDOUT:")
    logger.debug(stdout)
    logger.debug("STDERR:")
    logger.debug(stderr)
    logger.debug(f"END   subprocess call to '{command}'")
    return stdout

get_timestamp()

Get timezone aware timestamp.

Source code in fractal_server/utils.py

def get_timestamp() -> datetime:
    """
    Get timezone aware timestamp.
    """
    return datetime.now(tz=timezone.utc)