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()
andAsyncDriver.fetch()
. It also has extended behaviour and can parse JSON ifmodel
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 thedriver.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 tomodel
type.Awaitable[T] if asynchronous driver and model is provided. The
T
corresponds tomodel
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 afetch
call, because mypy can’t infer proper type for that case and the return type will beobject
- 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
andaiohttp
.*middleware – middleware to apply to driver. Dependant on
driver_type
it should be of one kind - eitherType[Middleware]
for regular drivers andType[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
isrequests
.AsyncDriver if
driver_type
isaiohttp
.
- 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
- 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 theAsyncDriver.timeout
.
- Return type
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
- 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
- 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 theDriver.timeout
.
- Return type
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
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.
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
orfiles
orjson
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
- json()¶
Returns the json-encoded content of the response.
- 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'
Exceptions¶
- exception apiwrappers.DriverError¶
Base class for driver-specific errors.
- exception apiwrappers.ConnectionFailed¶
A Connection error occurred.
- exception apiwrappers.Timeout¶
The request timed out.