async_http.fetch

Classes

HasUrl

Base class for protocol classes.

Functions

`raise_by_default`(→ None)
`default_callback`(→ tuple[UrlType, Any])
`fetch`(…)	Asynchronous get request. Pass handle_exceptions in order to get your
`fetch_many`(…)	Registers a task per url using the coroutine fetch_func with correct
`async_aiohttp_get_all`(…)	Performs asynchronous get requests.

Module Contents

class async_http.fetch.HasUrl[source]

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

url: str[source]

async_http.fetch.raise_by_default(url: Any, exception: Exception) → None[source]

async_http.fetch.default_callback(url: UrlType, response: Any) → tuple[UrlType, Any][source]

async async_http.fetch.fetch(url: UrlType, session: aiohttp.ClientSession, response_attr: str = 'json', callback: FetchCallback[UrlType, tuple[UrlType, Any]] = default_callback, handle_exceptions: ErrFunc[UrlType] | None = raise_by_default) → tuple[UrlType, Any][source]

async async_http.fetch.fetch(url: UrlType, session: aiohttp.ClientSession, response_attr: str, callback: FetchCallback[_UrlTypeT, _T], handle_exceptions: ErrFunc[_UrlTypeT] | None = raise_by_default) → _T

async async_http.fetch.fetch(url: UrlType, session: aiohttp.ClientSession, response_attr: str = 'json', *, callback: FetchCallback[_UrlTypeT, _T], handle_exceptions: ErrFunc[_UrlTypeT] | None = raise_by_default) → _T

Asynchronous get request. Pass handle_exceptions in order to get your desired error handling.

Parameters:

callback – callback to handle the response.
url – object that has an attribute url or a string
session – instance of aiohttp.ClientSession()
response_attr – valid, (awaitable) attribute for response object
handle_exceptions – optional callback for handling exceptions

Returns:

response_attr or handled exception return for each url

async async_http.fetch.fetch_many(urls: Sequence[UrlType], response_attr: str = 'json', fetch_func: FetchFunc[UrlType, tuple[UrlType, Any]] = fetch, callback: FetchCallback[UrlType, tuple[UrlType, Any]] = default_callback, handle_exceptions: HandleExceptionType[UrlType] = raise_by_default, timeout: aiohttp.ClientTimeout | None = None) → list[tuple[UrlType, Any]][source]
async async_http.fetch.fetch_many(urls: Sequence[UrlType], response_attr: str, fetch_func: FetchFunc[_UrlTypeT, _T], callback: FetchCallback[_UrlTypeT, _T], handle_exceptions: HandleExceptionType[_UrlTypeT] = raise_by_default, timeout: aiohttp.ClientTimeout | None = None) → list[_T]
async async_http.fetch.fetch_many(urls: Sequence[UrlType], response_attr: str = 'json', fetch_func: FetchFunc[_UrlTypeT, _T] = fetch, *, callback: FetchCallback[_UrlTypeT, _T], handle_exceptions: HandleExceptionType[_UrlTypeT] = raise_by_default, timeout: aiohttp.ClientTimeout | None = None) → list[_T]: Registers a task per url using the coroutine fetch_func with correct signature.

async_http.fetch.async_aiohttp_get_all(urls: Sequence[UrlType], response_attr: str = 'json', callback: FetchCallback[UrlType, tuple[UrlType, Any]] = default_callback, handle_exceptions: HandleExceptionType[UrlType] = raise_by_default, timeout: aiohttp.ClientTimeout | None = None) → list[tuple[UrlType, Any]][source]

async_http.fetch.async_aiohttp_get_all(urls: Sequence[UrlType], response_attr: str, callback: FetchCallback[_UrlTypeT, _T], handle_exceptions: HandleExceptionType[_UrlTypeT] = raise_by_default, timeout: aiohttp.ClientTimeout | None = None) → list[_T]

async_http.fetch.async_aiohttp_get_all(urls: Sequence[UrlType], response_attr: str = 'json', *, callback: FetchCallback[_UrlTypeT, _T], handle_exceptions: HandleExceptionType[_UrlTypeT] = raise_by_default, timeout: aiohttp.ClientTimeout | None = None) → list[_T]

Performs asynchronous get requests.

Example only checking the status without awaiting content:: result = async_aiohttp_get_all(test_urls, response_attr=’status’)

Other valid response attributes are ‘json’ and ‘text’, which are awaited.

If the callable of the exception handler function returns, it will be part of the returned results.