response.py

Response dataclass

Bases: Input[T]

IO to load data from an API using requests.

Example:

>>> from ordeq_requests import Response
>>> User = ResponseContent(
...     url="https://jsonplaceholder.typicode.com/users/1"
... )
>>> User.load()  # doctest: +SKIP
b'{"id":1,"name":"Leanne Graham", ...}'  # type bytes

Use ResponseText or ResponseJSON to parse as str or dict:

>>> from ordeq_requests import ResponseText
>>> UserText = ResponseText(
...     url="https://jsonplaceholder.typicode.com/users/1"
... )
>>> UserText.load()  # doctest: +SKIP
'{"id":1,"name":"Leanne Graham", ...}'  # type str
>>> from ordeq_requests import ResponseJSON
>>> UserJSON = ResponseJSON(
...     url="https://jsonplaceholder.typicode.com/users/1"
... )
>>> UserJSON.load()  # doctest: +SKIP
{'id': 1, 'name': 'Leanne Graham', ...}  # type dict

Example in a node:

>>> from ordeq import node
>>> from ordeq_files import JSON
>>> from pathlib import Path
>>> @node(
...     inputs=UserJSON,
...     outputs=JSON(path=Path("location.json"))
... )
... def parse_user_location(user: dict) -> dict:
...     return {
...         "id": user["id"],
...         "lat": user["geo"]["lat"],
...         "lng": user["geo"]["lng"]
...     }

By default, each Response instance will create a new requests.Session. To reuse a session across multiple datasets, pass it as attribute on init:

>>> from requests.auth import HTTPBasicAuth
>>> session = Session()
>>> RequestWithCookies = ResponseJSON(
...     url="https://httpbin.org/cookies/set/cookie/123",
...     method="GET",
...     session=session
... )
>>> RequestWithCookies.load()  # doctest: +SKIP
{'cookies': {'cookie': '123'}}
>>> session.cookies.items()  # doctest: +SKIP
[('cookie', '123')]
>>> NextRequestWithCookies = ResponseJSON(
...     url="https://jsonplaceholder.typicode.com/users/2",
...     method="GET",
...     session=session  # reuse session and cookies in next request
... )

Authentication can also be set on the session attribute:

>>> from requests.auth import HTTPBasicAuth
>>> session = Session()
>>> session.auth = HTTPBasicAuth('user', 'password')
>>> RequestWithAuth = ResponseText(
...     url="https://httpbin.org/headers",
...     method="GET",
...     session=session
... )
>>> RequestWithAuth.load()  # doctest: +SKIP
{"headers":{"Accept":"*/*", ..., "Authorization": "Basic ******"}}

Similar patterns apply to other configuration like certificates and proxies. See 1 for more info.