Skip to content

process_manager_interface

process_manager.process_manager_interface

Module providing functions to interact with the drunc process manager.

Classes

ProcessAction

Bases: Enum

Enum for process actions.

Functions

boot_process(user, data)

Boot a process with the given data.

Parameters:

Name Type Description Default
user str

the user to boot the process as.

 required
data dict[str, str | int]

the data for the process.

 required
Source code in process_manager/process_manager_interface.py 
 97
 98
 99
100
101
102
103
104
def boot_process(user: str, data: dict[str, str | int]) -> None:
    """Boot a process with the given data.

    Args:
        user: the user to boot the process as.
        data: the data for the process.
    """
    return asyncio.run(_boot_process(user, data))

get_process_logs(uuid)

Retrieve logs for a process from the process manager.

Parameters:

Name Type Description Default
uuid str

UUID of the process.

 required

Returns:

Type Description
list[DecodedResponse]

The process logs.
Source code in process_manager/process_manager_interface.py 
79
80
81
82
83
84
85
86
87
88
def get_process_logs(uuid: str) -> list[DecodedResponse]:
    """Retrieve logs for a process from the process manager.

    Args:
      uuid: UUID of the process.

    Returns:
      The process logs.
    """
    return asyncio.run(_get_process_logs(uuid))

get_process_manager_driver()

Get a ProcessManagerDriver instance.

Source code in process_manager/process_manager_interface.py 
18
19
20
21
22
23
def get_process_manager_driver() -> ProcessManagerDriver:
    """Get a ProcessManagerDriver instance."""
    token = create_dummy_token_from_uname()
    return ProcessManagerDriver(
        settings.PROCESS_MANAGER_URL, token=token, aio_channel=True
    )

get_session_info()

Get info about all sessions from process manager.

Source code in process_manager/process_manager_interface.py 
32
33
34
def get_session_info() -> ProcessInstanceList:
    """Get info about all sessions from process manager."""
    return asyncio.run(_get_session_info())

process_call(uuids, action)

Perform an action on a process with a given UUID.

Parameters:

Name Type Description Default
uuids Iterable[str]

List of UUIDs of the process to be actioned.

 required
action ProcessAction

Action to be performed {restart,flush,kill}.

 required
Source code in process_manager/process_manager_interface.py 
62
63
64
65
66
67
68
69
def process_call(uuids: Iterable[str], action: ProcessAction) -> None:
    """Perform an action on a process with a given UUID.

    Args:
        uuids: List of UUIDs of the process to be actioned.
        action: Action to be performed {restart,flush,kill}.
    """
    return asyncio.run(_process_call(uuids, action))