Skip to content

session_manager_driver

drunc.session_manager.session_manager_driver

Driver for the session manager service.

Classes

SessionManagerDriver(address, token)

Provides an interface to the session manager service.

This class provides the client-side methods required to interact with a remote session manager service, via gRPC connections.

Create a new session manager driver instance.

Parameters:

Name Type Description Default
address str

The address of the session manager service.

 required
token Token

The token for authentication.

 required
**kwargs

Additional keyword arguments for the driver.

 required
Source code in drunc/session_manager/session_manager_driver.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, address: str, token: Token):
    """Create a new session manager driver instance.

    Args:
        address: The address of the session manager service.
        token: The token for authentication.
        **kwargs: Additional keyword arguments for the driver.
    """
    self.log = get_logger("controller.SessionManagerDriver")
    self.address = address
    options = [
        ("grpc.keepalive_time_ms", 60000)  # pings the server every 60 seconds
    ]
    self.channel = grpc.insecure_channel(self.address, options=options)
    self.stub = SessionManagerStub(self.channel)
    self.token = copy_token(token)
    self.log = get_logger("session_manager_driver")
Functions
describe(timeout=60)

Describe the session manager service.

Parameters:

Name Type Description Default
timeout int | float

The timeout for the gRPC call in seconds.

 60

Returns:

Type Description
Description

A response containing the description of the service.
Source code in drunc/session_manager/session_manager_driver.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def describe(self, timeout: int | float = 60) -> Description:
    """Describe the session manager service.

    Args:
        timeout: The timeout for the gRPC call in seconds.

    Returns:
        A response containing the description of the service.
    """
    request = Request(token=copy_token(self.token))

    try:
        response = self.stub.describe(request, timeout=timeout)
    except grpc.RpcError as e:
        try:
            error_details = extract_grpc_rich_error(e)
            self.log.error(error_details)
        except Exception as extraction_error:
            self.log.debug(
                f"Could not extract rich error details from gRPC error: {extraction_error}",
                exc_info=True,
            )

        handle_grpc_error(e)

    return response
list_all_configs(timeout=60)

List all available configurations in the session manager.

Parameters:

Name Type Description Default
timeout int | float

The timeout for the gRPC call in seconds.

 60

Returns:

Type Description
AllConfigKeys

A response containing all available configuration keys.
Source code in drunc/session_manager/session_manager_driver.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def list_all_configs(self, timeout: int | float = 60) -> AllConfigKeys:
    """List all available configurations in the session manager.

    Args:
        timeout: The timeout for the gRPC call in seconds.

    Returns:
        A response containing all available configuration keys.
    """
    request = Request(token=copy_token(self.token))

    try:
        response = self.stub.list_all_configs(request, timeout=timeout)
    except grpc.RpcError as e:
        try:
            error_details = extract_grpc_rich_error(e)
            self.log.error(error_details)
        except Exception as extraction_error:
            self.log.debug(
                f"Could not extract rich error details from gRPC error: {extraction_error}",
                exc_info=True,
            )

        handle_grpc_error(e)

    return response
list_all_sessions(timeout=60)

List all active sessions managed by the session manager.

Parameters:

Name Type Description Default
timeout int | float

The timeout for the gRPC call in seconds.

 60

Returns:

Type Description
AllActiveSessions

A response containing a list of all active sessions.
Source code in drunc/session_manager/session_manager_driver.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def list_all_sessions(self, timeout: int | float = 60) -> AllActiveSessions:
    """List all active sessions managed by the session manager.

    Args:
        timeout: The timeout for the gRPC call in seconds.

    Returns:
        A response containing a list of all active sessions.
    """
    request = Request(token=copy_token(self.token))

    try:
        response = self.stub.list_all_sessions(request, timeout=timeout)
    except grpc.RpcError as e:
        try:
            error_details = extract_grpc_rich_error(e)
            self.log.error(error_details)
        except Exception as extraction_error:
            self.log.debug(
                f"Could not extract rich error details from gRPC error: {extraction_error}",
                exc_info=True,
            )

        handle_grpc_error(e)

    return response

Functions