Http utils

Little module dealing with http call in an asynchronous way, wrapped in Option monads.

Basic api

get_opt and all associated method accept kwargs for dealing with timeout, ssl etc...

from fateful.http import try_get
from fateful.monad.opt import Some, Err, _, default
from fateful.monad.func import raise_error, identity
from aiohttp import ClientSession


async def http_call():
    async with ClientSession() as session:
        return (
            await try_get("http://google.com", session=session)
            .match(
                Some(_) >> identity,
                Err(_) >> raise_error,
                default() >> None
            )
        )

Without the match api:

from fateful.http import try_get
from aiohttp import ClientSession


async def http_call():
    async with  ClientSession() as session:
        return (
            await try_get("http://google.com", session=session)
            .or_none()
        )

Automatic json conversion

When server sends back a header Content-Type which contains application/json, response is automatically converted into a json object.

from fateful.http import try_get
from aiohttp import ClientSession
from fn import _


async def http_call():
    async with  ClientSession() as session:
        title = (
            await try_get(
                "https://jsonplaceholder.typicode.com/todos/1",
                session=session
            )
            .map(_.title)
            .or_else("Unknown")
        )
        assert title == "delectus aut autem"

Error management

Throws a client error

💻 API reference

ContentType

Content-Type header values.

Source code in fateful/http.py

class ContentType:
    """Content-Type header values."""

    value = "Content-Type"
    APPLICATION_JSON = "application/json"

HttpMethods

Bases: str, Enum

A list of basic HTTP methods.

Source code in fateful/http.py

class HttpMethods(str, Enum):
    """A list of basic HTTP methods."""

    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"
    OPTIONS = "OPTIONS"

request(method, url, /, *, session, object_hook=None, **kwargs) async

Generic method for making a request

Source code in fateful/http.py

async def request(
    method: HttpMethods,
    url: str | URL,
    /,
    *,
    session: ClientSession,
    object_hook: type[T] | None = None,
    **kwargs: t.Any,
) -> Ok[str | js_array | js_object | T | list[T]] | Err[ClientError | JSONDecodeError]:
    """
    Generic method for making a request
    """
    try:
        async with session.request(method.value, url, **kwargs) as resp:
            try:
                content_type = resp.headers.getall(ContentType.value)
                is_json = ContentType.APPLICATION_JSON in content_type
                resp.raise_for_status()
                resp_as_text = await resp.text()
                if is_json:
                    return try_parse(resp_as_text, object_hook=object_hook)
                return Ok(resp_as_text)
            except ClientError as e:  # pragma: no cover
                logging.error(e)
                return Err(e)
    except ClientError as e:
        logging.exception(e)
        return Err(e)

try_get(url, **kwargs)

return an async try of get

Parameters:

Name	Type	Description	Default
url	str | URL	url to get	required

Returns:

Type	Description
	async try

Source code in fateful/http.py

def try_get(url: str | URL, **kwargs):
    """
    return an async try of get

    Args:
        url: url to get

    Returns:
        : async try
    """
    return async_try(get)(url, **kwargs)