Asynchronous Sessions¶

Overview

CANFAR now supports asynchronous sessions using the AsyncSession class while maintaining 1-to-1 compatibility with the Session class.

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, and destroying existing sessions.

Parameters:

Name	Type	Description	Default
`HTTPClient`	`HTTPClient`	Base HTTP client for making API requests.	required

Examples:

>>> from canfar.session import AsyncSession
>>> session = AsyncSession(
        server="https://something.example.com",
        version="v1",
        token="token",
        timeout=30,
        concurrency=100,
        loglevel=40,
    )

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, and destroying existing sessions.

    Args:
        HTTPClient (canfar.client.HTTPClient): Base HTTP client for making API requests.

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession(
                server="https://something.example.com",
                version="v1",
                token="token",
                timeout=30,
                concurrency=100,
                loglevel=40,
            )
    """

    async def fetch(
        self,
        kind: Kind | None = None,
        status: Status | None = None,
        view: View | 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.
            view (View | None, optional): Session view level. Defaults to None.

        Notes:
            By default, only the calling user's sessions are listed. If views is
            set to 'all', all user sessions are listed (with limited information).

        Returns:
            list: Sessions information.

        Examples:
            >>> from canfar.session 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, view)
        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.

        Examples:
            >>> from canfar.session import AsyncSession
            >>> session = AsyncSession()
            >>> await session.stats()
            {'instances': {
             'session': 88, 'desktopApp': 30, 'headless': 0, 'total': 118},
             '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) -> list[dict[str, Any]]:
        """Get information about session[s].

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

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

        Examples:
            >>> from canfar.session import AsyncSession
            >>> session = AsyncSession()
            >>> await session.info(session_id="hjko98yghj")
            >>> await session.info(id=["hjko98yghj", "ikvp1jtp"])
        """
        # Convert id to list if it is a string
        if isinstance(ids, str):
            ids = [ids]
        parameters: dict[str, str] = {"view": "event"}
        results: list[dict[str, Any]] = []
        tasks: list[Any] = []
        semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)

        async def bounded(value: str) -> dict[str, Any]:
            async with semaphore:
                response = await self.asynclient.get(
                    url=f"session/{value}",
                    params=parameters,
                )
                data: dict[str, Any] = response.json()
                return data

        tasks = [bounded(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            if isinstance(reply, Exception):
                log.error(reply)
            elif isinstance(reply, dict):
                results.append(reply)
        return results

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

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

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

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

        semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)
        tasks: list[Any] = []

        async def bounded(value: str) -> tuple[str, str]:
            async with semaphore:
                response = await self.asynclient.get(
                    url=f"session/{value}",
                    params=parameters,
                )
                return value, response.text

        tasks = [bounded(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            if isinstance(reply, Exception):
                log.error(reply)
            elif isinstance(reply, tuple):
                results[reply[0]] = reply[1]

        # Print logs to stdout if verbose is set to True
        if verbose:
            for key, value in results.items():
                log.info("Session ID: %s\n", key)
                log.info(value)
            return None
        return results

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

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

        Notes:
            The name of the session suffixed with the replica number. eg. test-1, test-2
            Each container will have the following environment variables injected:
                * REPLICA_ID - The replica number
                * REPLICA_COUNT - The total number of replicas

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

        Examples:
            >>> from canfar.session import AsyncSession
            >>> session = AsyncSession()
            >>> 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] = []
        tasks: list[Any] = []
        semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)

        async def bounded(parameters: list[tuple[str, Any]]) -> Any:
            async with semaphore:
                log.debug("HTTP Request Parameters: %s", parameters)
                response = await self.asynclient.post(url="session", params=parameters)
                return response.text.rstrip("\r\n")

        tasks = [bounded(payload) for payload in payloads]
        msg = f"Creating {replicas} {kind} session[s]."
        log.debug(msg)
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            if isinstance(reply, Exception):
                log.error(reply)
            elif isinstance(reply, str):
                results.append(reply)
        return results

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

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

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

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

        Examples:
            >>> from canfar.session import AsyncSession
            >>> session = AsyncSession()
            >>> await session.events(id="hjko98yghj")
            >>> await session.events(id=["hjko98yghj", "ikvp1jtp"])
        """
        if isinstance(ids, str):
            ids = [ids]
        results: list[dict[str, str]] = []
        parameters: dict[str, str] = {"view": "events"}
        tasks: list[Any] = []
        semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)

        async def bounded(value: str) -> dict[str, str]:
            async with semaphore:
                response = await self.asynclient.get(
                    url=f"session/{value}",
                    params=parameters,
                )
                return {value: response.text}

        tasks = [bounded(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            if isinstance(reply, Exception):
                log.error(reply)
            elif isinstance(reply, dict):
                results.append(dict(reply))

        if verbose and results:
            for result in results:
                for key, value in result.items():
                    log.info("Session ID: %s", key)
                    log.info(value)
        return results if not verbose else None

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

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

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

        Examples:
            >>> from canfar.session import AsyncSession
            >>> session = AsyncSession()
            >>> await session.destroy(id="hjko98yghj")
            >>> await session.destroy(id=["hjko98yghj", "ikvp1jtp"])
        """
        if isinstance(ids, str):
            ids = [ids]
        results: dict[str, bool] = {}
        semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)
        tasks: list[Any] = []

        async def bounded(value: str) -> tuple[str, bool]:
            async with semaphore:
                try:
                    await self.asynclient.delete(url=f"session/{value}")
                except HTTPError as err:
                    msg = f"Failed to destroy session {value}: {err}"
                    log.exception(msg)
                    return value, False
                else:
                    return value, True

        tasks = [bounded(value) for value in ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for reply in responses:
            if isinstance(reply, tuple):
                results[reply[0]] = reply[1]
        return results

    async def destroy_with(
        self,
        prefix: str,
        kind: Kind = "headless",
        status: Status = "Succeeded",
    ) -> dict[str, bool]:
        """Destroy session[s] matching search criteria.

        Args:
            prefix (str): Prefix to match in the session name.
            kind (Kind): Type of session. Defaults to "headless".
            status (Status): Status of the session. Defaults to "Succeeded".


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

        Notes:
            The prefix is case-sensitive.
            This method is useful for destroying multiple sessions at once.

        Examples:
            >>> from canfar.session import AsyncSession
            >>> session = AsyncSession()
            >>> await session.destroy_with(prefix="test")
            >>> await session.destroy_with(prefix="test", kind="desktop")
            >>> await session.destroy_with(prefix="car", kind="carta", status="Running")

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

`create(name, image, cores=2, ram=4, kind='headless', gpu=None, cmd=None, args=None, env=None, replicas=1)` `async` ¶

Launch a canfar session.

Parameters:

Name	Type	Description	Default
`name`	`str`	A unique name for the session.	required
`image`	`str`	Container image to use for the session.	required
`cores`	`int`	Number of cores. Defaults to 2.	`2`
`ram`	`int`	Amount of RAM (GB). Defaults to 4.	`4`
`kind`	`str`	Type of canfar session. Defaults to "headless".	`'headless'`
`gpu`	`Optional[int]`	Number of GPUs. Defaults to None.	`None`
`cmd`	`Optional[str]`	Command to run. Defaults to None.	`None`
`args`	`Optional[str]`	Arguments to the command. Defaults to None.	`None`
`env`	`Optional[Dict[str, Any]]`	Environment variables to inject. Defaults to None.	`None`
`replicas`	`int`	Number of sessions to launch. Defaults to 1.	`1`

Notes

The name of the session suffixed with the replica number. eg. test-1, test-2 Each container will have the following environment variables injected: * REPLICA_ID - The replica number * REPLICA_COUNT - The total number of replicas

Returns:

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

Examples:

>>> from canfar.session import AsyncSession
>>> session = AsyncSession()
>>> 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(
    self,
    name: str,
    image: str,
    cores: int = 2,
    ram: int = 4,
    kind: Kind = "headless",
    gpu: int | None = None,
    cmd: str | None = None,
    args: str | None = None,
    env: dict[str, Any] | None = None,
    replicas: int = 1,
) -> list[str]:
    """Launch a canfar session.

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

    Notes:
        The name of the session suffixed with the replica number. eg. test-1, test-2
        Each container will have the following environment variables injected:
            * REPLICA_ID - The replica number
            * REPLICA_COUNT - The total number of replicas

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

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession()
        >>> 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] = []
    tasks: list[Any] = []
    semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)

    async def bounded(parameters: list[tuple[str, Any]]) -> Any:
        async with semaphore:
            log.debug("HTTP Request Parameters: %s", parameters)
            response = await self.asynclient.post(url="session", params=parameters)
            return response.text.rstrip("\r\n")

    tasks = [bounded(payload) for payload in payloads]
    msg = f"Creating {replicas} {kind} session[s]."
    log.debug(msg)
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        if isinstance(reply, Exception):
            log.error(reply)
        elif isinstance(reply, str):
            results.append(reply)
    return results

`destroy(ids)` `async` ¶

Destroy session[s].

Parameters:

Name	Type	Description	Default
`ids`	`Union[str, List[str]]`	Session ID[s].	required

Returns:

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

Examples:

>>> from canfar.session import AsyncSession
>>> session = AsyncSession()
>>> await session.destroy(id="hjko98yghj")
>>> await session.destroy(id=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

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

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

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

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession()
        >>> await session.destroy(id="hjko98yghj")
        >>> await session.destroy(id=["hjko98yghj", "ikvp1jtp"])
    """
    if isinstance(ids, str):
        ids = [ids]
    results: dict[str, bool] = {}
    semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)
    tasks: list[Any] = []

    async def bounded(value: str) -> tuple[str, bool]:
        async with semaphore:
            try:
                await self.asynclient.delete(url=f"session/{value}")
            except HTTPError as err:
                msg = f"Failed to destroy session {value}: {err}"
                log.exception(msg)
                return value, False
            else:
                return value, True

    tasks = [bounded(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        if isinstance(reply, tuple):
            results[reply[0]] = reply[1]
    return results

`destroy_with(prefix, kind='headless', status='Succeeded')` `async` ¶

Destroy session[s] matching search criteria.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Prefix to match in the session name.	required
`kind`	`Kind`	Type of session. Defaults to "headless".	`'headless'`
`status`	`Status`	Status of the session. Defaults to "Succeeded".	`'Succeeded'`

Returns:

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

Notes

The prefix is case-sensitive. This method is useful for destroying multiple sessions at once.

Examples:

>>> from canfar.session import AsyncSession
>>> session = AsyncSession()
>>> await session.destroy_with(prefix="test")
>>> await session.destroy_with(prefix="test", kind="desktop")
>>> await session.destroy_with(prefix="car", kind="carta", status="Running")

Source code in canfar/sessions.py

async def destroy_with(
    self,
    prefix: str,
    kind: Kind = "headless",
    status: Status = "Succeeded",
) -> dict[str, bool]:
    """Destroy session[s] matching search criteria.

    Args:
        prefix (str): Prefix to match in the session name.
        kind (Kind): Type of session. Defaults to "headless".
        status (Status): Status of the session. Defaults to "Succeeded".


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

    Notes:
        The prefix is case-sensitive.
        This method is useful for destroying multiple sessions at once.

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession()
        >>> await session.destroy_with(prefix="test")
        >>> await session.destroy_with(prefix="test", kind="desktop")
        >>> await session.destroy_with(prefix="car", kind="carta", status="Running")

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

`events(ids, verbose=False)` `async` ¶

Get deployment events for a session[s].

Parameters:

Name	Type	Description	Default
`ids`	`Union[str, List[str]]`	Session ID[s].	required
`verbose`	`bool`	Print events to stdout. Defaults to False.	`False`

Returns:

Type	Description
`list[dict[str, str]] \| None`	Optional[List[Dict[str, str]]]: A list of events for the session[s].

Notes

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

Examples:

>>> from canfar.session import AsyncSession
>>> session = AsyncSession()
>>> await session.events(id="hjko98yghj")
>>> await session.events(id=["hjko98yghj", "ikvp1jtp"])

Source code in canfar/sessions.py

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

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

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

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

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession()
        >>> await session.events(id="hjko98yghj")
        >>> await session.events(id=["hjko98yghj", "ikvp1jtp"])
    """
    if isinstance(ids, str):
        ids = [ids]
    results: list[dict[str, str]] = []
    parameters: dict[str, str] = {"view": "events"}
    tasks: list[Any] = []
    semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)

    async def bounded(value: str) -> dict[str, str]:
        async with semaphore:
            response = await self.asynclient.get(
                url=f"session/{value}",
                params=parameters,
            )
            return {value: response.text}

    tasks = [bounded(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        if isinstance(reply, Exception):
            log.error(reply)
        elif isinstance(reply, dict):
            results.append(dict(reply))

    if verbose and results:
        for result in results:
            for key, value in result.items():
                log.info("Session ID: %s", key)
                log.info(value)
    return results if not verbose else None

`fetch(kind=None, status=None, view=None)` `async` ¶

List open sessions for the user.

Parameters:

Name	Type	Description	Default
`kind`	`Kind \| None`	Session kind. Defaults to None.	`None`
`status`	`Status \| None`	Session status. Defaults to None.	`None`
`view`	`View \| None`	Session view level. Defaults to None.	`None`

Notes

By default, only the calling user's sessions are listed. If views is set to 'all', all user sessions are listed (with limited information).

Returns:

Name	Type	Description
`list`	`list[dict[str, str]]`	Sessions information.

Examples:

>>> from canfar.session 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,
    view: View | 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.
        view (View | None, optional): Session view level. Defaults to None.

    Notes:
        By default, only the calling user's sessions are listed. If views is
        set to 'all', all user sessions are listed (with limited information).

    Returns:
        list: Sessions information.

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

`info(ids)` `async` ¶

Get information about session[s].

Parameters:

Name	Type	Description	Default
`ids`	`Union[List[str], str]`	Session ID[s].	required

Returns:

Type	Description
`list[dict[str, Any]]`	Dict[str, Any]: Session information.

Examples:

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

Source code in canfar/sessions.py

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

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

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

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession()
        >>> await session.info(session_id="hjko98yghj")
        >>> await session.info(id=["hjko98yghj", "ikvp1jtp"])
    """
    # Convert id to list if it is a string
    if isinstance(ids, str):
        ids = [ids]
    parameters: dict[str, str] = {"view": "event"}
    results: list[dict[str, Any]] = []
    tasks: list[Any] = []
    semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)

    async def bounded(value: str) -> dict[str, Any]:
        async with semaphore:
            response = await self.asynclient.get(
                url=f"session/{value}",
                params=parameters,
            )
            data: dict[str, Any] = response.json()
            return data

    tasks = [bounded(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        if isinstance(reply, Exception):
            log.error(reply)
        elif isinstance(reply, dict):
            results.append(reply)
    return results

`logs(ids, verbose=False)` `async` ¶

Get logs from a session[s].

Parameters:

Name	Type	Description	Default
`ids`	`Union[List[str], str]`	Session ID[s].	required
`verbose`	`bool`	Print logs to stdout. Defaults to False.	`False`

Returns:

Type	Description
`dict[str, str] \| None`	Dict[str, str]: Logs in text/plain format.

Examples:

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

Source code in canfar/sessions.py

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

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

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

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

    semaphore: asyncio.Semaphore = asyncio.Semaphore(self.concurrency)
    tasks: list[Any] = []

    async def bounded(value: str) -> tuple[str, str]:
        async with semaphore:
            response = await self.asynclient.get(
                url=f"session/{value}",
                params=parameters,
            )
            return value, response.text

    tasks = [bounded(value) for value in ids]
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    for reply in responses:
        if isinstance(reply, Exception):
            log.error(reply)
        elif isinstance(reply, tuple):
            results[reply[0]] = reply[1]

    # Print logs to stdout if verbose is set to True
    if verbose:
        for key, value in results.items():
            log.info("Session ID: %s\n", key)
            log.info(value)
        return None
    return results

`stats()` `async` ¶

Get statistics for the canfar cluster.

Returns:

Type	Description
`dict[str, Any]`	Dict[str, Any]: Cluster statistics.

Examples:

>>> from canfar.session import AsyncSession
>>> session = AsyncSession()
>>> await session.stats()
{'instances': {
 'session': 88, 'desktopApp': 30, 'headless': 0, 'total': 118},
 '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.

    Examples:
        >>> from canfar.session import AsyncSession
        >>> session = AsyncSession()
        >>> await session.stats()
        {'instances': {
         'session': 88, 'desktopApp': 30, 'headless': 0, 'total': 118},
         '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¶

create(name, image, cores=2, ram=4, kind='headless', gpu=None, cmd=None, args=None, env=None, replicas=1) async ¶

destroy(ids) async ¶

destroy_with(prefix, kind='headless', status='Succeeded') async ¶

events(ids, verbose=False) async ¶

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

info(ids) async ¶

logs(ids, verbose=False) async ¶

stats() async ¶

`create(name, image, cores=2, ram=4, kind='headless', gpu=None, cmd=None, args=None, env=None, replicas=1)` `async` ¶

`destroy(ids)` `async` ¶

`destroy_with(prefix, kind='headless', status='Succeeded')` `async` ¶

`events(ids, verbose=False)` `async` ¶

`fetch(kind=None, status=None, view=None)` `async` ¶

`info(ids)` `async` ¶

`logs(ids, verbose=False)` `async` ¶

`stats()` `async` ¶