utils

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

execute_command(*, cwd, command, 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	Path	The working directory for the command execution.	required
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(
    *,
    cwd: Path,
    command: str,
    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)
    proc = await asyncio.create_subprocess_exec(
        cmd,
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=cwd,
    )
    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")

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)