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.framework 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.