API Reference¶

This page is reference for the public API. Each class or function can be imported directly from apiwrappers.

apiwrappers.fetch(driver: apiwrappers.protocols.Driver, request: apiwrappers.entities.Request, timeout: Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue] = NoValue(), model: None = None, source: Optional[str] = None) → apiwrappers.entities.Response¶

apiwrappers.fetch(driver: apiwrappers.protocols.AsyncDriver, request: apiwrappers.entities.Request, timeout: Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue] = NoValue(), model: None = None, source: Optional[str] = None) → Awaitable[apiwrappers.entities.Response]

apiwrappers.fetch(driver: apiwrappers.protocols.Driver, request: apiwrappers.entities.Request, timeout: Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue] = NoValue(), model: Union[Callable[[...], apiwrappers.shortcuts.T], Type[apiwrappers.shortcuts.T]] = None, source: Optional[str] = None) → apiwrappers.shortcuts.T

apiwrappers.fetch(driver: apiwrappers.protocols.AsyncDriver, request: apiwrappers.entities.Request, timeout: Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue] = NoValue(), model: Union[Callable[[...], apiwrappers.shortcuts.T], Type[apiwrappers.shortcuts.T]] = None, source: Optional[str] = None) → Awaitable[apiwrappers.shortcuts.T]

Makes a request and returns response from server.

This is shortcut function for making requests. Prefer this over Driver.fetch() and AsyncDriver.fetch(). It also has extended behaviour and can parse JSON if model arg provided.

Parameters

driver – driver that actually makes a request.
request – request object.
timeout – how many seconds to wait for the server to send data before giving up. If set to None waits infinitely. If provided, will take precedence over the driver.timeout.
model – parser for a json response. This can be either type, e.g. List[int], or a callable that accepts json.
source – name of the key in the json, which value will be passed to the model. You may use dotted notation to traverse keys, e.g. key1.key2.

Returns

Response if regular driver is provided and model is not.
Awaitable[Response] if asynchronous driver is provided, model is not.
T if regular driver and model is provided. The T corresponds to model type.
Awaitable[T] if asynchronous driver and model is provided. The T corresponds to model type.

Raises

Timeout – the request timed out.
ssl.SSLError – An SSL error occurred.
ConnectionFailed – a connection error occurred.
DriverError – in case of any other error in driver underlying library.

Simple Usage:

>>> from apiwrappers import Method, Request, fetch, make_driver
>>> driver = make_driver("requests")
>>> request = Request(Method.GET, "https://example.org")
>>> response = fetch(driver, request)
<Response [200]>

To use it in asynchronous code just use proper driver and don’t forget to await:

>>> from apiwrappers import Method, Request, fetch, make_driver
>>> driver = make_driver("aiohttp")
>>> request = Request(Method.GET, "https://example.org")
>>> response = await fetch(driver, request)
<Response [200]>

If you provide model argument the JSON response will be parsed:

>>> from dataclasses import dataclass
>>> from typing import List
>>> from apiwrappers import Method, Request, fetch, make_driver
>>> @dataclass
... class Repo:
...     name: str
>>> driver = make_driver("requests")
>>> Request(
...     Method.GET,
...     "https://api.github.com/users/unmade/repos",
... )
>>> fetch(driver, request, model=List[Repo])
[Repo(name='am-date-picker'), ...]

Do note, it’s highly discourage to use Optional if a fetch call, because mypy can’t infer proper type for that case and the return type will be object

apiwrappers.make_driver(driver_type: typing_extensions.Literal[requests], *middleware: Type[apiwrappers.protocols.Middleware], timeout: Union[int, float, None, datetime.timedelta] = 'DEFAULT_TIMEOUT', verify: Union[bool, str] = 'True', cert: Union[str, None, Tuple[str, str]] = 'None') → apiwrappers.protocols.Driver¶

apiwrappers.make_driver(driver_type: typing_extensions.Literal[aiohttp], *middleware: Type[apiwrappers.protocols.AsyncMiddleware], timeout: Union[int, float, None, datetime.timedelta] = 'DEFAULT_TIMEOUT', verify: Union[bool, str] = 'True', cert: Union[str, None, Tuple[str, str]] = 'None') → apiwrappers.protocols.AsyncDriver

