Rest Adapter

Internal Functions

These functions are meant for use in other parts of the module. You probably shouldn't be calling these manually.
If there's an endpoint method missing from the main FloweryAPI class, you should open an issue (or a pull request).

This module contains the RestAdapter class, which is used to make requests to the Flowery API.

RestAdapter

Constructor for RestAdapter

Parameters:

Name	Type	Description	Default
config	FloweryAPIConfig	Configuration object for the FloweryAPI class	FloweryAPIConfig

Raises:

Type	Description
ValueError	Raised when the keyword arguments passed to the class constructor conflict.

Source code in pyflowery/rest_adapter.py

class RestAdapter:
    """Constructor for RestAdapter

    Args:
        config (FloweryAPIConfig): Configuration object for the FloweryAPI class

    Raises:
        ValueError: Raised when the keyword arguments passed to the class constructor conflict.
    """

    def __init__(self, config=FloweryAPIConfig) -> None:
        self._url = "https://api.flowery.pw/v1"
        self.config = config

    async def _do(self, http_method: str, endpoint: str, params: dict | None = None, timeout: float = 60) -> Result | None:
        """Internal method to make a request to the Flowery API. You shouldn't use this directly.

        Args:
            http_method (str): The [HTTP method](https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods) to use
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Raises:
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        full_url = self._url + endpoint
        headers = {
            "User-Agent": self.config.prepended_user_agent,
        }
        sanitized_params = {k: str(v) if isinstance(v, bool) else v for k, v in params.items()} if params else None
        retry_counter = 0

        async with aiohttp.ClientSession() as session:
            while retry_counter < self.config.retry_limit:
                self.config.logger.debug("Making %s request to %s with headers %s and params %s", http_method, full_url, headers, sanitized_params)
                async with session.request(method=http_method, url=full_url, params=sanitized_params, headers=headers, timeout=timeout) as response:  # type: ignore
                    try:
                        data = await response.json()
                    except (JSONDecodeError, aiohttp.ContentTypeError):
                        data = await response.read()

                    result = Result(
                        success=response.status < 300,
                        status_code=response.status,
                        message=response.reason or "Unknown error",
                        data=data,
                    )
                    self.config.logger.debug("Received response: %s %s", response.status, response.reason)
                    try:
                        if result.status_code == 429:
                            raise TooManyRequests(f"{result.message} - {result.data!r}")
                        if 400 <= result.status_code < 500:
                            raise ClientError(f"{result.status_code} - {result.message} - {result.data!r}")
                        if 500 <= result.status_code < 600:
                            raise InternalServerError(f"{result.status_code} - {result.message} - {result.data!r}")
                    except (TooManyRequests, InternalServerError) as e:
                        if retry_counter < self.config.retry_limit:
                            interval = self.config.interval * retry_counter
                            self.config.logger.error("%s - retrying in %s seconds", e, interval, exc_info=True)
                            retry_counter += 1
                            await asleep(interval)
                            continue
                        raise RetryLimitExceeded(message=f"Request failed more than {self.config.retry_limit} times, not retrying") from e
                    return result
        return None

    async def get(self, endpoint: str, params: dict | None = None, timeout: float = 60) -> Result | None:
        """Make a GET request to the Flowery API. You should almost never have to use this directly.

        If you need to use this method because an endpoint is missing, please open an issue on the [CoastalCommits repository](https://www.coastalcommits.com/cswimr/PyFlowery/issues).

        Args:
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Raises:
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        return await self._do(http_method="GET", endpoint=endpoint, params=params, timeout=timeout)

get(endpoint, params=None, timeout=60) async

Make a GET request to the Flowery API. You should almost never have to use this directly.

If you need to use this method because an endpoint is missing, please open an issue on the CoastalCommits repository.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to make the request to.	required
params	dict	Python dictionary of query parameters to send with the request.	None
timeout	float	Number of seconds to wait for the request to complete.	60

Raises:

Type	Description
TooManyRequests	Raised when the Flowery API returns a 429 status code
ClientError	Raised when the Flowery API returns a 4xx status code
InternalServerError	Raised when the Flowery API returns a 5xx status code
RetryLimitExceeded	Raised when the retry limit defined in the FloweryAPIConfig class (default 3) is exceeded

Returns:

Name	Type	Description
Result	Result | None	A Result object containing the status code, message, and data from the request.

Source code in pyflowery/rest_adapter.py

async def get(self, endpoint: str, params: dict | None = None, timeout: float = 60) -> Result | None:
    """Make a GET request to the Flowery API. You should almost never have to use this directly.

    If you need to use this method because an endpoint is missing, please open an issue on the [CoastalCommits repository](https://www.coastalcommits.com/cswimr/PyFlowery/issues).

    Args:
        endpoint (str): The endpoint to make the request to.
        params (dict): Python dictionary of query parameters to send with the request.
        timeout (float): Number of seconds to wait for the request to complete.

    Raises:
        TooManyRequests: Raised when the Flowery API returns a 429 status code
        ClientError: Raised when the Flowery API returns a 4xx status code
        InternalServerError: Raised when the Flowery API returns a 5xx status code
        RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

    Returns:
        Result: A Result object containing the status code, message, and data from the request.
    """
    return await self._do(http_method="GET", endpoint=endpoint, params=params, timeout=timeout)