athena.request

Client

Client class for making http requests

Attributes:

Name	Type	Description
name	str	Name identifier for client instance

Source code in athena/request.py

class Client:
    """Client class for making http requests

    Attributes:
        name (str): Name identifier for client instance
    """
    def __init__(
            self,
            session: requests.Session,
            async_session: aiohttp.ClientSession,
            partial_request_builder: Callable[[RequestBuilder], RequestBuilder] | None=None,
            name=None,
            pre_hook: Callable[[str], None] | None=None,
            post_hook: Callable[[AthenaTrace], None] | None=None
        ):

        if partial_request_builder is not None:
            self.__base_request_apply = partial_request_builder(RequestBuilder()).compile()
        else:
            self.__base_request_apply = lambda rq: rq
        self.__session = session
        self.__async_session = async_session
        self.name = name or ""
        self.__pre_hook = pre_hook or (lambda _: None)
        self.__post_hook = post_hook or (lambda _: None)
        self.__async_lock = asyncio.Lock()

    def _generate_trace_id(self):
        return str(uuid.uuid4())

    def send(self, method, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends a synchronous HTTP request.

        Args:
            method (str): HTTP method ('GET'/'POST'/'PUT'/'DELETE').
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        trace_id = self._generate_trace_id()
        athena_request = self.__base_request_apply(AthenaRequest())
        athena_request.url = url
        athena_request.method = method
        if build_request is not None:
            athena_request = build_request(RequestBuilder()).apply(athena_request)

        athena_request._run_before_hooks()
        self.__pre_hook(trace_id)
        request = athena_request._to_requests_request(self.__session)

        start = time()
        response = self.__session.send(request, allow_redirects=athena_request.allow_redirects, timeout=athena_request.timeout, verify=athena_request.verify_ssl)
        end = time()

        trace_name = ""
        if self.name is not None and len(self.name) > 0:
            trace_name += self.name + "+"
        trace_name += athena_request.url
        trace = AthenaTrace(trace_id, trace_name, response.request, response, start, end)

        if(athena_request.verify_ssl == False):
            trace.warnings.append("request was executed with ssl verification disabled")

        self.__post_hook(trace)
        athena_request._run_after_hooks(trace.response)

        return trace.response

    async def send_async(self, method, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends an asynchronous HTTP request.

        Args:
            method (str): HTTP method ('GET', 'POST', etc).
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        trace_id = self._generate_trace_id()
        athena_request = self.__base_request_apply(AthenaRequest())
        athena_request.url = url
        athena_request.method = method
        if build_request is not None:
            athena_request = build_request(RequestBuilder()).apply(athena_request)

        athena_request._run_before_hooks()
        async with self.__async_lock:
            self.__pre_hook(trace_id)
        request = athena_request._to_aiohttp_request()

        trace = None
        timeout = aiohttp.ClientTimeout(total=athena_request.timeout)

        start = time()
        async with self.__async_session.request(request.method, request.url, timeout=timeout, **request.kwargs) as response:
            end = time()
            trace_name = ""
            if self.name is not None and len(self.name) > 0:
                trace_name += self.name + "+"
            trace_name += athena_request.url
            assert isinstance(response, LinkedResponse)
            request = response.athena_get_request()
            assert request is not None
            if isinstance(request.body, aiohttp.BytesPayload):
                writer = BasicStringWriter()
                await request.body.write(writer)
                request_text = writer.decode()
                trace = AthenaTrace(trace_id, trace_name, request, response, start, end, request_text=request_text, response_text=await response.text())
            else:
                trace = AthenaTrace(trace_id, trace_name, request, response, start, end, response_text=await response.text())

        if(athena_request.verify_ssl == False):
            trace.warnings.append("request was executed with ssl verification disabled")

        async with self.__async_lock:
            self.__post_hook(trace)
        athena_request._run_after_hooks(trace.response)
        return trace.response

    def get(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends a synchronous GET request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return self.send("get", url, build_request)
    def post(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends a synchronous POST request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return self.send("post", url, build_request)
    def delete(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends a synchronous DELETE request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return self.send("delete", url, build_request)
    def put(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends a synchronous PUT request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return self.send("put", url, build_request)
    async def get_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends an asynchronous GET request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return await self.send_async("get", url, build_request)
    async def post_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends an asynchronous POST request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return await self.send_async("post", url, build_request)
    async def delete_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends an asynchronous DELETE request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return await self.send_async("delete", url, build_request)
    async def put_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
        """
        Sends an asynchronous PUT request.

        Args:
            url (str): URL endpoint for the request
            build_request (Callable[[RequestBuilder], RequestBuilder], optional):
                Optional function to build or modify the request.

        Returns:
            ResponseTrace: Response trace object containing request and response details.
        """
        return await self.send_async("put", url, build_request)

delete(url, build_request=None)

Sends a synchronous DELETE request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

def delete(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends a synchronous DELETE request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return self.send("delete", url, build_request)

delete_async(url, build_request=None)

Sends an asynchronous DELETE request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

async def delete_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends an asynchronous DELETE request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return await self.send_async("delete", url, build_request)

get(url, build_request=None)

Sends a synchronous GET request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

def get(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends a synchronous GET request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return self.send("get", url, build_request)

get_async(url, build_request=None)

Sends an asynchronous GET request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

async def get_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends an asynchronous GET request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return await self.send_async("get", url, build_request)

post(url, build_request=None)

Sends a synchronous POST request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

def post(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends a synchronous POST request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return self.send("post", url, build_request)

post_async(url, build_request=None)

Sends an asynchronous POST request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

async def post_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends an asynchronous POST request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return await self.send_async("post", url, build_request)

put(url, build_request=None)

Sends a synchronous PUT request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

def put(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends a synchronous PUT request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return self.send("put", url, build_request)

put_async(url, build_request=None)

Sends an asynchronous PUT request.

Parameters:

Name	Type	Description	Default
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

async def put_async(self, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends an asynchronous PUT request.

    Args:
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    return await self.send_async("put", url, build_request)

send(method, url, build_request=None)

Sends a synchronous HTTP request.

Parameters:

Name	Type	Description	Default
method	str	HTTP method ('GET'/'POST'/'PUT'/'DELETE').	required
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

def send(self, method, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends a synchronous HTTP request.

    Args:
        method (str): HTTP method ('GET'/'POST'/'PUT'/'DELETE').
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    trace_id = self._generate_trace_id()
    athena_request = self.__base_request_apply(AthenaRequest())
    athena_request.url = url
    athena_request.method = method
    if build_request is not None:
        athena_request = build_request(RequestBuilder()).apply(athena_request)

    athena_request._run_before_hooks()
    self.__pre_hook(trace_id)
    request = athena_request._to_requests_request(self.__session)

    start = time()
    response = self.__session.send(request, allow_redirects=athena_request.allow_redirects, timeout=athena_request.timeout, verify=athena_request.verify_ssl)
    end = time()

    trace_name = ""
    if self.name is not None and len(self.name) > 0:
        trace_name += self.name + "+"
    trace_name += athena_request.url
    trace = AthenaTrace(trace_id, trace_name, response.request, response, start, end)

    if(athena_request.verify_ssl == False):
        trace.warnings.append("request was executed with ssl verification disabled")

    self.__post_hook(trace)
    athena_request._run_after_hooks(trace.response)

    return trace.response

send_async(method, url, build_request=None)

Sends an asynchronous HTTP request.

Parameters:

Name	Type	Description	Default
method	str	HTTP method ('GET', 'POST', etc).	required
url	str	URL endpoint for the request	required
build_request	Callable[[RequestBuilder], RequestBuilder]	Optional function to build or modify the request.	None

Returns:

Name	Type	Description
ResponseTrace	ResponseTrace	Response trace object containing request and response details.

Source code in athena/request.py

async def send_async(self, method, url, build_request: Callable[[RequestBuilder], RequestBuilder] | None=None) -> ResponseTrace:
    """
    Sends an asynchronous HTTP request.

    Args:
        method (str): HTTP method ('GET', 'POST', etc).
        url (str): URL endpoint for the request
        build_request (Callable[[RequestBuilder], RequestBuilder], optional):
            Optional function to build or modify the request.

    Returns:
        ResponseTrace: Response trace object containing request and response details.
    """
    trace_id = self._generate_trace_id()
    athena_request = self.__base_request_apply(AthenaRequest())
    athena_request.url = url
    athena_request.method = method
    if build_request is not None:
        athena_request = build_request(RequestBuilder()).apply(athena_request)

    athena_request._run_before_hooks()
    async with self.__async_lock:
        self.__pre_hook(trace_id)
    request = athena_request._to_aiohttp_request()

    trace = None
    timeout = aiohttp.ClientTimeout(total=athena_request.timeout)

    start = time()
    async with self.__async_session.request(request.method, request.url, timeout=timeout, **request.kwargs) as response:
        end = time()
        trace_name = ""
        if self.name is not None and len(self.name) > 0:
            trace_name += self.name + "+"
        trace_name += athena_request.url
        assert isinstance(response, LinkedResponse)
        request = response.athena_get_request()
        assert request is not None
        if isinstance(request.body, aiohttp.BytesPayload):
            writer = BasicStringWriter()
            await request.body.write(writer)
            request_text = writer.decode()
            trace = AthenaTrace(trace_id, trace_name, request, response, start, end, request_text=request_text, response_text=await response.text())
        else:
            trace = AthenaTrace(trace_id, trace_name, request, response, start, end, response_text=await response.text())

    if(athena_request.verify_ssl == False):
        trace.warnings.append("request was executed with ssl verification disabled")

    async with self.__async_lock:
        self.__post_hook(trace)
    athena_request._run_after_hooks(trace.response)
    return trace.response

AuthStepFactory

Factory for adding authentication to the request.

Source code in athena/request.py

class AuthStepFactory:
    """Factory for adding authentication to the request."""
    def __init__(self, 
        add_build_step: Callable[[Callable[[AthenaRequest], AthenaRequest]], None],
        parent: RequestBuilder):
        self._add_build_step = add_build_step
        self._parent = parent
    def bearer(self, token: str) -> RequestBuilder:
        """Add bearer token authentication."""
        def set_bearer(rq: AthenaRequest):
            rq.headers["Authorization"] = f"Bearer {token}" 
            return rq
        self._add_build_step(set_bearer)
        return self._parent
    def basic(self, username: str, password: str) -> RequestBuilder:
        """Add basic (username and password) authentication."""
        def set_auth(rq: AthenaRequest):
            rq.auth = (username, password)
            return rq
        self._add_build_step(set_auth)
        return self._parent

basic(username, password)

Add basic (username and password) authentication.

Source code in athena/request.py

def basic(self, username: str, password: str) -> RequestBuilder:
    """Add basic (username and password) authentication."""
    def set_auth(rq: AthenaRequest):
        rq.auth = (username, password)
        return rq
    self._add_build_step(set_auth)
    return self._parent

bearer(token)

Add bearer token authentication.

Source code in athena/request.py

def bearer(self, token: str) -> RequestBuilder:
    """Add bearer token authentication."""
    def set_bearer(rq: AthenaRequest):
        rq.headers["Authorization"] = f"Bearer {token}" 
        return rq
    self._add_build_step(set_bearer)
    return self._parent

HookStepFactory

Factory for adding pre or post request hooks to the request.

Source code in athena/request.py

class HookStepFactory:
    """Factory for adding pre or post request hooks to the request."""
    def __init__(self,
        add_build_step: Callable[[Callable[[AthenaRequest], AthenaRequest]], None],
        parent: RequestBuilder):
        self._add_build_step = add_build_step
        self._parent = parent
    def before(self, hook: Callable[[AthenaRequest], None]) -> RequestBuilder:
        """Add pre-request hook."""
        def add_hook(rq: AthenaRequest):
            rq._before_hooks.append(hook)
            return rq
        self._add_build_step(add_hook)
        return self._parent
    def after(self, hook: Callable[[ResponseTrace], None]) -> RequestBuilder:
        """Add post-request hook."""
        def add_hook(rq: AthenaRequest):
            rq._after_hooks.append(hook)
            return rq
        self._add_build_step(add_hook)
        return self._parent

after(hook)

Add post-request hook.

Source code in athena/request.py

def after(self, hook: Callable[[ResponseTrace], None]) -> RequestBuilder:
    """Add post-request hook."""
    def add_hook(rq: AthenaRequest):
        rq._after_hooks.append(hook)
        return rq
    self._add_build_step(add_hook)
    return self._parent

before(hook)

Add pre-request hook.

Source code in athena/request.py

def before(self, hook: Callable[[AthenaRequest], None]) -> RequestBuilder:
    """Add pre-request hook."""
    def add_hook(rq: AthenaRequest):
        rq._before_hooks.append(hook)
        return rq
    self._add_build_step(add_hook)
    return self._parent

BodyStepFactory

Factory for adding a payload to the request.

Source code in athena/request.py

class BodyStepFactory:
    """Factory for adding a payload to the request."""
    def __init__(self,
        add_build_step: Callable[[Callable[[AthenaRequest], AthenaRequest]], None],
        parent: RequestBuilder):
        self._add_build_step = add_build_step
        self._parent = parent
    def json(self, payload) -> RequestBuilder:
        """Set the json payload."""
        def add_json(rq: AthenaRequest):
            rq.json = payload
            return rq
        self._add_build_step(add_json)
        return self._parent

    def form(self, payload: dict[str, str | int | float | bool]) -> RequestBuilder:
        """Set the form payload

        Args:
            payload (dict[str, str | int | float | bool]): form data

        """
        def add_data(rq: AthenaRequest):
            rq.data = payload
            return rq
        self._add_build_step(add_data)
        return self._parent

    def form_append(self, form_key: str, form_value: str | int | float | bool | list[str | int | float | bool]) -> RequestBuilder:
        """Set a value in the form payload, without overwriting the existing form payload.

        Args:
            form_key (str): key in form to append to
            form_value (str | int | float | bool | list[str | int | float | bool]): value to append to form
        """
        def add_data(rq: AthenaRequest):
            if rq.data is None:
                rq.data = {}
            if form_key not in rq.data:
                rq.data[form_key] = []
            elif not isinstance(rq.data[form_key], list):
                tmp = rq.data[form_key]
                rq.data[form_key] = [tmp]
            if isinstance(form_value, list):
                rq.data[form_key] += form_value
            else:
                rq.data[form_key].append(form_value)
            return rq
        self._add_build_step(add_data)
        return self._parent

form(payload)

Set the form payload

Parameters:

Name	Type	Description	Default
payload	dict[str, str | int | float | bool]	form data	required

Source code in athena/request.py

def form(self, payload: dict[str, str | int | float | bool]) -> RequestBuilder:
    """Set the form payload

    Args:
        payload (dict[str, str | int | float | bool]): form data

    """
    def add_data(rq: AthenaRequest):
        rq.data = payload
        return rq
    self._add_build_step(add_data)
    return self._parent

form_append(form_key, form_value)

Set a value in the form payload, without overwriting the existing form payload.

Parameters:

Name	Type	Description	Default
form_key	str	key in form to append to	required
form_value	str | int | float | bool | list[str | int | float | bool]	value to append to form	required

Source code in athena/request.py

def form_append(self, form_key: str, form_value: str | int | float | bool | list[str | int | float | bool]) -> RequestBuilder:
    """Set a value in the form payload, without overwriting the existing form payload.

    Args:
        form_key (str): key in form to append to
        form_value (str | int | float | bool | list[str | int | float | bool]): value to append to form
    """
    def add_data(rq: AthenaRequest):
        if rq.data is None:
            rq.data = {}
        if form_key not in rq.data:
            rq.data[form_key] = []
        elif not isinstance(rq.data[form_key], list):
            tmp = rq.data[form_key]
            rq.data[form_key] = [tmp]
        if isinstance(form_value, list):
            rq.data[form_key] += form_value
        else:
            rq.data[form_key].append(form_value)
        return rq
    self._add_build_step(add_data)
    return self._parent

json(payload)

Set the json payload.

Source code in athena/request.py

def json(self, payload) -> RequestBuilder:
    """Set the json payload."""
    def add_json(rq: AthenaRequest):
        rq.json = payload
        return rq
    self._add_build_step(add_json)
    return self._parent

RequestBuilder

Builder for configuring an AthenaRequest.

Attributes:

Name	Type	Description
auth	AuthStepFactory	Factory for configuring request authentication.
hook	HookStepFactory	Factory for configuring pre and post request hooks.
body	BodyStepFactory	Factory for configuring request payload.

Source code in athena/request.py

class RequestBuilder:
    """Builder for configuring an `AthenaRequest`.

    Attributes:
        auth (AuthStepFactory): Factory for configuring request authentication.
        hook (HookStepFactory): Factory for configuring pre and post request hooks.
        body (BodyStepFactory): Factory for configuring request payload.
    """
    def __init__(self):
        self._build_steps: list[Callable[[AthenaRequest], AthenaRequest]] = []
        self.auth: AuthStepFactory = AuthStepFactory(lambda rq: self._build_steps.append(rq), self)
        self.hook: HookStepFactory = HookStepFactory(lambda rq: self._build_steps.append(rq), self)
        self.body: BodyStepFactory = BodyStepFactory(lambda rq: self._build_steps.append(rq), self)

    def base_url(self, base_url) -> RequestBuilder:
        """Set the request base url."""
        def set_base_url(rq: AthenaRequest):
            rq.base_url = base_url
            return rq
        self._build_steps.append(set_base_url)
        return self

    def verify_ssl(self, verify_ssl: bool) -> RequestBuilder:
        """Enable or disable ssl verification. This is enabled by default."""
        def set_verify_ssl(rq: AthenaRequest):
            rq.verify_ssl = verify_ssl
            return rq
        self._build_steps.append(set_verify_ssl)
        return self

    def allow_redirects(self, allow_redirects: bool) -> RequestBuilder:
        """Enable or disable following redirects. This is enabled by default."""
        def set_allow_redirects(rq: AthenaRequest):
            rq.allow_redirects = allow_redirects
            return rq
        self._build_steps.append(set_allow_redirects)
        return self

    def timeout(self, seconds: float) -> RequestBuilder:
        """Set the timeout of the request.

        Args:
            seconds (float): seconds to wait before timing out the request (default 30)
        """
        def add_timeout(rq: AthenaRequest):
            rq.timeout = seconds
            return rq
        self._build_steps.append(add_timeout)
        return self


    def header(self, header_key, header_value) -> RequestBuilder:
        """Add a header to the request.

        Raises:
            AthenaException: If the given header key has already been set.
        """
        def add_header(rq: AthenaRequest):
            if header_key in  rq.headers:
                raise AthenaException(f"key \"{header_key}\" already present in request headers")
            rq.headers[header_key] = header_value
            return rq
        self._build_steps.append(add_header)
        return self

    def query(self, param_key: str, param_value: str | int | float | bool | list[str | int | float | bool]) -> RequestBuilder:
        """Add a value to the request query.

        Args:
            param_key (str): key in query to set
            param_value (str | int | float | bool | list[str | int | float | bool]): value to use. can be a single value or a list of values.
        """
        def add_query_param(rq: AthenaRequest):
            param_value_list = param_value if isinstance(param_value, list) else [param_value]
            param_value_list = [str(i) for i in param_value_list]
            for str_param_value in param_value_list:
                rq.params.append((param_key, str_param_value))
            return rq
        self._build_steps.append(add_query_param)
        return self

    def compile(self) -> Callable[[AthenaRequest], AthenaRequest]:
        def apply(request: AthenaRequest):
            for step in self._build_steps:
                request = step(request)
            return request
        return apply

    def apply(self, request: AthenaRequest) -> AthenaRequest:
        for step in self._build_steps:
            request = step(request)
        return request

allow_redirects(allow_redirects)

Enable or disable following redirects. This is enabled by default.

Source code in athena/request.py

def allow_redirects(self, allow_redirects: bool) -> RequestBuilder:
    """Enable or disable following redirects. This is enabled by default."""
    def set_allow_redirects(rq: AthenaRequest):
        rq.allow_redirects = allow_redirects
        return rq
    self._build_steps.append(set_allow_redirects)
    return self

base_url(base_url)

Set the request base url.

Source code in athena/request.py

def base_url(self, base_url) -> RequestBuilder:
    """Set the request base url."""
    def set_base_url(rq: AthenaRequest):
        rq.base_url = base_url
        return rq
    self._build_steps.append(set_base_url)
    return self

header(header_key, header_value)

Add a header to the request.

Raises:

Type	Description
AthenaException	If the given header key has already been set.

Source code in athena/request.py

def header(self, header_key, header_value) -> RequestBuilder:
    """Add a header to the request.

    Raises:
        AthenaException: If the given header key has already been set.
    """
    def add_header(rq: AthenaRequest):
        if header_key in  rq.headers:
            raise AthenaException(f"key \"{header_key}\" already present in request headers")
        rq.headers[header_key] = header_value
        return rq
    self._build_steps.append(add_header)
    return self

query(param_key, param_value)

Add a value to the request query.

Parameters:

Name	Type	Description	Default
param_key	str	key in query to set	required
param_value	str | int | float | bool | list[str | int | float | bool]	value to use. can be a single value or a list of values.	required

Source code in athena/request.py

def query(self, param_key: str, param_value: str | int | float | bool | list[str | int | float | bool]) -> RequestBuilder:
    """Add a value to the request query.

    Args:
        param_key (str): key in query to set
        param_value (str | int | float | bool | list[str | int | float | bool]): value to use. can be a single value or a list of values.
    """
    def add_query_param(rq: AthenaRequest):
        param_value_list = param_value if isinstance(param_value, list) else [param_value]
        param_value_list = [str(i) for i in param_value_list]
        for str_param_value in param_value_list:
            rq.params.append((param_key, str_param_value))
        return rq
    self._build_steps.append(add_query_param)
    return self

timeout(seconds)

Set the timeout of the request.

Parameters:

Name	Type	Description	Default
seconds	float	seconds to wait before timing out the request (default 30)	required

Source code in athena/request.py

def timeout(self, seconds: float) -> RequestBuilder:
    """Set the timeout of the request.

    Args:
        seconds (float): seconds to wait before timing out the request (default 30)
    """
    def add_timeout(rq: AthenaRequest):
        rq.timeout = seconds
        return rq
    self._build_steps.append(add_timeout)
    return self

verify_ssl(verify_ssl)

Enable or disable ssl verification. This is enabled by default.

Source code in athena/request.py

def verify_ssl(self, verify_ssl: bool) -> RequestBuilder:
    """Enable or disable ssl verification. This is enabled by default."""
    def set_verify_ssl(rq: AthenaRequest):
        rq.verify_ssl = verify_ssl
        return rq
    self._build_steps.append(set_verify_ssl)
    return self