apiwrappers.make_driver(driver_type: str, *middleware: Union[Type[apiwrappers.protocols.Middleware], Type[apiwrappers.protocols.AsyncMiddleware]], timeout: Union[int, float, None, datetime.timedelta] = 'DEFAULT_TIMEOUT', verify: Union[bool, str] = 'True', cert: Union[str, None, Tuple[str, str]] = 'None') → Union[apiwrappers.protocols.Driver, apiwrappers.protocols.AsyncDriver]

Creates driver instance and returns it

This is a factory function to ease driver instantiation. That way you can abstract from specific driver class - no need to import it, no need to know how the class is called.

Parameters

driver_type – specifies what kind of driver to create. Valid choices are request and aiohttp.
*middleware – middleware to apply to driver. Dependant on driver_type it should be of one kind - either Type[Middleware] for regular drivers and Type[AsyncMiddleware] for asynchronous ones.
timeout – how many seconds to wait for the server to send data before giving up. If set to None waits infinitely.
verify – Either a boolean, in which case it controls whether to verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use.
cert – Either a path to SSL client cert file (.pem) or a (‘cert’, ‘key’) tuple.

Returns

Driver if driver_type is requests.
AsyncDriver if driver_type is aiohttp.

Raises

ValueError – if unknown driver type specified

Usage:

>>> from apiwrappers import make_driver
>>> make_driver("requests")
RequestsDriver(timeout=300, verify=True)

Driver Protocols¶

class apiwrappers.AsyncDriver(*args, **kwds)¶

Protocol describing asynchronous driver.

middleware¶

list of middleware to be run on every request.

Type: MiddlewareChain

timeout¶

how many seconds to wait for the server to send data before giving up. If set to None should wait infinitely.

Type: Timeout

verify¶

Either a boolean, in which case it controls whether to verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use.

Type: Verify

cert¶

Either a path to SSL client cert file (.pem) or a (‘cert’, ‘key’) tuple.

Type: ClientCert

async fetch(request, timeout=NoValue())¶

Makes actual request and returns response from the server.

Parameters

request (apiwrappers.entities.Request) – a request object with data to send to server.
timeout (Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue]) – how many seconds to wait for the server to send data before giving up. If set to None waits infinitely. If provided, will take precedence over the AsyncDriver.timeout.

Return type

apiwrappers.entities.Response

Returns: response from the server.

Raises

Timeout – the request timed out.
ssl.SSLError – An SSL error occurred.
ConnectionFailed – a connection error occurred.
DriverError – in case of any other error in driver underlying library.

Parameters

request (apiwrappers.entities.Request) –
timeout (Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue]) –

Return type

apiwrappers.entities.Response

class apiwrappers.Driver(*args, **kwds)¶

Protocol describing regular synchronous driver.

middleware¶

list of middleware to be run on every request.

Type: MiddlewareChain

timeout¶

how many seconds to wait for the server to send data before giving up. If set to None should wait infinitely.

Type: Timeout

verify¶

Either a boolean, in which case it controls whether to verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use.

Type: Verify

cert¶

Either a path to SSL client cert file (.pem) or a (‘cert’, ‘key’) tuple.

Type: ClientCert

fetch(request, timeout=NoValue())¶

Makes actual request and returns response from the server.

Parameters

request (apiwrappers.entities.Request) – a request object with data to send to server.
timeout (Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue]) – how many seconds to wait for the server to send data before giving up. If set to None waits infinitely. If provided, will take precedence over the Driver.timeout.

Return type

apiwrappers.entities.Response

Returns: response from the server.

Raises

Timeout – The request timed out.
ssl.SSLError – An SSL error occurred.
ConnectionFailed – A Connection error occurred.
DriverError – In case of any other error in driver underlying library.

Parameters

request (apiwrappers.entities.Request) –
timeout (Union[int, float, None, datetime.timedelta, apiwrappers.structures.NoValue]) –

Return type

apiwrappers.entities.Response

Request and Response¶

class apiwrappers.Method(value)¶

A subclass of enum.Enum that defines a set of HTTP methods

The available methods are:

DELETE
HEAD
GET
POST
PUT
PATCH

Usage:

>>> from apiwrappers import Method
>>> Method.GET
<Method.GET: 'GET'>
>>> Method.POST == 'POST'
True

class apiwrappers.Request(method, url, query_params=None, headers=None, cookies=None, auth=None, data=None, files=None, json=None)¶

A container holding a request information

Parameters

method (apiwrappers.entities.Method) – HTTP Method to use.
url (apiwrappers.structures.Url) – URL to send request to.
query_params (Mapping[str, Optional[Iterable[str]]]) – dictionary or list of tuples to send in the query string. Param with None values will not be added to the query string. Default value is empty dict.
headers (MutableMapping[str, str]) – headers to send.
cookies (MutableMapping[str, str]) – cookies to send.
auth (Optional[Union[Tuple[str, str], Callable[[], Dict[str, str]], Callable[[], Generator[Request, Response, Dict[str, str]]]]]) – Auth tuple to enable Basic Auth or callable returning dict with authorization headers, e.g. ‘{“Authorization”: “Bearer …”}’
data (Union[str, None, Mapping[str, Any], Iterable[Tuple[str, Any]]]) – the body to attach to the request. If a dictionary or list of tuples [(key, value)] is provided, form-encoding will take place.
files (Optional[Dict[str, Union[BinaryIO, Tuple[str, BinaryIO], Tuple[str, BinaryIO, str]]]]) – Dictionary of 'name': file-like-objects (or {'name': file-tuple}) for multipart encoding upload. file-tuple can be a 2-tuple ('filename', fileobj), 3-tuple ('filename', fileobj, 'content_type'), where 'content-type' is a string defining the content type of the given file.
json (Union[str, int, float, bool, None, Mapping[str, Any], List[Any]]) – json for the body to attach to the request (mutually exclusive with data arg).

Raises

ValueError – If both data or files or json args provided.

Usage:

>>> from apiwrappers import Request
>>> Request(Method.GET, 'https://example.org')
Request(method=<Method.GET: 'GET'>, ...)

class apiwrappers.Response(request, status_code, url, headers, cookies, content, encoding)¶

A container holding a response from server.

Parameters

request (apiwrappers.entities.Request) – request object to which this is a response.
status_code (int) – integer Code of responded HTTP Status, e.g. 404 or 200.
url (str) – final URL location of Response
headers (apiwrappers.structures.CaseInsensitiveDict[str]) – case-insensitive dict of response headers. For example, headers['content-encoding'] will return the value of a 'Content-Encoding' response header.
cookies (http.cookies.SimpleCookie) – cookies the server sent back.
content (bytes) – content of the response, in bytes.
encoding (str) – encoding or the response.

Return type

None

json()¶

Returns the json-encoded content of the response.

Raises: ValueError – if the response body does not contain valid json.
Return type: Union[str, int, float, bool, None, Mapping[str, Any], List[Any]]

text()¶

Returns content of the response, in unicode.

If server response doesn’t specified encoding, utf-8 will be used instead.

Return type: str

class apiwrappers.Url(template, **replacements)¶

Class to work with formatted string URLs and joining urls and path.

Sometimes it useful to keep original format string in place, for example, for logging or metrics. This class stores original format string and its replacements fields, substituting it when needed.

Parameters

template (str) – a URL as format string, e.g. “https://example.org/users/{id}”.
replacements (Any) – values to format template with.

Usage:

>>> from apiwrappers import Url
>>> url = Url("https://example.org")
>>> url("/users/{id}", id=1)
Url('https://example.org/users/{id}', id=1)
>>> str(url("/users/{id}", id=1))
'https://example.org/users/1'

__call__(path, **replacements)¶

Joins path with current URL and return a new instance.

Parameters

path (str) – a path as format string, e.g. “/users/{id}”.
replacements (Any) – values to path with.

Return type

apiwrappers.structures.Url

Returns: New instance with a url joined with path.

Exceptions¶

exception apiwrappers.DriverError¶: Base class for driver-specific errors.

exception apiwrappers.ConnectionFailed¶: A Connection error occurred.

exception apiwrappers.Timeout¶: The request timed out.

API Reference¶

Driver Protocols¶

Request and Response¶

Exceptions¶

apiwrappers

Navigation

Related Topics