client

AsyncClient

Create a client instance.

Usage:

>>> from sfapi_client import AsyncClient
>>> async with AsyncClient(client_id, client_secret) as client:
>>>    # Use client

Parameters:

Name	Type	Description	Default
client_id	Optional[str]	The client ID	None
secret	Optional[str]	The client secret	None

Returns:

Type	Description
AsyncClient	The client instance

Source code in sfapi_client/_async/client.py

def __init__(
    self,
    client_id: Optional[str] = None,
    secret: Optional[str] = None,
    key: Optional[Union[str, Path]] = None,
    api_base_url: Optional[str] = SFAPI_BASE_URL,
    token_url: Optional[str] = SFAPI_TOKEN_URL,
    wait_interval: int = 10,
):
    """
    Create a client instance.

    Usage:

    ```python
    >>> from sfapi_client import AsyncClient
    >>> async with AsyncClient(client_id, client_secret) as client:
    >>>    # Use client
    ```

    :param client_id: The client ID
    :param secret: The client secret

    :return: The client instance
    :rtype: AsyncClient
    """
    self._client_id = None
    self._secret = None
    if any(arg is None for arg in [client_id, secret]):
        self._read_client_secret_from_file(key)
    else:
        self._client_id = client_id
        self._secret = secret
    self._api_base_url = api_base_url
    self._token_url = token_url
    self._client_user = None
    self.__oauth2_session = None
    self._api = None
    self._resources = None
    self._wait_interval = wait_interval

api: AsyncApi property

API related information.

resources: AsyncResources property

Resource related information.

token: str async property

Bearer token string which can be helpful for debugging through swagger UI.

close() async

Release resources associated with the client instance.

Source code in sfapi_client/_async/client.py

async def close(self):
    """
    Release resources associated with the client instance.
    """
    if self.__oauth2_session is not None:
        await self.__oauth2_session.aclose()

compute(machine) async

Create a compute site to submit jobs or view jobs in the queue

Parameters:

Name	Type	Description	Default
machine	Union[Machine, str]	Name of the compute machine to use	required

Returns:

Type	Description
AsyncCompute	Compute object that can be used to start jobs, view the queue on the system, or list files and directories.

Source code in sfapi_client/_async/client.py

async def compute(self, machine: Union[Machine, str]) -> AsyncCompute:
    """Create a compute site to submit jobs or view jobs in the queue

    :param machine: Name of the compute machine to use
    :return: Compute object that can be used to start jobs,
    view the queue on the system, or list files and directories.
    """
    # Allows for creating a compute from a name string
    machine = Machine(machine)
    response = await self.get(f"status/{machine.value}")

    values = response.json()
    values["client"] = self
    compute = AsyncCompute.model_validate(values)

    return compute

group(name) async

Get a group.

Parameters:

Name	Type	Description	Default
name	str	The group name	required

Returns:

Type	Description
AsyncGroup	The group

Source code in sfapi_client/_async/client.py

async def group(self, name: str) -> AsyncGroup:
    """
    Get a group.

    :param name: The group name
    :return: The group
    :rtype: AsyncGroup
    """
    return await AsyncGroup._fetch_group(self, name)

user(username=None) async

Get a user.

Parameters:

Name	Type	Description	Default
username	Optional[str]	The username	None

Returns:

Type	Description
UserGroup	The user

Source code in sfapi_client/_async/client.py

async def user(self, username: Optional[str] = None) -> AsyncUser:
    """
    Get a user.

    :param username: The username
    :return: The user
    :rtype: UserGroup
    """
    return await AsyncUser._fetch_user(self, username)