rs_client/osam_client.md

<< Back to index

OsamClient class implementation.

BucketCredentials

Bases: BaseModel

Bucket credentials, with the same field names as those returned by the osam service.

Source code in docs/rs-client-libraries/rs_client/osam_client.py

class BucketCredentials(BaseModel):
    """Bucket credentials, with the same field names as those returned by the osam service."""

    access_key: str
    secret_key: str
    endpoint: str
    region: str

OsamClient

Bases: RsClient

OsamClient class implementation.

Attributes: see class:RsClient

Source code in docs/rs-client-libraries/rs_client/osam_client.py

class OsamClient(RsClient):
    """
    OsamClient class implementation.

    Attributes: see :py:class:`RsClient`
    """

    def __init__(
        self,
        rs_server_href: str | None,
        rs_server_api_key: str | None,
        logger: logging.Logger | None = None,
    ):
        """
        Initializes an OsamClient instance.

        Args:
            rs_server_href (str | None): The URL of the RS-Server. Pass None for local mode.
            rs_server_api_key (str | None): API key for authentication.
            logger (logging.Logger | None, optional): Logger instance (default: None).
        """
        super().__init__(
            rs_server_href,
            rs_server_api_key,
            None,
            logger,
        )

    @property
    def href_service(self) -> str:
        """
        Return the RS-Server OSAM URL hostname.
        This URL can be overwritten using the RSPY_HOST_OSAM env variable (used e.g. for local mode).
        Otherwise it should just be the RS-Server URL.
        """
        return get_href_service(self.rs_server_href, "RSPY_HOST_OSAM")

    async def get_credentials(self, timeout: int = TIMEOUT) -> BucketCredentials:
        """Get user credentials from cloud provider."""
        response = await asyncio.to_thread(
            self.http_session.get,
            f"{self.href_service}/storage/account/credentials",
            timeout=timeout,
            **self.apikey_headers,
        )
        response.raise_for_status()
        return BucketCredentials.model_validate(response.json())

href_service property

Return the RS-Server OSAM URL hostname. This URL can be overwritten using the RSPY_HOST_OSAM env variable (used e.g. for local mode). Otherwise it should just be the RS-Server URL.

__init__(rs_server_href, rs_server_api_key, logger=None)

Initializes an OsamClient instance.

Parameters:

Name	Type	Description	Default
rs_server_href	str | None	The URL of the RS-Server. Pass None for local mode.	required
rs_server_api_key	str | None	API key for authentication.	required
logger	Logger | None	Logger instance (default: None).	None

Source code in docs/rs-client-libraries/rs_client/osam_client.py

def __init__(
    self,
    rs_server_href: str | None,
    rs_server_api_key: str | None,
    logger: logging.Logger | None = None,
):
    """
    Initializes an OsamClient instance.

    Args:
        rs_server_href (str | None): The URL of the RS-Server. Pass None for local mode.
        rs_server_api_key (str | None): API key for authentication.
        logger (logging.Logger | None, optional): Logger instance (default: None).
    """
    super().__init__(
        rs_server_href,
        rs_server_api_key,
        None,
        logger,
    )

get_credentials(timeout=TIMEOUT) async

Get user credentials from cloud provider.

Source code in docs/rs-client-libraries/rs_client/osam_client.py

async def get_credentials(self, timeout: int = TIMEOUT) -> BucketCredentials:
    """Get user credentials from cloud provider."""
    response = await asyncio.to_thread(
        self.http_session.get,
        f"{self.href_service}/storage/account/credentials",
        timeout=timeout,
        **self.apikey_headers,
    )
    response.raise_for_status()
    return BucketCredentials.model_validate(response.json())