AbstractRecordsOnlyAPI

Bases: AbstractAPI

Abstract class for CRUD operations on records-only (no file storage) endpoints.

Source code in yeastdnnexplorer/interface/AbstractRecordsOnlyAPI.py

class AbstractRecordsOnlyAPI(AbstractAPI):
    """Abstract class for CRUD operations on records-only (no file storage)
    endpoints."""

    def __init__(self, **kwargs):
        """
        Initialize the RecordsOnlyAPI object.

        :param kwargs: Additional parameters to pass to AbstractAPI.

        """
        self.logger = logging.getLogger(__name__)
        super().__init__(**kwargs)

    async def read(
        self,
        callback: Callable[
            [pd.DataFrame, dict[str, Any] | None, Any], Any
        ] = lambda metadata, data, cache, **kwargs: {
            "metadata": metadata,
            "data": data,
        },
        export_url_suffix="export",
        **kwargs,
    ) -> Any:
        """
        Retrieve data from the endpoint. The data will be returned as a dataframe. The
        callback function must take metadata, data, and cache as parameters.

        :param callback: The function to call with the data. Signature must
            include `metadata`, `data`, and `cache` as parameters.
        :param export_url_suffix: The URL suffix for the export endpoint. This will
            return a response object with a csv file.
        :param kwargs: This can be used to pass "params" to the request to use in place
            of `self.params`. If those are passed, they will be popped off and then
            the remaining kwargs will be passed to the callback function

        """
        if not callable(callback) or {"metadata", "data", "cache"} - set(
            callback.__code__.co_varnames
        ):
            raise ValueError(
                "The callback must be a callable function with `metadata`,",
                "`data`, and `cache` as parameters.",
            )

        export_url = f"{self.url.rstrip('/')}/{export_url_suffix}"
        self.logger.debug("read() export_url: %s", export_url)

        async with aiohttp.ClientSession() as session:
            try:
                # note that the url and the export suffix are joined such that
                # the url is stripped of any trailing slashes and the export suffix is
                # added without a leading slash
                async with session.get(
                    export_url,
                    headers=self.header,
                    params=kwargs.pop("params", self.params),
                ) as response:
                    response.raise_for_status()
                    content = await response.content.read()
                    with gzip.GzipFile(fileobj=BytesIO(content)) as f:
                        records_df = pd.read_csv(f)
                    return callback(records_df, None, self.cache, **kwargs)
            except aiohttp.ClientError as e:
                self.logger.error(f"Error in GET request: {e}")
                raise
            except pd.errors.ParserError as e:
                self.logger.error(f"Error reading request content: {e}")
                raise

__init__(**kwargs)

Initialize the RecordsOnlyAPI object.

Parameters:

Name	Type	Description	Default
kwargs		Additional parameters to pass to AbstractAPI.	{}

Source code in yeastdnnexplorer/interface/AbstractRecordsOnlyAPI.py

def __init__(self, **kwargs):
    """
    Initialize the RecordsOnlyAPI object.

    :param kwargs: Additional parameters to pass to AbstractAPI.

    """
    self.logger = logging.getLogger(__name__)
    super().__init__(**kwargs)

read(callback=lambda , , : {'metadata': metadata, 'data': data}, export_url_suffix='export', **kwargs) async

Retrieve data from the endpoint. The data will be returned as a dataframe. The callback function must take metadata, data, and cache as parameters.

Parameters:

Name	Type	Description	Default
callback	Callable[[DataFrame, dict[str, Any] | None, Any], Any]	The function to call with the data. Signature must include metadata, data, and cache as parameters.	lambda , , : {'metadata': metadata, 'data': data}
export_url_suffix		The URL suffix for the export endpoint. This will return a response object with a csv file.	'export'
kwargs		This can be used to pass “params” to the request to use in place of self.params. If those are passed, they will be popped off and then the remaining kwargs will be passed to the callback function	{}

Source code in yeastdnnexplorer/interface/AbstractRecordsOnlyAPI.py

async def read(
    self,
    callback: Callable[
        [pd.DataFrame, dict[str, Any] | None, Any], Any
    ] = lambda metadata, data, cache, **kwargs: {
        "metadata": metadata,
        "data": data,
    },
    export_url_suffix="export",
    **kwargs,
) -> Any:
    """
    Retrieve data from the endpoint. The data will be returned as a dataframe. The
    callback function must take metadata, data, and cache as parameters.

    :param callback: The function to call with the data. Signature must
        include `metadata`, `data`, and `cache` as parameters.
    :param export_url_suffix: The URL suffix for the export endpoint. This will
        return a response object with a csv file.
    :param kwargs: This can be used to pass "params" to the request to use in place
        of `self.params`. If those are passed, they will be popped off and then
        the remaining kwargs will be passed to the callback function

    """
    if not callable(callback) or {"metadata", "data", "cache"} - set(
        callback.__code__.co_varnames
    ):
        raise ValueError(
            "The callback must be a callable function with `metadata`,",
            "`data`, and `cache` as parameters.",
        )

    export_url = f"{self.url.rstrip('/')}/{export_url_suffix}"
    self.logger.debug("read() export_url: %s", export_url)

    async with aiohttp.ClientSession() as session:
        try:
            # note that the url and the export suffix are joined such that
            # the url is stripped of any trailing slashes and the export suffix is
            # added without a leading slash
            async with session.get(
                export_url,
                headers=self.header,
                params=kwargs.pop("params", self.params),
            ) as response:
                response.raise_for_status()
                content = await response.content.read()
                with gzip.GzipFile(fileobj=BytesIO(content)) as f:
                    records_df = pd.read_csv(f)
                return callback(records_df, None, self.cache, **kwargs)
        except aiohttp.ClientError as e:
            self.logger.error(f"Error in GET request: {e}")
            raise
        except pd.errors.ParserError as e:
            self.logger.error(f"Error reading request content: {e}")
            raise