utils

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

execute_command_async(*, command, cwd=None, logger_name=None) async

Execute arbitrary command

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

Parameters:

Name	Type	Description	Default
cwd	Optional[Path]	The working directory for the command execution.	None
command	str	The command to execute.	required

Returns:

Name	Type	Description
stdout	str	The stdout from the command execution.

Raises:

Type	Description
RuntimeError	if the process exited with non-zero status. The error string is set to the stderr of the process.

Source code in fractal_server/utils.py

async def execute_command_async(
    *,
    command: str,
    cwd: Optional[Path] = None,
    logger_name: Optional[str] = None,
) -> str:
    """
    Execute arbitrary command

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

    Args:
        cwd:
            The working directory for the command execution.
        command:
            The command to execute.

    Returns:
        stdout:
            The stdout from the command execution.

    Raises:
        RuntimeError: if the process exited with non-zero status. The error
            string is set to the `stderr` of the process.
    """
    command_split = shlex.split(command)
    cmd, *args = command_split

    logger = get_logger(logger_name)
    cwd_kwarg = dict() if cwd is None else dict(cwd=cwd)
    proc = await asyncio.create_subprocess_exec(
        cmd,
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        **cwd_kwarg,
    )
    stdout, stderr = await proc.communicate()
    logger.debug(f"Subprocess call to: {command}")
    logger.debug(stdout.decode("utf-8"))
    logger.debug(stderr.decode("utf-8"))
    if proc.returncode != 0:
        raise RuntimeError(stderr.decode("utf-8"))
    return stdout.decode("utf-8")

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.

Parameters:

Name	Type	Description	Default
command	str	Command to be executed.	required
logger_name	Optional[str]	Name of the logger.	None
allow_char	Optional[str]	Argument propagated to validate_cmd.	None

Source code in fractal_server/utils.py

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

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

    Arguments:
        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
    logger.debug(f"{returncode=}")
    logger.debug("STDOUT:")
    logger.debug(stdout)
    logger.debug("STDERR:")
    logger.debug(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"
            f"{stdout=}\n"
            f"{stderr=}\n"
        )
    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)