odatadol.transport

The one request odatadol makes: an authenticated GET, through a swappable fetch.

Everything that touches the network goes through odata_get(), and odata_get() has no verb argument: odatadol cannot write, by construction.

The transport seam is fetch=(url, headers) -> (status, body bytes). The default, urllib_fetch(), is the standard library; a test scripts a fake server instead, and a host behind a proxy or with its own session pool passes its own.

>>> calls = []
>>> def fake(url, headers):
...     calls.append((url, dict(headers)))
...     return 200, b'{"value": []}'
>>> odata_get('https://example.com/odata/Products', fetch=fake)
b'{"value": []}'
>>> calls[0][1]
{'Accept': 'application/json'}

Credentials are anything with a header() -> str method (BasicAuth, BearerToken), or a zero-argument callable returning one, so a host can defer a keychain or file lookup until a read starts (it runs once per read). Neither the password nor the Authorization header ever appears in an error message, and the header never follows a redirect or a nextLink to another origin.

Module Attributes

DEFAULT_TIMEOUT	Seconds before the default transport gives up on one request.
Fetch	the transport seam.
Credentials	None (anonymous), an object with header(), or a zero-argument callable returning either.

Functions

urllib_fetch(url, headers, *[, timeout])	The default transport: one GET with urllib.
odata_get(url, *[, credentials, accept, fetch])	GET one OData URL and return the body.
redact(url)	url without any user:password@ it carries.
resolve_credentials(credentials)	credentials as an object with header() (or None), calling a factory.
same_origin(a, b)	Whether two URLs share scheme, host and port: where credentials may follow.

Classes

BasicAuth(account, password)	An account and its password, for HTTP Basic.
BearerToken(token)	An OAuth2 or similar bearer token.

Exceptions

ODataError(message, *[, status, code])

The service answered with an error.

class odatadol.transport.BasicAuth(account, password)

Bases: object

An account and its password, for HTTP Basic. The password is never shown.

>>> BasicAuth('reader', 'secret')
BasicAuth(account='reader')
>>> BasicAuth('reader', 'secret').header()
'Basic cmVhZGVyOnNlY3JldA=='

class odatadol.transport.BearerToken(token)

Bases: object

An OAuth2 or similar bearer token. Never shown.

odatadol.transport.Credentials

None (anonymous), an object with header(), or a zero-argument callable returning either.

Type:

What credentials= accepts

alias of None | _HasHeader | Callable[[], Any]

odatadol.transport.DEFAULT_TIMEOUT = 120.0

Seconds before the default transport gives up on one request.

odatadol.transport.Fetch

the transport seam. Only GET is ever issued through it.

Type:

(url, headers) -> (status, body bytes)

alias of Callable[[str, Mapping[str, str]], tuple[int, bytes]]

exception odatadol.transport.ODataError(message, *, status=None, code='')

Bases: RuntimeError

The service answered with an error. Carries the HTTP status and the OData code.

odatadol.transport.odata_get(url, *, credentials=None, accept='application/json', fetch=None)

GET one OData URL and return the body. The only request odatadol makes.

Raises:

ODataError – On any non-2xx status, with the service’s error code and message. The message names the URL (without userinfo) and never the Authorization header.

Return type:

bytes

odatadol.transport.redact(url)

url without any user:password@ it carries.

>>> redact('https://user:pw@example.com:8443/odata/Products?$top=1')
'https://example.com:8443/odata/Products?$top=1'
>>> redact('https://u:p@[::1]:8443/x')
'https://[::1]:8443/x'

Return type:

str

odatadol.transport.resolve_credentials(credentials)

credentials as an object with header() (or None), calling a factory.

Functions that make several requests resolve once, so a factory (a keychain lookup) runs once per read, not once per page.

Return type:

_HasHeader | None

odatadol.transport.same_origin(a, b)

Whether two URLs share scheme, host and port: where credentials may follow.

>>> same_origin('https://example.com/odata/A', 'https://example.com/odata/B?x=1')
True
>>> same_origin('https://example.com/odata', 'https://elsewhere.example/odata')
False

Return type:

bool

odatadol.transport.urllib_fetch(url, headers, *, timeout=120.0)

The default transport: one GET with urllib. An HTTP error is returned, not raised, so odata_get() reads every status the same way.

A redirect to another origin is followed without the Authorization header. Use functools.partial() to change timeout.

Raises:

ODataError – If no HTTP answer came back (DNS, refused connection, timeout, a malformed URL), naming the URL without any userinfo it carries.

Return type:

tuple[int, bytes]