Asynchronous Sessions¶

AsyncSession is the native asynchronous counterpart to Session. It uses a native httpx2.AsyncClient; it is not a synchronous client wrapped in an event loop. Use it inside an existing async application and close it with async with.

Return shapes and parity¶

The async methods preserve the synchronous result contracts:

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

By default, create() omits failed launches and returns [] when every launch fails; pass errors="raise" to raise instead (see Raise on failures). Invalid request values raise before the HTTP request. Verbose logs and events are sent to the canfar.sessions logger and return None.

stats() returns aggregate Science Platform Server resource statistics.

Async workflow¶

from canfar.sessions import AsyncSession

async def main() -> None:
    async with AsyncSession() as session:
        ids = await session.create(
            name="async-analysis",
            image="images.canfar.net/skaha/astroml:latest",
            kind="notebook",
        )
        if ids:
            await session.connect(ids)
            print(await session.fetch(kind="notebook", status="Running"))
            print(await session.info(ids))
            print(await session.logs(ids))
            print(await session.events(ids))
            print(await session.destroy(ids))

For a headless Session, add cmd, args, and env. cores and ram request fixed resources; omitting them uses the Science Platform Server's flexible allocation policy.

Select Sessions for cleanup¶

destroy_with is keyword-only after prefix:

async with AsyncSession() as session:
    result = await session.destroy_with(
        "batch-",
        kind="headless",
        status="Completed",
    )

Its signature is destroy_with(prefix, *, kind="headless", status="Completed", errors=None). Literal prefixes are anchored at the beginning; prefixes containing regular expression metacharacters are treated as regular expressions.

Renew an interactive Session¶

renew() resets the lifetime of interactive Sessions before they expire. Renewals run concurrently:

async with AsyncSession() as session:
    print(await session.renew(["hjko98yghj", "ikvp1jtp"]))

Raise on failures¶

errors="raise" behaves as in the synchronous client: every request runs, then SessionRequestError carries results and the per-ID (or per-replica) errors:

from canfar.exceptions.session import SessionRequestError

async with AsyncSession(errors="raise") as session:
    try:
        await session.renew(["hjko98yghj", "ikvp1jtp"])
    except SessionRequestError as err:
        print(err.results, err.errors)

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

Bases: HTTPClient

Asynchronous CANFAR Session Management Client.

This class provides methods to manage sessions in the system, including fetching session details, creating new sessions, retrieving logs, renewing and destroying existing sessions.

This class is a subclass of the HTTPClient class and inherits its attributes and methods.

Examples:

>>> from canfar.sessions import AsyncSession
>>> session = AsyncSession(
        url="https://ws-uv.canfar.net/skaha",
        token="token",
        timeout=30,
        concurrency=100,
    )

