Session API

Session is the native synchronous Python client for user-owned Sessions on a Science Platform Server. It inherits the HTTP and credential behavior of HTTPClient.

Return shapes and failure behavior

The current interface has these return shapes. Check the installation and version guidance before using branch examples:

Method	Result
fetch(kind=None, status=None)	list[dict[str, str]]
create(...)	list[str] containing the accepted Session IDs
info(ids)	list[dict[str, Any]]
logs(ids, verbose=False)	dict[str, str], or None when verbose=True
events(ids, verbose=False)	list[dict[str, str]], or None when verbose=True
destroy(ids) / destroy_with(...)	dict[str, bool]
renew(ids)	dict[str, bool]
connect(ids)	None; opens ready Session URLs

By default, create() skips an individual launch after an HTTP or network failure and logs the failure without raising. If all requested launches fail, it returns []. Validation errors in the request are raised before the HTTP call. info, logs, events, destroy, and renew likewise log and leave out failed IDs (or report False); see Raise on failures to raise instead.

stats() returns the Science Platform Server's aggregate resource statistics as a dictionary.

logs(..., verbose=True) and events(..., verbose=True) route their results to the canfar.sessions logger and return None; configure application logging with canfar.configure_logging() if the output should be visible.

Create and manage a Session

from canfar.sessions import Session

with Session() as session:
    ids = session.create(
        name="my-analysis",
        image="images.canfar.net/skaha/astroml:latest",
        kind="notebook",
    )
    if ids:
        session.connect(ids)
        print(session.fetch(kind="notebook", status="Running"))
        print(session.info(ids))
        print(session.logs(ids))
        print(session.events(ids))
        print(session.destroy(ids))

When kind="headless", pass cmd, args, and env for the command. cores and ram request fixed resources; omitting them uses the server's flexible allocation policy. replicas requests multiple Sessions and the return value contains the successful IDs only.

from canfar.sessions import Session

with Session() as session:
    ids = session.create(
        name="batch",
        image="images.canfar.net/skaha/terminal:latest",
        kind="headless",
        cmd="python",
        args="/arc/projects/demo/run.py",
        replicas=3,
    )

replicas accepts 1 to 256; other ranges are in Session limits. Request GPUs with the gpu argument. The CLI option is --gpu as well, while the request model and the Server's API name the field gpus:

from canfar.sessions import Session

with Session() as session:
    ids = session.create(
        name="training",
        image="IMAGE_WITH_GPU_LIBRARIES",
        kind="headless",
        cmd="python",
        args="/arc/projects/demo/train.py",
        cores=8,
        ram=32,
        gpu=1,
    )

Use an image built for the Server's GPU stack, and confirm the device inside the Session with nvidia-smi.

Select Sessions for cleanup

destroy_with has a keyword-only filter contract:

session.destroy_with(
    "batch-",
    kind="headless",
    status="Completed",
)

Its signature is destroy_with(prefix, *, kind="headless", status="Completed", errors=None). A prefix is matched literally unless it contains regular-expression metacharacters; such a value is treated as a regular expression.

Renew an interactive Session

renew() resets the lifetime of interactive Sessions before they expire, as the Science Portal does. It returns True for each renewed ID:

with Session() as session:
    print(session.renew("hjko98yghj"))  # {"hjko98yghj": True}

Raise on failures

Set errors="raise" on the client, or pass it to one call, to raise SessionRequestError instead of returning partial results. Every request still runs first. The exception carries results (what the call would have returned) and errors, the httpx2.HTTPError for each failed Session ID, or for each 1-based replica number from create(). CANFAR_ERRORS=raise sets the client default; a per-call errors="ignore" overrides it. Authentication and validation errors raise either way.

from canfar.exceptions.session import SessionRequestError

with Session(errors="raise") as session:
    try:
        session.renew(["hjko98yghj", "ikvp1jtp"])
    except SessionRequestError as err:
        print(err.results)  # {"hjko98yghj": True, "ikvp1jtp": False}
        print(err.errors)  # {"ikvp1jtp": HTTPStatusError(...)}

To share one connection pool across clients, see Share a connection pool.

Bases: HTTPClient

CANFAR Session Management Client.

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

Examples:

>>> from canfar.sessions import Session
>>> session = Session(
        timeout=120,
        concurrency=100, # No effect on sync client
    )

Source code in canfar/sessions.py

class Session(HTTPClient):
    """CANFAR Session Management Client.

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

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

    errors: Literal["ignore", "raise"] = Field(
        "ignore", title="Error Policy", description=_ERRORS_DESCRIPTION
    )

    def fetch(
        self,
        kind: Kind | None = None,
        status: Status | 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.

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

        Examples:
            >>> from canfar.sessions 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)
        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, unchanged from the server.
                The optional ``gpu`` section reports GPU usage and capacity.

        Examples:
            >>> from canfar.sessions 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,
        *,
        errors: Literal["ignore", "raise"] | None = None,
    ) -> list[dict[str, Any]]:
        """Get information about session[s].

        Args:
            ids (Union[List[str], str]): Session ID[s].
            errors: Failure policy for this call. Defaults to the client's
                ``errors`` setting.

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

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

        Examples:
            >>> session.info(ids="hjko98yghj")
            >>> session.info(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: list[dict[str, Any]] = []
        failures: dict[str | int, HTTPError] = {}
        for value in ids:
            try:
                response: Response = self.client.get(url=_session_url(value))
                results.append(response.json())
            except HTTPError as err:
                failures[value] = err
                _task_result("failed to fetch session info for", value, err)
        _raise_failures("info", errors or self.errors, results, failures)
        return results

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

        Args:
            ids (Union[List[str], str]): Session ID[s].
            verbose: Send logs to the ``canfar.sessions`` logger and return None.
                Defaults to False, which returns the collected logs.
            errors: Failure policy for this call. Defaults to the client's
                ``errors`` setting.

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

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

        Examples:
            >>> session.logs(ids="hjko98yghj")
            >>> session.logs(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        parameters: dict[str, str] = {"view": "logs"}
        results: dict[str, str] = {}
        failures: dict[str | int, HTTPError] = {}

        for value in ids:
            try:
                response: Response = self.client.get(
                    url=_session_url(value),
                    params=parameters,
                )
                results[value] = response.text
            except HTTPError as err:
                failures[value] = err
                _task_result("failed to fetch logs for session", value, err)

        if verbose:
            for key, value in results.items():
                log.info("Session ID: %s\n", key)
                log.info(value)
        output = None if verbose else results
        _raise_failures("logs", errors or self.errors, output, failures)
        return output

    def create(  # noqa: PLR0917
        self,
        name: str | CreateRequest,
        image: str | None = None,
        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,
        *,
        errors: Literal["ignore", "raise"] | None = None,
    ) -> list[str]:
        """Launch a canfar session.

        Args:
            name: Domain request or a unique name for the Session.
            image: Container Image when ``name`` is a string.
            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.
            errors: Failure policy for this call. Defaults to the client's
                ``errors`` setting.

        Notes:
            - If cores and ram are not specified, the Session uses the Server's
              flexible resource allocation policy. Limits depend on the Server.
            - 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]: Session IDs for launched sessions. With ``errors="ignore"``,
                an attempt that fails with an HTTP or network error is omitted; if
                all attempts fail, returns an empty list.

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any replica failed.
                Its ``errors`` are keyed by 1-based replica number.

        Examples:
            >>> from canfar.sessions 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] = []
        failures: dict[str | int, HTTPError] = {}
        session_kind = name.kind if isinstance(name, CreateRequest) else kind
        log.debug("Creating %d %s session[s].", len(payloads), session_kind)
        for replica, payload in enumerate(payloads, start=1):
            try:
                response: Response = self.client.post(url="session", params=payload)
                results.append(_response_session_id(response))
            except HTTPError as err:
                failures[replica] = err
                _task_result(
                    "Failed to create session",
                    f"replica {replica}/{len(payloads)}",
                    err,
                )
        _raise_failures("create", errors or self.errors, results, failures)
        return results

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

        Args:
            ids (Union[str, List[str]]): Session ID[s].
            verbose: Send events to the ``canfar.sessions`` logger and return None.
                Defaults to False, which returns the collected events.
            errors: Failure policy for this call. Defaults to the client's
                ``errors`` setting.

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

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

        Notes:
            Configure application logging to display verbose events. Their output
            follows the configured handlers, rather than printing to stdout.

        Examples:
            >>> from canfar.sessions import Session
            >>> session = Session()
            >>> session.events(ids="hjko98yghj")
            >>> session.events(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: list[dict[str, str]] = []
        failures: dict[str | int, HTTPError] = {}
        parameters: dict[str, str] = {"view": "events"}
        for value in ids:
            try:
                response: Response = self.client.get(
                    url=_session_url(value),
                    params=parameters,
                )
                results.append({value: response.text})
            except HTTPError as err:
                failures[value] = err
                _task_result("Failed to fetch events for session", value, 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)
        output = results if not verbose else None
        _raise_failures("events", errors or self.errors, output, failures)
        return output

    def destroy(
        self,
        ids: str | list[str],
        *,
        errors: Literal["ignore", "raise"] | None = None,
    ) -> dict[str, bool]:
        """Destroy canfar session[s].

        Args:
            ids (Union[str, List[str]]): Session ID[s].
            errors: Failure policy for this call. Defaults to the client's
                ``errors`` setting.

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

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

        Examples:
            >>> from canfar.sessions import Session
            >>> session = Session()
            >>> session.destroy(ids="hjko98yghj")
            >>> session.destroy(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: dict[str, bool] = {}
        failures: dict[str | int, HTTPError] = {}
        for value in ids:
            try:
                self.client.delete(url=_session_url(value))
                results[value] = True
            except HTTPError as err:
                failures[value] = err
                results[value] = _destroy_failure(value)
        _raise_failures("destroy", errors or self.errors, results, failures)
        return results

    def renew(
        self,
        ids: str | list[str],
        *,
        errors: Literal["ignore", "raise"] | None = None,
    ) -> dict[str, bool]:
        """Renew interactive session[s], resetting their lifetime.

        Args:
            ids (Union[str, List[str]]): Session ID[s].
            errors: Failure policy for this call. Defaults to the client's
                ``errors`` setting.

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

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

        Examples:
            >>> from canfar.sessions import Session
            >>> session = Session()
            >>> session.renew(ids="hjko98yghj")
            >>> session.renew(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: dict[str, bool] = {}
        failures: dict[str | int, HTTPError] = {}
        for value in ids:
            try:
                self.client.post(url=_session_url(value), params={"action": "renew"})
                results[value] = True
            except HTTPError as err:
                failures[value] = err
                results[value] = _renew(value)
        _raise_failures("renew", errors or self.errors, results, failures)
        return results

    def destroy_with(
        self,
        prefix: str,
        *,
        kind: Kind = "headless",
        status: Status = "Completed",
        errors: Literal["ignore", "raise"] | None = None,
    ) -> 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".
            errors: Failure policy for the deletions. Defaults to the client's
                ``errors`` setting.

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

        Raises:
            SessionRequestError: If ``errors`` is ``"raise"`` and any deletion failed.

        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.sessions import Session
            >>> session = Session()
            >>> session.destroy_with(prefix="test")  # literal prefix
            >>> session.destroy_with(prefix="desktop$")  # regex
        """
        regex = _session_name_pattern(prefix)
        sessions = self.fetch(kind=kind, status=status)
        return self.destroy(_matching_session_ids(sessions, regex), errors=errors)

    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.sessions import Session
            >>> session = Session()
            >>> session.connect(ids="hjko98yghj")
            >>> session.connect(ids=["hjko98yghj", "ikvp1jtp"])
        """
        info = self.info(_ids(ids))
        log.debug(info)
        for session in info:
            status: str = session.get("status", "unknown")
            url = connection_url(session)
            if url is None and status != "Running":
                log.warning("Session %s is currently %s.", session["id"], status)
                log.warning("Please wait for the session to be ready.")
                continue
            if url is not None:
                open_new_tab(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.sessions import Session
>>> session = Session()
>>> session.connect(ids="hjko98yghj")
>>> session.connect(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

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.sessions import Session
        >>> session = Session()
        >>> session.connect(ids="hjko98yghj")
        >>> session.connect(ids=["hjko98yghj", "ikvp1jtp"])
    """
    info = self.info(_ids(ids))
    log.debug(info)
    for session in info:
        status: str = session.get("status", "unknown")
        url = connection_url(session)
        if url is None and status != "Running":
            log.warning("Session %s is currently %s.", session["id"], status)
            log.warning("Please wait for the session to be ready.")
            continue
        if url is not None:
            open_new_tab(url)

create

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

Launch a canfar session.

PARAMETER	DESCRIPTION
name	Domain request or a unique name for the Session. TYPE: str | CreateRequest
image	Container Image when name is a string. TYPE: str | None DEFAULT: None
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
errors	Failure policy for this call. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

Notes

• If cores and ram are not specified, the Session uses the Server's flexible resource allocation policy. Limits depend on the Server.
• 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]: Session IDs for launched sessions. With errors="ignore", an attempt that fails with an HTTP or network error is omitted; if all attempts fail, returns an empty list.

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any replica failed. Its errors are keyed by 1-based replica number.

Examples:

>>> from canfar.sessions 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

def create(  # noqa: PLR0917
    self,
    name: str | CreateRequest,
    image: str | None = None,
    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,
    *,
    errors: Literal["ignore", "raise"] | None = None,
) -> list[str]:
    """Launch a canfar session.

    Args:
        name: Domain request or a unique name for the Session.
        image: Container Image when ``name`` is a string.
        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.
        errors: Failure policy for this call. Defaults to the client's
            ``errors`` setting.

    Notes:
        - If cores and ram are not specified, the Session uses the Server's
          flexible resource allocation policy. Limits depend on the Server.
        - 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]: Session IDs for launched sessions. With ``errors="ignore"``,
            an attempt that fails with an HTTP or network error is omitted; if
            all attempts fail, returns an empty list.

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any replica failed.
            Its ``errors`` are keyed by 1-based replica number.

    Examples:
        >>> from canfar.sessions 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] = []
    failures: dict[str | int, HTTPError] = {}
    session_kind = name.kind if isinstance(name, CreateRequest) else kind
    log.debug("Creating %d %s session[s].", len(payloads), session_kind)
    for replica, payload in enumerate(payloads, start=1):
        try:
            response: Response = self.client.post(url="session", params=payload)
            results.append(_response_session_id(response))
        except HTTPError as err:
            failures[replica] = err
            _task_result(
                "Failed to create session",
                f"replica {replica}/{len(payloads)}",
                err,
            )
    _raise_failures("create", errors or self.errors, results, failures)
    return results

destroy

destroy(ids, *, errors=None)

Destroy canfar session[s].

PARAMETER	DESCRIPTION
ids	Session ID[s]. TYPE: Union[str, List[str]]
errors	Failure policy for this call. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

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.

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any request failed.

Examples:

>>> from canfar.sessions import Session
>>> session = Session()
>>> session.destroy(ids="hjko98yghj")
>>> session.destroy(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

def destroy(
    self,
    ids: str | list[str],
    *,
    errors: Literal["ignore", "raise"] | None = None,
) -> dict[str, bool]:
    """Destroy canfar session[s].

    Args:
        ids (Union[str, List[str]]): Session ID[s].
        errors: Failure policy for this call. Defaults to the client's
            ``errors`` setting.

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

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

    Examples:
        >>> from canfar.sessions import Session
        >>> session = Session()
        >>> session.destroy(ids="hjko98yghj")
        >>> session.destroy(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: dict[str, bool] = {}
    failures: dict[str | int, HTTPError] = {}
    for value in ids:
        try:
            self.client.delete(url=_session_url(value))
            results[value] = True
        except HTTPError as err:
            failures[value] = err
            results[value] = _destroy_failure(value)
    _raise_failures("destroy", errors or self.errors, results, failures)
    return results

destroy_with

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

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'
errors	Failure policy for the deletions. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

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.

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any deletion failed.

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.sessions import Session
>>> session = Session()
>>> session.destroy_with(prefix="test")  # literal prefix
>>> session.destroy_with(prefix="desktop$")  # regex

Source code in canfar/sessions.py

def destroy_with(
    self,
    prefix: str,
    *,
    kind: Kind = "headless",
    status: Status = "Completed",
    errors: Literal["ignore", "raise"] | None = None,
) -> 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".
        errors: Failure policy for the deletions. Defaults to the client's
            ``errors`` setting.

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

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any deletion failed.

    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.sessions import Session
        >>> session = Session()
        >>> session.destroy_with(prefix="test")  # literal prefix
        >>> session.destroy_with(prefix="desktop$")  # regex
    """
    regex = _session_name_pattern(prefix)
    sessions = self.fetch(kind=kind, status=status)
    return self.destroy(_matching_session_ids(sessions, regex), errors=errors)

events

events(ids, verbose=False, *, errors=None)

Get deployment events for a session[s].

PARAMETER	DESCRIPTION
ids	Session ID[s]. TYPE: Union[str, List[str]]
verbose	Send events to the canfar.sessions logger and return None. Defaults to False, which returns the collected events. TYPE: bool DEFAULT: False
errors	Failure policy for this call. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

RETURNS	DESCRIPTION
list[dict[str, str]] | None	Optional[List[Dict[str, str]]]: A list of events for the session[s].

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any request failed.

Notes

Configure application logging to display verbose events. Their output follows the configured handlers, rather than printing to stdout.

Examples:

>>> from canfar.sessions import Session
>>> session = Session()
>>> session.events(ids="hjko98yghj")
>>> session.events(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

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

    Args:
        ids (Union[str, List[str]]): Session ID[s].
        verbose: Send events to the ``canfar.sessions`` logger and return None.
            Defaults to False, which returns the collected events.
        errors: Failure policy for this call. Defaults to the client's
            ``errors`` setting.

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

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

    Notes:
        Configure application logging to display verbose events. Their output
        follows the configured handlers, rather than printing to stdout.

    Examples:
        >>> from canfar.sessions import Session
        >>> session = Session()
        >>> session.events(ids="hjko98yghj")
        >>> session.events(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: list[dict[str, str]] = []
    failures: dict[str | int, HTTPError] = {}
    parameters: dict[str, str] = {"view": "events"}
    for value in ids:
        try:
            response: Response = self.client.get(
                url=_session_url(value),
                params=parameters,
            )
            results.append({value: response.text})
        except HTTPError as err:
            failures[value] = err
            _task_result("Failed to fetch events for session", value, 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)
    output = results if not verbose else None
    _raise_failures("events", errors or self.errors, output, failures)
    return output

fetch

fetch(kind=None, status=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

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

Examples:

>>> from canfar.sessions 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

def fetch(
    self,
    kind: Kind | None = None,
    status: Status | 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.

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

    Examples:
        >>> from canfar.sessions 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)
    response: Response = self.client.get(url="session", params=parameters)
    data: list[dict[str, str]] = response.json()
    return data

info

info(ids, *, errors=None)

Get information about session[s].

PARAMETER	DESCRIPTION
ids	Session ID[s]. TYPE: Union[List[str], str]
errors	Failure policy for this call. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

RETURNS	DESCRIPTION
list[dict[str, Any]]	list[dict[str, Any]]: Session information.

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any request failed.

Examples:

>>> session.info(ids="hjko98yghj")
>>> session.info(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

def info(
    self,
    ids: list[str] | str,
    *,
    errors: Literal["ignore", "raise"] | None = None,
) -> list[dict[str, Any]]:
    """Get information about session[s].

    Args:
        ids (Union[List[str], str]): Session ID[s].
        errors: Failure policy for this call. Defaults to the client's
            ``errors`` setting.

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

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

    Examples:
        >>> session.info(ids="hjko98yghj")
        >>> session.info(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: list[dict[str, Any]] = []
    failures: dict[str | int, HTTPError] = {}
    for value in ids:
        try:
            response: Response = self.client.get(url=_session_url(value))
            results.append(response.json())
        except HTTPError as err:
            failures[value] = err
            _task_result("failed to fetch session info for", value, err)
    _raise_failures("info", errors or self.errors, results, failures)
    return results

logs

logs(ids, verbose=False, *, errors=None)

Get logs from a session[s].

PARAMETER	DESCRIPTION
ids	Session ID[s]. TYPE: Union[List[str], str]
verbose	Send logs to the canfar.sessions logger and return None. Defaults to False, which returns the collected logs. TYPE: bool DEFAULT: False
errors	Failure policy for this call. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, str] | None	Dict[str, str]: Logs in text/plain format.

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any request failed.

Examples:

>>> session.logs(ids="hjko98yghj")
>>> session.logs(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

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

    Args:
        ids (Union[List[str], str]): Session ID[s].
        verbose: Send logs to the ``canfar.sessions`` logger and return None.
            Defaults to False, which returns the collected logs.
        errors: Failure policy for this call. Defaults to the client's
            ``errors`` setting.

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

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

    Examples:
        >>> session.logs(ids="hjko98yghj")
        >>> session.logs(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    parameters: dict[str, str] = {"view": "logs"}
    results: dict[str, str] = {}
    failures: dict[str | int, HTTPError] = {}

    for value in ids:
        try:
            response: Response = self.client.get(
                url=_session_url(value),
                params=parameters,
            )
            results[value] = response.text
        except HTTPError as err:
            failures[value] = err
            _task_result("failed to fetch logs for session", value, err)

    if verbose:
        for key, value in results.items():
            log.info("Session ID: %s\n", key)
            log.info(value)
    output = None if verbose else results
    _raise_failures("logs", errors or self.errors, output, failures)
    return output

renew

renew(ids, *, errors=None)

Renew interactive session[s], resetting their lifetime.

PARAMETER	DESCRIPTION
ids	Session ID[s]. TYPE: Union[str, List[str]]
errors	Failure policy for this call. Defaults to the client's errors setting. TYPE: Literal['ignore', 'raise'] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, bool]	Dict[str, bool]: A dictionary of session IDs
dict[str, bool]	and a bool indicating if the session was renewed.

RAISES	DESCRIPTION
SessionRequestError	If errors is "raise" and any request failed.

Examples:

>>> from canfar.sessions import Session
>>> session = Session()
>>> session.renew(ids="hjko98yghj")
>>> session.renew(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

def renew(
    self,
    ids: str | list[str],
    *,
    errors: Literal["ignore", "raise"] | None = None,
) -> dict[str, bool]:
    """Renew interactive session[s], resetting their lifetime.

    Args:
        ids (Union[str, List[str]]): Session ID[s].
        errors: Failure policy for this call. Defaults to the client's
            ``errors`` setting.

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

    Raises:
        SessionRequestError: If ``errors`` is ``"raise"`` and any request failed.

    Examples:
        >>> from canfar.sessions import Session
        >>> session = Session()
        >>> session.renew(ids="hjko98yghj")
        >>> session.renew(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: dict[str, bool] = {}
    failures: dict[str | int, HTTPError] = {}
    for value in ids:
        try:
            self.client.post(url=_session_url(value), params={"action": "renew"})
            results[value] = True
        except HTTPError as err:
            failures[value] = err
            results[value] = _renew(value)
    _raise_failures("renew", errors or self.errors, results, failures)
    return results

stats

stats()

Get statistics for the entire platform.

RETURNS	DESCRIPTION
dict[str, Any]	Dict[str, Any]: Cluster statistics, unchanged from the server. The optional gpu section reports GPU usage and capacity.

Examples:

>>> from canfar.sessions 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

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

    Returns:
        Dict[str, Any]: Cluster statistics, unchanged from the server.
            The optional ``gpu`` section reports GPU usage and capacity.

    Examples:
        >>> from canfar.sessions 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