Skip to content

process_manager_interface

interfaces.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 interfaces/process_manager_interface.py 
102
103
104
105
106
107
108
109
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_hostnames(user)

Get the hostnames of the processes for the given user.

Parameters:

Name Type Description Default
user str

The user to get the hostnames for.

 required

Returns:

Type Description
dict[str, str]

The hostnames of the processes for the given user.
Source code in interfaces/process_manager_interface.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def get_hostnames(user: str) -> dict[str, str]:
    """Get the hostnames of the processes for the given user.

    Args:
        user: The user to get the hostnames for.

    Returns:
        The hostnames of the processes for the given user.
    """
    session = get_session_info(user)
    hostnames = {}
    for process_instance in session.data.values:  # type: ignore [attr-defined]
        hostnames[process_instance.process_description.metadata.name] = (
            process_instance.process_description.metadata.hostname
        )
    return hostnames

get_process_logs(uuid, username)

Retrieve logs for a process from the process manager.

Parameters:

Name Type Description Default
uuid str

UUID of the process.

 required
username str

Username of the user requesting the logs

 required

Returns:

Type Description
list[DecodedResponse]

The process logs.
Source code in interfaces/process_manager_interface.py 
83
84
85
86
87
88
89
90
91
92
93
def get_process_logs(uuid: str, username: str) -> list[DecodedResponse]:
    """Retrieve logs for a process from the process manager.

    Args:
      uuid: UUID of the process.
      username: Username of the user requesting the logs

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

get_process_manager_driver(username)

Get a ProcessManagerDriver instance.

Source code in interfaces/process_manager_interface.py 
19
20
21
22
23
24
def get_process_manager_driver(username: str) -> ProcessManagerDriver:
    """Get a ProcessManagerDriver instance."""
    token = Token(token=f"{username}-token", user_name=username)
    return ProcessManagerDriver(
        settings.PROCESS_MANAGER_URL, token=token, aio_channel=True
    )

get_session_info(username)

Get info about all sessions from process manager.

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

process_call(uuids, action, username)

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
username str

Username of the user performing the action

 required
Source code in interfaces/process_manager_interface.py 
65
66
67
68
69
70
71
72
73
def process_call(uuids: Iterable[str], action: ProcessAction, username: str) -> 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}.
        username: Username of the user performing the action
    """
    return asyncio.run(_process_call(uuids, action, username))