Skip to content

Session API

Overview

The Session API is the core of canfar, enabling you to create, manage, and destroy sessions on the CANFAR Science Platform.

Bases: HTTPClient

CANFAR Session Management Client.

This class provides methods to manage sessions, including fetching session details, creating new sessions, retrieving logs, and destroying existing sessions. It is a subclass of the HTTPClient class and inherits its attributes and methods.

Examples:

>>> from canfar.session import Session
>>> session = Session(
        timeout=120,
        concurrency=100, # No effect on sync client
        loglevel=40,
    )
Source code in canfar/sessions.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 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
 69
 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
 96
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
class Session(HTTPClient):
    """CANFAR Session Management Client.

    This class provides methods to manage sessions, including fetching
    session details, creating new sessions, retrieving logs, and
    destroying existing sessions. It is a subclass of the `HTTPClient`
    class and inherits its attributes and methods.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session(
                timeout=120,
                concurrency=100, # No effect on sync client
                loglevel=40,
            )
    """

    def fetch(
        self,
        kind: Kind | None = None,
        status: Status | None = None,
        view: View | None = None,
    ) -> list[dict[str, str]]:
        """Fetch open sessions for the user.

        Args:
            kind (Kind | None, optional): Session kind. Defaults to None.
            status (Status | None, optional): Session status. Defaults to None.
            view (View | None, optional): View leve. Defaults to None.

        Returns:
            list[dict[str, str]]: Session[s] information.

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.fetch(kind="notebook")
            [{'id': 'ikvp1jtp',
              'userid': 'username',
              'image': 'image-server/image/label:latest',
              'type': 'notebook',
              'status': 'Running',
              'name': 'example-notebook',
              'startTime': '2222-12-14T02:24:06Z',
              'connectURL': 'https://something.example.com/ikvp1jtp',
              'requestedRAM': '16G',
              'requestedCPUCores': '2',
              'requestedGPUCores': '<none>',
              'coresInUse': '0m',
              'ramInUse': '101Mi'}]
        """
        parameters: dict[str, Any] = build.fetch_parameters(kind, status, view)
        response: Response = self.client.get(url="session", params=parameters)
        data: list[dict[str, str]] = response.json()
        return data

    def stats(self) -> dict[str, Any]:
        """Get statistics for the entire platform.

        Returns:
            Dict[str, Any]: Cluster statistics.

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.stats()
            {'cores': {'requestedCPUCores': 377,
             'coresAvailable': 960,
             'maxCores': {'cores': 32, 'withRam': '147Gi'}},
             'ram': {'maxRAM': {'ram': '226Gi', 'withCores': 32}}}
        """
        parameters = {"view": "stats"}
        response: Response = self.client.get("session", params=parameters)
        data: dict[str, Any] = response.json()
        return data

    def info(self, ids: list[str] | str) -> list[dict[str, Any]]:
        """Get information about session[s].

        Args:
            ids (Union[List[str], str]): Session ID[s].

        Returns:
            Dict[str, Any]: Session information.

        Examples:
            >>> session.info(ids="hjko98yghj")
            >>> session.info(ids=["hjko98yghj", "ikvp1jtp"])
        """
        # Convert id to list if it is a string
        if isinstance(ids, str):
            ids = [ids]
        results: list[dict[str, Any]] = []
        for value in ids:
            try:
                response: Response = self.client.get(url=f"session/{value}")
                results.append(response.json())
            except HTTPError:
                err = f"failed to fetch session info for {value}"
                log.exception(err)
        return results

    def logs(
        self,
        ids: list[str] | str,
        verbose: bool = False,
    ) -> dict[str, str] | None:
        """Get logs from a session[s].

        Args:
            ids (Union[List[str], str]): Session ID[s].
            verbose (bool, optional): Print logs to stdout. Defaults to False.

        Returns:
            Dict[str, str]: Logs in text/plain format.

        Examples:
            >>> session.logs(id="hjko98yghj")
            >>> session.logs(id=["hjko98yghj", "ikvp1jtp"])
        """
        if isinstance(ids, str):
            ids = [ids]
        parameters: dict[str, str] = {"view": "logs"}
        results: dict[str, str] = {}

        for value in ids:
            try:
                response: Response = self.client.get(
                    url=f"session/{value}",
                    params=parameters,
                )
                results[value] = response.text
            except HTTPError:
                err = f"failed to fetch logs for session {value}"
                log.exception(err)

        if verbose:
            for key, value in results.items():
                log.info("Session ID: %s\n", key)
                log.info(value)
            return None

        return results

    def create(
        self,
        name: str,
        image: str,
        cores: int | None = None,
        ram: int | None = None,
        kind: Kind = "headless",
        gpu: int | None = None,
        cmd: str | None = None,
        args: str | None = None,
        env: dict[str, Any] | None = None,
        replicas: int = 1,
    ) -> list[str]:
        """Launch a canfar session.

        Args:
            name (str): A unique name for the session.
            image (str): Container image to use for the session.
            cores (int, optional): Number of cores.
                Defaults to None, i.e. flexible mode.
            ram (int, optional): Amount of RAM (GB).
                Defaults to None, i.e. flexible mode.
            kind (str, optional): Type of canfar session. Defaults to "headless".
            gpu (Optional[int], optional): Number of GPUs. Defaults to None.
            cmd (Optional[str], optional): Command to run. Defaults to None.
            args (Optional[str], optional): Arguments to the command. Defaults to None.
            env (Optional[Dict[str, Any]], optional): Environment variables to inject.
                Defaults to None.
            replicas (int, optional): Number of sessions to launch. Defaults to 1.

        Notes:
            - If cores and ram are not specified, the session will be created with
              flexible resource allocation of upto 8 cores and 32GB of RAM.
            - The name of the session suffixed with the replica number. eg. test-42
              when replicas > 1.
            - Each container will have the following environment variables injected:
                * REPLICA_ID - The replica number
                * REPLICA_COUNT - The total number of replicas

        Returns:
            List[str]: A list of session IDs for the launched sessions.

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.create(
                    name="test",
                    image='images.canfar.net/skaha/terminal:1.1.1',
                    cores=2,
                    ram=8,
                    gpu=1,
                    kind="headless",
                    cmd="env",
                    env={"TEST": "test"},
                    replicas=2,
                )
            >>> ["hjko98yghj", "ikvp1jtp"]
        """
        payloads = build.create_parameters(
            name,
            image,
            cores,
            ram,
            kind,
            gpu,
            cmd,
            args,
            env,
            replicas,
        )
        results: list[str] = []
        log.debug("Creating %d %s session[s].", replicas, kind)
        for payload in payloads:
            try:
                response: Response = self.client.post(url="session", params=payload)
                results.append(response.text.rstrip("\r\n"))
            except HTTPError:
                err = f"Failed to create session with payload: {payload}"
                log.exception(err)
        return results

    def events(
        self,
        ids: str | list[str],
        verbose: bool = False,
    ) -> list[dict[str, str]] | None:
        """Get deployment events for a session[s].

        Args:
            ids (Union[str, List[str]]): Session ID[s].
            verbose (bool, optional): Print events to stdout. Defaults to False.

        Returns:
            Optional[List[Dict[str, str]]]: A list of events for the session[s].

        Notes:
            When verbose is True, the events will be printed to stdout only.

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.events(ids="hjko98yghj")
            >>> session.events(ids=["hjko98yghj", "ikvp1jtp"])
        """
        if isinstance(ids, str):
            ids = [ids]
        results: list[dict[str, str]] = []
        parameters: dict[str, str] = {"view": "events"}
        for value in ids:
            try:
                response: Response = self.client.get(
                    url=f"session/{value}",
                    params=parameters,
                )
                results.append({value: response.text})
            except HTTPError:
                err = f"Failed to fetch events for session {value}"
                log.exception(err)
        if verbose and results:
            for result in results:
                for key, value in result.items():
                    log.info("Session ID: %s", key)
                    log.info("\n %s", value)
        return results if not verbose else None

    def destroy(self, ids: str | list[str]) -> dict[str, bool]:
        """Destroy canfar session[s].

        Args:
            ids (Union[str, List[str]]): Session ID[s].

        Returns:
            Dict[str, bool]: A dictionary of session IDs
            and a bool indicating if the session was destroyed.

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.destroy(id="hjko98yghj")
            >>> session.destroy(id=["hjko98yghj", "ikvp1jtp"])
        """
        if isinstance(ids, str):
            ids = [ids]
        results: dict[str, bool] = {}
        for value in ids:
            try:
                self.client.delete(url=f"session/{value}")
                results[value] = True
            except HTTPError:
                msg = f"Failed to destroy session {value}"
                log.exception(msg)
                results[value] = False
        return results

    def destroy_with(
        self,
        prefix: str,
        *,
        kind: Kind = "headless",
        status: Status = "Completed",
    ) -> dict[str, bool]:
        """Destroy session[s] matching a prefix or regex.

        Args:
            prefix (str): Prefix to match.
                Treated literally unless regex meta-characters are found.
            kind (Kind): Type of session. Defaults to "headless".
            status (Status): Status of the session. Defaults to "Completed".

        Returns:
            Dict[str, bool]: A dictionary of session IDs
            and a bool indicating if the session was destroyed.

        Notes:
            - If the value contains regex metacharacters (e.g., `.^$*+?{}[]()|`),
              it is treated as a regex with :func:`re.search`.
            - Otherwise it is treated as a literal prefix (anchored with `^`).
            This method is useful for destroying multiple sessions at once.

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.destroy_with(prefix="test")  # literal prefix
            >>> session.destroy_with(prefix="desktop$")  # regex
        """
        meta = set(".^$*+?{}[]()|")
        has_meta = any(ch in meta for ch in prefix)
        try:
            if has_meta:
                log.info("destroy_with using regex pattern: %s", prefix)
                regex = re.compile(prefix)
            else:
                log.info("destroy_with using literal prefix: %s", prefix)
                regex = re.compile(rf"^{re.escape(prefix)}")
        except re.error as exc:
            msg = f"Invalid regex pattern '{prefix}': {exc}"
            log.exception(msg)
            raise ValueError(msg) from exc

        sessions = self.fetch(kind=kind, status=status)
        ids: list[str] = [
            session["id"] for session in sessions if regex.search(session["name"])
        ]
        return self.destroy(ids)

    def connect(self, ids: list[str] | str) -> None:
        """Open session[s] in a web browser.

        Args:
            ids (Union[List[str], str]): Session ID[s].

        Examples:
            >>> from canfar.session import Session
            >>> session = Session()
            >>> session.connect(ids="hjko98yghj")
            >>> session.connect(ids=["hjko98yghj", "ikvp1jtp"])
        """
        if isinstance(ids, str):
            ids = [ids]
        info = self.info(ids)
        log.debug(info)
        for session in info:
            status: str = session.get("status", "unknown")
            if status != "Running":
                log.warning("Session %s is currently %s.", session["id"], status)
                log.warning("Please wait for the session to be ready.")
                continue
            connect_url = session.get("connectURL")
            if connect_url:
                open_new_tab(connect_url)

connect 

connect(ids)

Open session[s] in a web browser.

PARAMETER DESCRIPTION
ids

Session ID[s].

TYPE: Union[List[str], str]

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.connect(ids="hjko98yghj")
>>> session.connect(ids=["hjko98yghj", "ikvp1jtp"])
Source code in canfar/sessions.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def connect(self, ids: list[str] | str) -> None:
    """Open session[s] in a web browser.

    Args:
        ids (Union[List[str], str]): Session ID[s].

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.connect(ids="hjko98yghj")
        >>> session.connect(ids=["hjko98yghj", "ikvp1jtp"])
    """
    if isinstance(ids, str):
        ids = [ids]
    info = self.info(ids)
    log.debug(info)
    for session in info:
        status: str = session.get("status", "unknown")
        if status != "Running":
            log.warning("Session %s is currently %s.", session["id"], status)
            log.warning("Please wait for the session to be ready.")
            continue
        connect_url = session.get("connectURL")
        if connect_url:
            open_new_tab(connect_url)

create 

create(
    name,
    image,
    cores=None,
    ram=None,
    kind="headless",
    gpu=None,
    cmd=None,
    args=None,
    env=None,
    replicas=1,
)

Launch a canfar session.

PARAMETER DESCRIPTION
name

A unique name for the session.

TYPE: str

image

Container image to use for the session.

TYPE: str

cores

Number of cores. Defaults to None, i.e. flexible mode.

TYPE: int DEFAULT: None

ram

Amount of RAM (GB). Defaults to None, i.e. flexible mode.

TYPE: int DEFAULT: None

kind

Type of canfar session. Defaults to "headless".

TYPE: str DEFAULT: 'headless'

gpu

Number of GPUs. Defaults to None.

TYPE: Optional[int] DEFAULT: None

cmd

Command to run. Defaults to None.

TYPE: Optional[str] DEFAULT: None

args

Arguments to the command. Defaults to None.

TYPE: Optional[str] DEFAULT: None

env

Environment variables to inject. Defaults to None.

TYPE: Optional[Dict[str, Any]] DEFAULT: None

replicas

Number of sessions to launch. Defaults to 1.

TYPE: int DEFAULT: 1

Notes
  • If cores and ram are not specified, the session will be created with flexible resource allocation of upto 8 cores and 32GB of RAM.
  • The name of the session suffixed with the replica number. eg. test-42 when replicas > 1.
  • Each container will have the following environment variables injected:
    • REPLICA_ID - The replica number
    • REPLICA_COUNT - The total number of replicas
RETURNS DESCRIPTION
list[str]

List[str]: A list of session IDs for the launched sessions.

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.create(
        name="test",
        image='images.canfar.net/skaha/terminal:1.1.1',
        cores=2,
        ram=8,
        gpu=1,
        kind="headless",
        cmd="env",
        env={"TEST": "test"},
        replicas=2,
    )
>>> ["hjko98yghj", "ikvp1jtp"]
Source code in canfar/sessions.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def create(
    self,
    name: str,
    image: str,
    cores: int | None = None,
    ram: int | None = None,
    kind: Kind = "headless",
    gpu: int | None = None,
    cmd: str | None = None,
    args: str | None = None,
    env: dict[str, Any] | None = None,
    replicas: int = 1,
) -> list[str]:
    """Launch a canfar session.

    Args:
        name (str): A unique name for the session.
        image (str): Container image to use for the session.
        cores (int, optional): Number of cores.
            Defaults to None, i.e. flexible mode.
        ram (int, optional): Amount of RAM (GB).
            Defaults to None, i.e. flexible mode.
        kind (str, optional): Type of canfar session. Defaults to "headless".
        gpu (Optional[int], optional): Number of GPUs. Defaults to None.
        cmd (Optional[str], optional): Command to run. Defaults to None.
        args (Optional[str], optional): Arguments to the command. Defaults to None.
        env (Optional[Dict[str, Any]], optional): Environment variables to inject.
            Defaults to None.
        replicas (int, optional): Number of sessions to launch. Defaults to 1.

    Notes:
        - If cores and ram are not specified, the session will be created with
          flexible resource allocation of upto 8 cores and 32GB of RAM.
        - The name of the session suffixed with the replica number. eg. test-42
          when replicas > 1.
        - Each container will have the following environment variables injected:
            * REPLICA_ID - The replica number
            * REPLICA_COUNT - The total number of replicas

    Returns:
        List[str]: A list of session IDs for the launched sessions.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.create(
                name="test",
                image='images.canfar.net/skaha/terminal:1.1.1',
                cores=2,
                ram=8,
                gpu=1,
                kind="headless",
                cmd="env",
                env={"TEST": "test"},
                replicas=2,
            )
        >>> ["hjko98yghj", "ikvp1jtp"]
    """
    payloads = build.create_parameters(
        name,
        image,
        cores,
        ram,
        kind,
        gpu,
        cmd,
        args,
        env,
        replicas,
    )
    results: list[str] = []
    log.debug("Creating %d %s session[s].", replicas, kind)
    for payload in payloads:
        try:
            response: Response = self.client.post(url="session", params=payload)
            results.append(response.text.rstrip("\r\n"))
        except HTTPError:
            err = f"Failed to create session with payload: {payload}"
            log.exception(err)
    return results

