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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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("session_manager.iface.SessionManagerDriver")
    self.address = address
    options = [
        ("grpc.keepalive_time_ms", 60000)  # pings the server every 60 seconds
    ]
    raw_channel = grpc.insecure_channel(self.address, options=options)
    rich_interceptor = RichErrorClientInterceptor(logger=self.log)
    self.channel = grpc.intercept_channel(raw_channel, rich_interceptor)
    self.stub = SessionManagerStub(self.channel)
    self.token = copy_token(token)
    self.log = get_logger("session_manager_driver", rich_handler=True)
Methods:
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 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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.
    """
    self.log.info(f"Sending describe request to session manager at {self.address}")

    request = Request(token=copy_token(self.token))

    response: Description = self.stub.describe(request, timeout=timeout)

    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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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.
    """
    self.log.info(
        f"Sending list_all_configs request to session manager at {self.address}"
    )

    request = Request(token=copy_token(self.token))

    response: AllConfigKeys = self.stub.list_all_configs(request, timeout=timeout)

    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 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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.
    """
    self.log.info(
        f"Sending list_all_sessions request to session manager at {self.address}"
    )

    request = Request(token=copy_token(self.token))

    response: AllActiveSessions = self.stub.list_all_sessions(
        request, timeout=timeout
    )

    return response
load_session(file, session_id, timeout=60)

Load a session based on the provided configuration key.

Parameters:

Name Type Description Default
file str

The file containing the session to load.

 required
session_id str

The ID of the session to load.

 required
timeout int | float

The timeout for the gRPC call in seconds.

 60

Returns:

Name Type Description
LoadSessionResponse LoadSessionResponse

A response containing loaded session metadata.
Source code in drunc/session_manager/session_manager_driver.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def load_session(
    self, file: str, session_id: str, timeout: int | float = 60
) -> LoadSessionResponse:
    """Load a session based on the provided configuration key.

    Args:
        file: The file containing the session to load.
        session_id: The ID of the session to load.
        timeout: The timeout for the gRPC call in seconds.

    Returns:
        LoadSessionResponse: A response containing loaded session metadata.
    """
    self.log.info(
        f"Sending load_session request to session manager at {self.address}"
    )

    request = LoadSessionRequest(
        config_key=ConfigKey(
            file=file,
            session_id=session_id,
        )
    )

    response: LoadSessionResponse = self.stub.load_session(request, timeout=timeout)

    return response

Functions: