Source code for redditwarp.dark.core.http_client_ASYNC


from __future__ import annotations
from typing import TYPE_CHECKING, Optional, MutableMapping
if TYPE_CHECKING:
    from ...http.handler_ASYNC import Handler

from ...core.http_client_ASYNC import HTTPClient
from ...core.reddit_please_send_json_ASYNC import RedditPleaseSendJSON
from ...http.misc_handlers.apply_params_and_headers_ASYNC import ApplyDefaultHeaders
from ...http.transport.auto_ASYNC import new_connector
from ...http.util.case_insensitive_dict import CaseInsensitiveDict
from ..auth.token_obtainment_client_ASYNC import new_token_obtainment_client
from .authorizer_ASYNC import Authorizer, Authorized
from .rate_limited_ASYNC import RateLimited
from ...core.ua_ASYNC import get_suitable_user_agent
from ...http.http_client_ASYNC import HTTPClient as BaseHTTPClient
from ...core.direct_by_origin_ASYNC import DirectByOrigin
from .const import TRUSTED_ORIGINS


[docs]class RedditHTTPClient(HTTPClient):
    @property
    def authorizer(self) -> Authorizer:
        return self.fetch_authorizer()

    def __init__(self, handler: Handler, *,
        headers: Optional[MutableMapping[str, str]] = None,
        authorizer: Optional[Authorizer] = None,
    ) -> None:
        super().__init__(handler, headers=headers)
        self._authorizer: Optional[Authorizer] = authorizer

[docs]    def fetch_authorizer(self) -> Authorizer:
        if self._authorizer is None:
            raise RuntimeError('value not set')
        return self._authorizer

[docs]    def fast_set_authorizer(self, value: Optional[Authorizer]) -> None:
        self._authorizer = value


[docs]def build_reddit_http_client() -> RedditHTTPClient:
    connector = new_connector()
    ua = get_suitable_user_agent(connector.__module__)
    headers = CaseInsensitiveDict({'User-Agent': ua})

    http = BaseHTTPClient(ApplyDefaultHeaders(connector, headers))
    authorizer = Authorizer(new_token_obtainment_client(http))
    hdlr = RedditPleaseSendJSON(RateLimited(Authorized(connector, authorizer)))
    hdlr1 = DirectByOrigin(connector, {x: hdlr for x in TRUSTED_ORIGINS})

    http = RedditHTTPClient(hdlr1, headers=headers, authorizer=authorizer)
    return http