destroy 

destroy(ids)

Destroy canfar session[s].

PARAMETER DESCRIPTION
ids

Session ID[s].

TYPE: Union[str, List[str]]

RETURNS DESCRIPTION
dict[str, bool]

Dict[str, bool]: A dictionary of session IDs
dict[str, bool]

and a bool indicating if the session was destroyed.

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.destroy(id="hjko98yghj")
>>> session.destroy(id=["hjko98yghj", "ikvp1jtp"])
Source code in canfar/sessions.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
def destroy(self, ids: str | list[str]) -> dict[str, bool]:
    """Destroy canfar session[s].

    Args:
        ids (Union[str, List[str]]): Session ID[s].

    Returns:
        Dict[str, bool]: A dictionary of session IDs
        and a bool indicating if the session was destroyed.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.destroy(id="hjko98yghj")
        >>> session.destroy(id=["hjko98yghj", "ikvp1jtp"])
    """
    if isinstance(ids, str):
        ids = [ids]
    results: dict[str, bool] = {}
    for value in ids:
        try:
            self.client.delete(url=f"session/{value}")
            results[value] = True
        except HTTPError:
            msg = f"Failed to destroy session {value}"
            log.exception(msg)
            results[value] = False
    return results

destroy_with 

destroy_with(prefix, *, kind='headless', status='Completed')

Destroy session[s] matching a prefix or regex.

PARAMETER DESCRIPTION
prefix

Prefix to match. Treated literally unless regex meta-characters are found.

TYPE: str

kind

Type of session. Defaults to "headless".

TYPE: Kind DEFAULT: 'headless'

status

Status of the session. Defaults to "Completed".

TYPE: Status DEFAULT: 'Completed'

RETURNS DESCRIPTION
dict[str, bool]

Dict[str, bool]: A dictionary of session IDs
dict[str, bool]

and a bool indicating if the session was destroyed.
Notes
  • If the value contains regex metacharacters (e.g., .^$*+?{}[]()|), it is treated as a regex with :func:re.search.
  • Otherwise it is treated as a literal prefix (anchored with ^). This method is useful for destroying multiple sessions at once.

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.destroy_with(prefix="test")  # literal prefix
>>> session.destroy_with(prefix="desktop$")  # regex
Source code in canfar/sessions.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
def destroy_with(
    self,
    prefix: str,
    *,
    kind: Kind = "headless",
    status: Status = "Completed",
) -> dict[str, bool]:
    """Destroy session[s] matching a prefix or regex.

    Args:
        prefix (str): Prefix to match.
            Treated literally unless regex meta-characters are found.
        kind (Kind): Type of session. Defaults to "headless".
        status (Status): Status of the session. Defaults to "Completed".

    Returns:
        Dict[str, bool]: A dictionary of session IDs
        and a bool indicating if the session was destroyed.

    Notes:
        - If the value contains regex metacharacters (e.g., `.^$*+?{}[]()|`),
          it is treated as a regex with :func:`re.search`.
        - Otherwise it is treated as a literal prefix (anchored with `^`).
        This method is useful for destroying multiple sessions at once.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.destroy_with(prefix="test")  # literal prefix
        >>> session.destroy_with(prefix="desktop$")  # regex
    """
    meta = set(".^$*+?{}[]()|")
    has_meta = any(ch in meta for ch in prefix)
    try:
        if has_meta:
            log.info("destroy_with using regex pattern: %s", prefix)
            regex = re.compile(prefix)
        else:
            log.info("destroy_with using literal prefix: %s", prefix)
            regex = re.compile(rf"^{re.escape(prefix)}")
    except re.error as exc:
        msg = f"Invalid regex pattern '{prefix}': {exc}"
        log.exception(msg)
        raise ValueError(msg) from exc

    sessions = self.fetch(kind=kind, status=status)
    ids: list[str] = [
        session["id"] for session in sessions if regex.search(session["name"])
    ]
    return self.destroy(ids)

events 

events(ids, verbose=False)

Get deployment events for a session[s].

PARAMETER DESCRIPTION
ids

Session ID[s].

TYPE: Union[str, List[str]]

verbose

Print events to stdout. Defaults to False.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
list[dict[str, str]] | None

Optional[List[Dict[str, str]]]: A list of events for the session[s].
Notes

When verbose is True, the events will be printed to stdout only.

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.events(ids="hjko98yghj")
>>> session.events(ids=["hjko98yghj", "ikvp1jtp"])
Source code in canfar/sessions.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def events(
    self,
    ids: str | list[str],
    verbose: bool = False,
) -> list[dict[str, str]] | None:
    """Get deployment events for a session[s].

    Args:
        ids (Union[str, List[str]]): Session ID[s].
        verbose (bool, optional): Print events to stdout. Defaults to False.

    Returns:
        Optional[List[Dict[str, str]]]: A list of events for the session[s].

    Notes:
        When verbose is True, the events will be printed to stdout only.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.events(ids="hjko98yghj")
        >>> session.events(ids=["hjko98yghj", "ikvp1jtp"])
    """
    if isinstance(ids, str):
        ids = [ids]
    results: list[dict[str, str]] = []
    parameters: dict[str, str] = {"view": "events"}
    for value in ids:
        try:
            response: Response = self.client.get(
                url=f"session/{value}",
                params=parameters,
            )
            results.append({value: response.text})
        except HTTPError:
            err = f"Failed to fetch events for session {value}"
            log.exception(err)
    if verbose and results:
        for result in results:
            for key, value in result.items():
                log.info("Session ID: %s", key)
                log.info("\n %s", value)
    return results if not verbose else None

fetch 

fetch(kind=None, status=None, view=None)

Fetch open sessions for the user.

PARAMETER DESCRIPTION
kind

Session kind. Defaults to None.

TYPE: Kind | None DEFAULT: None

status

Session status. Defaults to None.

TYPE: Status | None DEFAULT: None

view

View leve. Defaults to None.

TYPE: View | None DEFAULT: None

RETURNS DESCRIPTION
list[dict[str, str]]

list[dict[str, str]]: Session[s] information.

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.fetch(kind="notebook")
[{'id': 'ikvp1jtp',
  'userid': 'username',
  'image': 'image-server/image/label:latest',
  'type': 'notebook',
  'status': 'Running',
  'name': 'example-notebook',
  'startTime': '2222-12-14T02:24:06Z',
  'connectURL': 'https://something.example.com/ikvp1jtp',
  'requestedRAM': '16G',
  'requestedCPUCores': '2',
  'requestedGPUCores': '<none>',
  'coresInUse': '0m',
  'ramInUse': '101Mi'}]
Source code in canfar/sessions.py 
38
39
40
41
42
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
69
70
71
72
73
74
75
def fetch(
    self,
    kind: Kind | None = None,
    status: Status | None = None,
    view: View | None = None,
) -> list[dict[str, str]]:
    """Fetch open sessions for the user.

    Args:
        kind (Kind | None, optional): Session kind. Defaults to None.
        status (Status | None, optional): Session status. Defaults to None.
        view (View | None, optional): View leve. Defaults to None.

    Returns:
        list[dict[str, str]]: Session[s] information.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.fetch(kind="notebook")
        [{'id': 'ikvp1jtp',
          'userid': 'username',
          'image': 'image-server/image/label:latest',
          'type': 'notebook',
          'status': 'Running',
          'name': 'example-notebook',
          'startTime': '2222-12-14T02:24:06Z',
          'connectURL': 'https://something.example.com/ikvp1jtp',
          'requestedRAM': '16G',
          'requestedCPUCores': '2',
          'requestedGPUCores': '<none>',
          'coresInUse': '0m',
          'ramInUse': '101Mi'}]
    """
    parameters: dict[str, Any] = build.fetch_parameters(kind, status, view)
    response: Response = self.client.get(url="session", params=parameters)
    data: list[dict[str, str]] = response.json()
    return data

