bovine.clients.bearer

BearerAuthClient dataclass

Client for using Bearer authentication.

Parameters:

Name	Type	Description	Default
session	ClientSession	The session	required
bearer_key	str	The bearer key used in the header Authorization: Bearer ${bearer_key}	required

Source code in bovine/bovine/clients/bearer.py

@dataclass
class BearerAuthClient:
    """Client for using Bearer authentication."""

    session: aiohttp.ClientSession = field(metadata={"description": "The session"})
    bearer_key: str = field(
        metadata={
            "description": "The bearer key used in the header `Authorization: Bearer ${bearer_key}`"
        }
    )

    async def get(self, url: str, headers: dict = {}):
        """GET of resource.  By default the accept header `application/json` is set. You can override this using `headers`.

        :param url: The target url
        :param headers: additional request headers.
        """
        request_headers = {
            "accept": "application/json",
            "data": get_gmt_now(),
            "user-agent": BOVINE_CLIENT_NAME,
            **headers,
            "authorization": f"Bearer {self.bearer_key}",
        }

        return await self.session.get(url, headers=request_headers)

    async def post(
        self,
        url: str,
        body: str,
        headers: dict = {},
        content_type: str = "application/activity+json",
    ):
        """POST to resource  By default the accept header `application/json` is set. You can override this using `headers`.

        :param url: The target url
        :param body: The request body
        :param headers: additional request headers.
        :param content_type: The content_type of the body
        """
        request_headers = {
            "accept": "application/json",
            "data": get_gmt_now(),
            "user-agent": BOVINE_CLIENT_NAME,
            "content-type": content_type,
            **headers,
            "authorization": f"Bearer {self.bearer_key}",
        }

        return await self.session.post(url, data=body, headers=request_headers)

    def event_source(self, url: str, headers: dict = {}) -> EventSource:
        """Returns an EventSource for the server sent events given by url. Accept header is `text/event-stream` by default

        :param url: The target url
        :param headers: additional request headers.
        """
        request_headers = {
            "accept": "text/event-stream",
            "data": get_gmt_now(),
            "user-agent": BOVINE_CLIENT_NAME,
            **headers,
            "authorization": f"Bearer {self.bearer_key}",
        }
        return EventSource(self.session, url, headers=request_headers)

event_source

event_source(url: str, headers: dict = {}) -> EventSource

Returns an EventSource for the server sent events given by url. Accept header is text/event-stream by default

Parameters:

Name	Type	Description	Default
url	str	The target url	required
headers	dict	additional request headers.	{}

Source code in bovine/bovine/clients/bearer.py

def event_source(self, url: str, headers: dict = {}) -> EventSource:
    """Returns an EventSource for the server sent events given by url. Accept header is `text/event-stream` by default

    :param url: The target url
    :param headers: additional request headers.
    """
    request_headers = {
        "accept": "text/event-stream",
        "data": get_gmt_now(),
        "user-agent": BOVINE_CLIENT_NAME,
        **headers,
        "authorization": f"Bearer {self.bearer_key}",
    }
    return EventSource(self.session, url, headers=request_headers)

get async

get(url: str, headers: dict = {})

GET of resource. By default the accept header application/json is set. You can override this using headers.

Parameters:

Name	Type	Description	Default
url	str	The target url	required
headers	dict	additional request headers.	{}

Source code in bovine/bovine/clients/bearer.py

async def get(self, url: str, headers: dict = {}):
    """GET of resource.  By default the accept header `application/json` is set. You can override this using `headers`.

    :param url: The target url
    :param headers: additional request headers.
    """
    request_headers = {
        "accept": "application/json",
        "data": get_gmt_now(),
        "user-agent": BOVINE_CLIENT_NAME,
        **headers,
        "authorization": f"Bearer {self.bearer_key}",
    }

    return await self.session.get(url, headers=request_headers)

post async

post(
    url: str,
    body: str,
    headers: dict = {},
    content_type: str = "application/activity+json",
)

POST to resource By default the accept header application/json is set. You can override this using headers.

Parameters:

Name	Type	Description	Default
url	str	The target url	required
body	str	The request body	required
headers	dict	additional request headers.	{}
content_type	str	The content_type of the body	'application/activity+json'

Source code in bovine/bovine/clients/bearer.py

async def post(
    self,
    url: str,
    body: str,
    headers: dict = {},
    content_type: str = "application/activity+json",
):
    """POST to resource  By default the accept header `application/json` is set. You can override this using `headers`.

    :param url: The target url
    :param body: The request body
    :param headers: additional request headers.
    :param content_type: The content_type of the body
    """
    request_headers = {
        "accept": "application/json",
        "data": get_gmt_now(),
        "user-agent": BOVINE_CLIENT_NAME,
        "content-type": content_type,
        **headers,
        "authorization": f"Bearer {self.bearer_key}",
    }

    return await self.session.post(url, data=body, headers=request_headers)