Source code in canfar/sessions.py

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

    This class provides methods to manage sessions in the system,
    including fetching session details, creating new sessions,
    retrieving logs, renewing and destroying existing sessions.

    This class is a subclass of the `HTTPClient` class and inherits its
    attributes and methods.

    Examples:
        >>> from canfar.sessions import AsyncSession
        >>> session = AsyncSession(
                url="https://ws-uv.canfar.net/skaha",
                token="token",
                timeout=30,
                concurrency=100,
            )
    """

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

    async def fetch(
        self,
        kind: Kind | None = None,
        status: Status | None = None,
    ) -> list[dict[str, str]]:
        """List 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: Sessions information.

        Examples:
            >>> from canfar.sessions import AsyncSession
            >>> session = AsyncSession()
            >>> await session.fetch(kind="notebook")
            [{'id': 'vl91sfzz',
            'userid': 'brars',
            'runAsUID': '166169204',
            'runAsGID': '166169204',
            'supplementalGroups': [34241,
            34337,
            35124,
            36227,
            1902365706,
            1454823273,
            1025424273],
            'appid': '<none>',
            'image': 'image-server/repo/image:version',
            'type': 'notebook',
            'status': 'Running',
            'name': 'notebook1',
            'startTime': '2025-03-05T21:48:29Z',
            'expiryTime': '2025-03-09T21:48:29Z',
            'connectURL': 'https://canfar.net/session/notebook/some/url',
            'requestedRAM': '8G',
            'requestedCPUCores': '2',
            'requestedGPUCores': '0',
            'ramInUse': '<none>',
            'gpuRAMInUse': '<none>',
            'cpuCoresInUse': '<none>',
            'gpuUtilization': '<none>'}]
        """
        parameters: dict[str, Any] = build.fetch_parameters(kind, status)
        response: Response = await self.asynclient.get(url="session", params=parameters)
        data: list[dict[str, str]] = response.json()
        return data

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

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

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

    async 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:
            >>> from canfar.sessions import AsyncSession
            >>> session = AsyncSession()
            >>> await session.info(ids="hjko98yghj")
            >>> await session.info(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: list[dict[str, Any]] = []
        failures: dict[str | int, HTTPError] = {}

        async def request(value: str) -> dict[str, Any]:
            response = await self.asynclient.get(url=_session_url(value))
            data: dict[str, Any] = response.json()
            return data

        tasks = [request(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for value, reply in zip(ids, responses, strict=True):
            if isinstance(reply, HTTPError):
                failures[value] = reply
            result = _task_result("failed to fetch session info for", value, reply)
            if isinstance(result, dict):
                results.append(result)
        log.debug("Session info records collected: %s", results)
        _raise_failures("info", errors or self.errors, results, failures)
        return results

    async 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:
            >>> from canfar.sessions import AsyncSession
            >>> session = AsyncSession()
            >>> await session.logs(ids="hjko98yghj")
            >>> await session.logs(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        parameters: dict[str, str] = {"view": "logs"}
        results: dict[str, str] = {}
        failures: dict[str | int, HTTPError] = {}

        async def request(value: str) -> tuple[str, str]:
            response = await self.asynclient.get(
                url=_session_url(value),
                params=parameters,
            )
            return value, response.text

        tasks = [request(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for value, reply in zip(ids, responses, strict=True):
            if isinstance(reply, HTTPError):
                failures[value] = reply
            result = _task_result("failed to fetch logs for session", value, reply)
            if isinstance(result, tuple):
                results[result[0]] = result[1]

        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

    async 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 AsyncSession
            >>> session = AsyncSession()
            >>> await 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: list[list[tuple[str, Any]]] = build.create_parameters(
            name,
            image,
            cores,
            ram,
            kind,
            gpu,
            cmd,
            args,
            env,
            replicas,
        )
        results: list[str] = []
        failures: dict[str | int, HTTPError] = {}

        async def request_session(parameters: list[tuple[str, Any]]) -> str:
            response = await self.asynclient.post(url="session", params=parameters)
            return _response_session_id(response)

        tasks = [request_session(payload) for payload in payloads]
        session_kind = name.kind if isinstance(name, CreateRequest) else kind
        msg = f"Creating {len(payloads)} {session_kind} session[s]."
        log.debug(msg)
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for replica, reply in enumerate(responses, start=1):
            if isinstance(reply, HTTPError):
                failures[replica] = reply
            result = _task_result(
                "Failed to create session",
                f"replica {replica}/{len(payloads)}",
                reply,
            )
            if isinstance(result, str):
                results.append(result)
        log.debug("Session IDs collected from create: %s", results)
        _raise_failures("create", errors or self.errors, results, failures)
        return results

    async 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 AsyncSession
            >>> session = AsyncSession()
            >>> await session.events(ids="hjko98yghj")
            >>> await session.events(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: list[dict[str, str]] = []
        failures: dict[str | int, HTTPError] = {}
        parameters: dict[str, str] = {"view": "events"}

        async def request(value: str) -> dict[str, str]:
            response = await self.asynclient.get(
                url=_session_url(value),
                params=parameters,
            )
            return {value: response.text}

        tasks = [request(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for value, reply in zip(ids, responses, strict=True):
            if isinstance(reply, HTTPError):
                failures[value] = reply
            result = _task_result(
                "Failed to fetch events for session",
                value,
                reply,
            )
            if isinstance(result, dict):
                results.append(result)

        if verbose and results:
            for result in results:
                for key, value in result.items():
                    log.info("Session ID: %s", key)
                    log.info(value)
        log.debug("Session events collected: %s", results)
        output = results if not verbose else None
        _raise_failures("events", errors or self.errors, output, failures)
        return output

    async def destroy(
        self,
        ids: str | list[str],
        *,
        errors: Literal["ignore", "raise"] | None = None,
    ) -> dict[str, bool]:
        """Destroy 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 AsyncSession
            >>> session = AsyncSession()
            >>> await session.destroy(ids="hjko98yghj")
            >>> await session.destroy(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: dict[str, bool] = {}
        failures: dict[str | int, HTTPError] = {}

        async def request(value: str) -> tuple[str, HTTPError | None]:
            try:
                await self.asynclient.delete(url=_session_url(value))
            except HTTPError as err:
                _destroy_failure(value, err)
                return value, err
            return value, None

        tasks = [request(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            _raise_authentication_failure(reply)
            if isinstance(reply, tuple):
                value, failure = reply
                results[value] = failure is None
                if failure is not None:
                    failures[value] = failure
        log.debug(results)
        _raise_failures("destroy", errors or self.errors, results, failures)
        return results

    async 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 AsyncSession
            >>> session = AsyncSession()
            >>> await session.renew(ids="hjko98yghj")
            >>> await session.renew(ids=["hjko98yghj", "ikvp1jtp"])
        """
        ids = _ids(ids)
        results: dict[str, bool] = {}
        failures: dict[str | int, HTTPError] = {}

        async def request(value: str) -> tuple[str, HTTPError | None]:
            try:
                await self.asynclient.post(
                    url=_session_url(value), params={"action": "renew"}
                )
            except HTTPError as err:
                _renew(value, err)
                return value, err
            return value, None

        tasks = [request(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            _raise_authentication_failure(reply)
            if isinstance(reply, tuple):
                value, failure = reply
                results[value] = failure is None
                if failure is not None:
                    failures[value] = failure
        log.debug(results)
        _raise_failures("renew", errors or self.errors, results, failures)
        return results

    async 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 pattern.

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

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

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

        Examples:
            >>> from canfar.sessions import AsyncSession
            >>> session = AsyncSession()
            >>> await session.connect(ids="hjko98yghj")
            >>> await session.connect(ids=["hjko98yghj", "ikvp1jtp"])
        """
        info = await 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 `async` ¶

connect(ids)

Connect to a session[s] in a web browser.

PARAMETER	DESCRIPTION
`ids`	Session ID[s]. TYPE: `Union[List[str], str]`

Examples:

>>> from canfar.sessions import AsyncSession
>>> session = AsyncSession()
>>> await session.connect(ids="hjko98yghj")
>>> await session.connect(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

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

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

    Examples:
        >>> from canfar.sessions import AsyncSession
        >>> session = AsyncSession()
        >>> await session.connect(ids="hjko98yghj")
        >>> await session.connect(ids=["hjko98yghj", "ikvp1jtp"])
    """
    info = await 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 `async` ¶

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 AsyncSession
>>> session = AsyncSession()
>>> await 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

async 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 AsyncSession
        >>> session = AsyncSession()
        >>> await 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: list[list[tuple[str, Any]]] = build.create_parameters(
        name,
        image,
        cores,
        ram,
        kind,
        gpu,
        cmd,
        args,
        env,
        replicas,
    )
    results: list[str] = []
    failures: dict[str | int, HTTPError] = {}

    async def request_session(parameters: list[tuple[str, Any]]) -> str:
        response = await self.asynclient.post(url="session", params=parameters)
        return _response_session_id(response)

    tasks = [request_session(payload) for payload in payloads]
    session_kind = name.kind if isinstance(name, CreateRequest) else kind
    msg = f"Creating {len(payloads)} {session_kind} session[s]."
    log.debug(msg)
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for replica, reply in enumerate(responses, start=1):
        if isinstance(reply, HTTPError):
            failures[replica] = reply
        result = _task_result(
            "Failed to create session",
            f"replica {replica}/{len(payloads)}",
            reply,
        )
        if isinstance(result, str):
            results.append(result)
    log.debug("Session IDs collected from create: %s", results)
    _raise_failures("create", errors or self.errors, results, failures)
    return results

destroy `async` ¶

destroy(ids, *, errors=None)

Destroy 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 AsyncSession
>>> session = AsyncSession()
>>> await session.destroy(ids="hjko98yghj")
>>> await session.destroy(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

async def destroy(
    self,
    ids: str | list[str],
    *,
    errors: Literal["ignore", "raise"] | None = None,
) -> dict[str, bool]:
    """Destroy 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 AsyncSession
        >>> session = AsyncSession()
        >>> await session.destroy(ids="hjko98yghj")
        >>> await session.destroy(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: dict[str, bool] = {}
    failures: dict[str | int, HTTPError] = {}

    async def request(value: str) -> tuple[str, HTTPError | None]:
        try:
            await self.asynclient.delete(url=_session_url(value))
        except HTTPError as err:
            _destroy_failure(value, err)
            return value, err
        return value, None

    tasks = [request(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        _raise_authentication_failure(reply)
        if isinstance(reply, tuple):
            value, failure = reply
            results[value] = failure is None
            if failure is not None:
                failures[value] = failure
    log.debug(results)
    _raise_failures("destroy", errors or self.errors, results, failures)
    return results

destroy_with `async` ¶

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

Destroy session[s] matching a prefix or regex pattern.

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

Source code in canfar/sessions.py

async 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 pattern.

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

events `async` ¶

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 AsyncSession
>>> session = AsyncSession()
>>> await session.events(ids="hjko98yghj")
>>> await session.events(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

async 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 AsyncSession
        >>> session = AsyncSession()
        >>> await session.events(ids="hjko98yghj")
        >>> await session.events(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: list[dict[str, str]] = []
    failures: dict[str | int, HTTPError] = {}
    parameters: dict[str, str] = {"view": "events"}

    async def request(value: str) -> dict[str, str]:
        response = await self.asynclient.get(
            url=_session_url(value),
            params=parameters,
        )
        return {value: response.text}

    tasks = [request(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for value, reply in zip(ids, responses, strict=True):
        if isinstance(reply, HTTPError):
            failures[value] = reply
        result = _task_result(
            "Failed to fetch events for session",
            value,
            reply,
        )
        if isinstance(result, dict):
            results.append(result)

    if verbose and results:
        for result in results:
            for key, value in result.items():
                log.info("Session ID: %s", key)
                log.info(value)
    log.debug("Session events collected: %s", results)
    output = results if not verbose else None
    _raise_failures("events", errors or self.errors, output, failures)
    return output

fetch `async` ¶

fetch(kind=None, status=None)

List 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`	Sessions information. TYPE: `list[dict[str, str]]`

Examples:

>>> from canfar.sessions import AsyncSession
>>> session = AsyncSession()
>>> await session.fetch(kind="notebook")
[{'id': 'vl91sfzz',
'userid': 'brars',
'runAsUID': '166169204',
'runAsGID': '166169204',
'supplementalGroups': [34241,
34337,
35124,
36227,
1902365706,
1454823273,
1025424273],
'appid': '<none>',
'image': 'image-server/repo/image:version',
'type': 'notebook',
'status': 'Running',
'name': 'notebook1',
'startTime': '2025-03-05T21:48:29Z',
'expiryTime': '2025-03-09T21:48:29Z',
'connectURL': 'https://canfar.net/session/notebook/some/url',
'requestedRAM': '8G',
'requestedCPUCores': '2',
'requestedGPUCores': '0',
'ramInUse': '<none>',
'gpuRAMInUse': '<none>',
'cpuCoresInUse': '<none>',
'gpuUtilization': '<none>'}]

Source code in canfar/sessions.py

async def fetch(
    self,
    kind: Kind | None = None,
    status: Status | None = None,
) -> list[dict[str, str]]:
    """List 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: Sessions information.

    Examples:
        >>> from canfar.sessions import AsyncSession
        >>> session = AsyncSession()
        >>> await session.fetch(kind="notebook")
        [{'id': 'vl91sfzz',
        'userid': 'brars',
        'runAsUID': '166169204',
        'runAsGID': '166169204',
        'supplementalGroups': [34241,
        34337,
        35124,
        36227,
        1902365706,
        1454823273,
        1025424273],
        'appid': '<none>',
        'image': 'image-server/repo/image:version',
        'type': 'notebook',
        'status': 'Running',
        'name': 'notebook1',
        'startTime': '2025-03-05T21:48:29Z',
        'expiryTime': '2025-03-09T21:48:29Z',
        'connectURL': 'https://canfar.net/session/notebook/some/url',
        'requestedRAM': '8G',
        'requestedCPUCores': '2',
        'requestedGPUCores': '0',
        'ramInUse': '<none>',
        'gpuRAMInUse': '<none>',
        'cpuCoresInUse': '<none>',
        'gpuUtilization': '<none>'}]
    """
    parameters: dict[str, Any] = build.fetch_parameters(kind, status)
    response: Response = await self.asynclient.get(url="session", params=parameters)
    data: list[dict[str, str]] = response.json()
    return data

info `async` ¶

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:

>>> from canfar.sessions import AsyncSession
>>> session = AsyncSession()
>>> await session.info(ids="hjko98yghj")
>>> await session.info(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

async 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:
        >>> from canfar.sessions import AsyncSession
        >>> session = AsyncSession()
        >>> await session.info(ids="hjko98yghj")
        >>> await session.info(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: list[dict[str, Any]] = []
    failures: dict[str | int, HTTPError] = {}

    async def request(value: str) -> dict[str, Any]:
        response = await self.asynclient.get(url=_session_url(value))
        data: dict[str, Any] = response.json()
        return data

    tasks = [request(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for value, reply in zip(ids, responses, strict=True):
        if isinstance(reply, HTTPError):
            failures[value] = reply
        result = _task_result("failed to fetch session info for", value, reply)
        if isinstance(result, dict):
            results.append(result)
    log.debug("Session info records collected: %s", results)
    _raise_failures("info", errors or self.errors, results, failures)
    return results

logs `async` ¶

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:

>>> from canfar.sessions import AsyncSession
>>> session = AsyncSession()
>>> await session.logs(ids="hjko98yghj")
>>> await session.logs(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

async 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:
        >>> from canfar.sessions import AsyncSession
        >>> session = AsyncSession()
        >>> await session.logs(ids="hjko98yghj")
        >>> await session.logs(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    parameters: dict[str, str] = {"view": "logs"}
    results: dict[str, str] = {}
    failures: dict[str | int, HTTPError] = {}

    async def request(value: str) -> tuple[str, str]:
        response = await self.asynclient.get(
            url=_session_url(value),
            params=parameters,
        )
        return value, response.text

    tasks = [request(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for value, reply in zip(ids, responses, strict=True):
        if isinstance(reply, HTTPError):
            failures[value] = reply
        result = _task_result("failed to fetch logs for session", value, reply)
        if isinstance(result, tuple):
            results[result[0]] = result[1]

    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 `async` ¶

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 AsyncSession
>>> session = AsyncSession()
>>> await session.renew(ids="hjko98yghj")
>>> await session.renew(ids=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

async 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 AsyncSession
        >>> session = AsyncSession()
        >>> await session.renew(ids="hjko98yghj")
        >>> await session.renew(ids=["hjko98yghj", "ikvp1jtp"])
    """
    ids = _ids(ids)
    results: dict[str, bool] = {}
    failures: dict[str | int, HTTPError] = {}

    async def request(value: str) -> tuple[str, HTTPError | None]:
        try:
            await self.asynclient.post(
                url=_session_url(value), params={"action": "renew"}
            )
        except HTTPError as err:
            _renew(value, err)
            return value, err
        return value, None

    tasks = [request(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        _raise_authentication_failure(reply)
        if isinstance(reply, tuple):
            value, failure = reply
            results[value] = failure is None
            if failure is not None:
                failures[value] = failure
    log.debug(results)
    _raise_failures("renew", errors or self.errors, results, failures)
    return results

stats `async` ¶

stats()

Get statistics for the canfar cluster.

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 AsyncSession
>>> session = AsyncSession()
>>> await session.stats()
{'cores': {'requestedCPUCores': 377,
 'coresAvailable': 960,
 'maxCores': {'cores': 32, 'withRam': '147Gi'}},
 'ram': {'maxRAM': {'ram': '226Gi', 'withCores': 32}}}

Source code in canfar/sessions.py

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

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

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

Asynchronous Sessions¶

Return shapes and parity¶

Async workflow¶

Select Sessions for cleanup¶

Renew an interactive Session¶

Raise on failures¶

connect async ¶

create async ¶

destroy async ¶

destroy_with async ¶

events async ¶

fetch async ¶

info async ¶

logs async ¶

renew async ¶

stats async ¶

connect `async` ¶

create `async` ¶

destroy `async` ¶

destroy_with `async` ¶

events `async` ¶

fetch `async` ¶

info `async` ¶

logs `async` ¶

renew `async` ¶

stats `async` ¶