info 

info(ids)

Get information about session[s].

PARAMETER DESCRIPTION
ids

Session ID[s].

TYPE: Union[List[str], str]

RETURNS DESCRIPTION
list[dict[str, Any]]

Dict[str, Any]: Session information.

Examples:

>>> session.info(ids="hjko98yghj")
>>> session.info(ids=["hjko98yghj", "ikvp1jtp"])
Source code in canfar/sessions.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
def info(self, ids: list[str] | str) -> list[dict[str, Any]]:
    """Get information about session[s].

    Args:
        ids (Union[List[str], str]): Session ID[s].

    Returns:
        Dict[str, Any]: Session information.

    Examples:
        >>> session.info(ids="hjko98yghj")
        >>> session.info(ids=["hjko98yghj", "ikvp1jtp"])
    """
    # Convert id to list if it is a string
    if isinstance(ids, str):
        ids = [ids]
    results: list[dict[str, Any]] = []
    for value in ids:
        try:
            response: Response = self.client.get(url=f"session/{value}")
            results.append(response.json())
        except HTTPError:
            err = f"failed to fetch session info for {value}"
            log.exception(err)
    return results

logs 

logs(ids, verbose=False)

Get logs from a session[s].

PARAMETER DESCRIPTION
ids

Session ID[s].

TYPE: Union[List[str], str]

verbose

Print logs to stdout. Defaults to False.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
dict[str, str] | None

Dict[str, str]: Logs in text/plain format.

Examples:

>>> session.logs(id="hjko98yghj")
>>> session.logs(id=["hjko98yghj", "ikvp1jtp"])
Source code in canfar/sessions.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def logs(
    self,
    ids: list[str] | str,
    verbose: bool = False,
) -> dict[str, str] | None:
    """Get logs from a session[s].

    Args:
        ids (Union[List[str], str]): Session ID[s].
        verbose (bool, optional): Print logs to stdout. Defaults to False.

    Returns:
        Dict[str, str]: Logs in text/plain format.

    Examples:
        >>> session.logs(id="hjko98yghj")
        >>> session.logs(id=["hjko98yghj", "ikvp1jtp"])
    """
    if isinstance(ids, str):
        ids = [ids]
    parameters: dict[str, str] = {"view": "logs"}
    results: dict[str, str] = {}

    for value in ids:
        try:
            response: Response = self.client.get(
                url=f"session/{value}",
                params=parameters,
            )
            results[value] = response.text
        except HTTPError:
            err = f"failed to fetch logs for session {value}"
            log.exception(err)

    if verbose:
        for key, value in results.items():
            log.info("Session ID: %s\n", key)
            log.info(value)
        return None

    return results

stats 

stats()

Get statistics for the entire platform.

RETURNS DESCRIPTION
dict[str, Any]

Dict[str, Any]: Cluster statistics.

Examples:

>>> from canfar.session import Session
>>> session = Session()
>>> session.stats()
{'cores': {'requestedCPUCores': 377,
 'coresAvailable': 960,
 'maxCores': {'cores': 32, 'withRam': '147Gi'}},
 'ram': {'maxRAM': {'ram': '226Gi', 'withCores': 32}}}
Source code in canfar/sessions.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def stats(self) -> dict[str, Any]:
    """Get statistics for the entire platform.

    Returns:
        Dict[str, Any]: Cluster statistics.

    Examples:
        >>> from canfar.session import Session
        >>> session = Session()
        >>> session.stats()
        {'cores': {'requestedCPUCores': 377,
         'coresAvailable': 960,
         'maxCores': {'cores': 32, 'withRam': '147Gi'}},
         'ram': {'maxRAM': {'ram': '226Gi', 'withCores': 32}}}
    """
    parameters = {"view": "stats"}
    response: Response = self.client.get("session", params=parameters)
    data: dict[str, Any] = response.json()
    return data