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.

Classes:

Name	Description
RestAdapter	Underlying class for making HTTP requests.
Result	Result returned from low-level RestAdapter.

Functions:

Name	Description
sanitize_mapping	Sanitize a mapping by converting all values to strings.

RestAdapter

Underlying class for making HTTP requests.

Parameters:

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

Methods:

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

Source code in pyflowery/rest_adapter.py

class RestAdapter:
    """Underlying class for making HTTP requests.

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

    def __init__(self, config: FloweryAPIConfig) -> None:
        self.config: FloweryAPIConfig = config
        self._logger: Logger = self.config.logger.getChild("rest_adapter")

    async def _on_request_start(
        self,
        session: ClientSession,  # pyright: ignore[reportUnusedParameter]
        context: SimpleNamespace,  # pyright: ignore[reportUnusedParameter]
        params: TraceRequestStartParams,
    ) -> None:
        token = params.headers.get("Authorization")
        if token:
            params.headers["Authorization"] = token[:6] + "... (truncated for security)"
        self._logger.debug("Starting request <%s>", params)

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

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

        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.
            headers (Mapping[str, str | float | int | bool] | None): Headers to send with the request. Note that `User-Agent` and `Authorization` headers will be overridden by the library if provided.
            params (Mapping[str, str | float | int | bool] | None): Query parameters to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

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

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        full_url = self.config.base_url + endpoint

        sanitized_headers = sanitize_mapping(headers) if headers else {}
        sanitized_headers.update(
            {
                "User-Agent": self.config.prepended_user_agent,
            }
        )
        if self.config.token is not None:
            sanitized_headers.update({"Authorization": f"Bearer {self.config.token}"})

        sanitized_params = sanitize_mapping(params) if params else {}
        retry_counter = 0

        trace_config = TraceConfig()
        trace_config.on_request_start.append(self._on_request_start)

        async with ClientSession(trace_configs=[trace_config]) as session:
            while retry_counter < self.config.retry_limit:
                async with session.request(
                    method=http_method, url=full_url, params=sanitized_params, headers=sanitized_headers, timeout=ClientTimeout(timeout)
                ) as response:  # type: ignore
                    try:
                        data = await response.json()  # pyright: ignore[reportAny]
                    except (JSONDecodeError, ContentTypeError):
                        data = await response.read()

                    result = Result(
                        success=response.status < 300,
                        status_code=response.status,
                        message=response.reason or "Unknown error",
                        data=data,
                    )
                    self._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,
        headers: Mapping[str, str | float | int | bool] | None = None,
        params: Mapping[str, str | float | int | bool] | 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://c.csw.im/cswimr/PyFlowery/issues).

        Args:
            endpoint (str): The endpoint to make the request to.
            headers (Mapping[str, str | float | int | bool] | None): Headers to send with the request. Note that `User-Agent` and `Authorization` headers will be overridden by the library if provided.
            params (Mapping[str, str | float | int | bool] | None): Query parameters to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

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

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

get(endpoint, headers=None, 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
headers	Mapping[str, str | float | int | bool] | None	Headers to send with the request. Note that User-Agent and Authorization headers will be overridden by the library if provided.	None
params	Mapping[str, str | float | int | bool] | None	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:

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

Source code in pyflowery/rest_adapter.py

async def get(
    self,
    endpoint: str,
    headers: Mapping[str, str | float | int | bool] | None = None,
    params: Mapping[str, str | float | int | bool] | 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://c.csw.im/cswimr/PyFlowery/issues).

    Args:
        endpoint (str): The endpoint to make the request to.
        headers (Mapping[str, str | float | int | bool] | None): Headers to send with the request. Note that `User-Agent` and `Authorization` headers will be overridden by the library if provided.
        params (Mapping[str, str | float | int | bool] | None): Query parameters to send with the request.
        timeout (float): Number of seconds to wait for the request to complete.

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

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

Result

Bases: BaseModel

Result returned from low-level RestAdapter.

Attributes:

Name	Type	Description
data	list[dict[str, Any]] | dict[str, Any] | bytes	The data returned by the request.
message	str	Human readable result message from the request.
raw_response	ClientResponse | None	The raw response object from the request.
status_code	int	The HTTP status code returned by the request.
success	bool	Whether or not the request was successful.

Source code in pyflowery/rest_adapter.py

class Result(BaseModel):
    """Result returned from low-level RestAdapter."""

    model_config: ClassVar[ConfigDict] = ConfigDict(arbitrary_types_allowed=True)
    success: bool
    "Whether or not the request was successful."
    status_code: int
    "The [HTTP status code](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Status) returned by the request."
    message: str = ""
    "Human readable result message from the request."
    data: list[dict[str, Any]] | dict[str, Any] | bytes = {}  # pyright: ignore[reportExplicitAny]
    "The data returned by the request."
    raw_response: aiohttp.ClientResponse | None = None
    "The raw response object from the request."

    @override
    def __str__(self) -> str:
        return f"{self.status_code}: {self.message}"

data = {} class-attribute instance-attribute

The data returned by the request.

message = '' class-attribute instance-attribute

Human readable result message from the request.

raw_response = None class-attribute instance-attribute

The raw response object from the request.

status_code instance-attribute

The HTTP status code returned by the request.

success instance-attribute

Whether or not the request was successful.

sanitize_mapping(mapping)

Sanitize a mapping by converting all values to strings.

Parameters:

Name	Type	Description	Default
mapping	Mapping[str, str | float | int | bool]	The mapping to sanitize.	required

Returns:

Type	Description
dict[str, str]	The sanitized dictionary.

Source code in pyflowery/rest_adapter.py

def sanitize_mapping(mapping: Mapping[str, str | float | int | bool]) -> dict[str, str]:
    """Sanitize a mapping by converting all values to strings.

    Args:
        mapping (Mapping[str, str | float | int | bool]): The mapping to sanitize.

    Returns:
        The sanitized dictionary.
    """
    return {k: str(v) for k, v in mapping.items()}