yggdrasil.io.session¶
session ¶
Session ¶
Session(
base_url: Optional[URL | str] = None,
verify: "bool | str | pathlib.Path" = True,
pool_maxsize: int = 10,
headers: "HTTPHeaders | Mapping[str, str] | None" = None,
waiting: WaitingConfig = DEFAULT_WAITING_CONFIG,
*,
auth: Optional[Authorization] = None,
proxy: Optional[URL | str] = None,
no_proxy: Optional[str] = None
)
Bases: Session
HTTP/HTTPS session — singleton-keyed by (base_url, verify, pool_maxsize, headers, waiting, auth).
Inherits the singleton + pickle + job_pool plumbing from
:class:~yggdrasil.io.session.Session and adds every HTTP-specific
concern: a stdlib-backed connection pool (HTTPSession owns the per-host
socket cache directly — no separate PoolManager indirection)
connection pool, the prepare → cache-lookup → wire-send → cache-writeback
pipeline (both single-request :meth:send and bulk :meth:send_many),
Spark fan-out via mapInArrow, the verb sugar
(:meth:get / :meth:post / :meth:put / :meth:patch /
:meth:delete / :meth:head / :meth:options / :meth:request),
cookie-header coercion, and the 403 auth-refresh retry loop.
No User-Agent generator, cookie jar, or browser-emulation layering is
built in — per-vendor integrations subclass this for their own auth /
pagination / rate-limit policy (see :class:yggdrasil.fxrate.FxRate for
a worked example).
to_singleton ¶
Promote this instance into the per-class _INSTANCES cache.
Hot listing paths (iterdir / _ls / glob) build
children with singleton_ttl=False so the bounded cache
doesn't fill up with thousands of short-lived entries. When a
caller decides one of those children is worth keeping around
(handing it to a long-running worker, returning it from an
API), :meth:to_singleton registers self into the cache
so the next constructor call with the same key collapses to
the same instance.
ttl defaults to the subclass's _SINGLETON_TTL
(... = no caching, None = process lifetime, or a
seconds count). When a different instance is already cached
under this key, that pre-existing one wins and is returned
unchanged — the cache is the source of truth.
invalidate_singleton ¶
Pop self from the per-class _INSTANCES cache.
Mutating ops on a Singleton-cached object (writes, deletes,
schema invalidations on a Databricks table, put_object on
an :class:S3Path) want to make sure the next caller asking
for the same key gets a fresh build rather than collapsing
onto this stale handle — that's what remove_global=True
(the default) does. The pop is :meth:identity-guarded:
only an entry that still points at self is removed, so
a concurrent re-construction that already raced past this
thread is left alone.
remove_global=False is a no-op. The keyword exists so
subclass invalidators (invalidate_singleton,
_invalidate_entity_tag_cache, …) can offer the same
switch without branching at the call site.
local_cache ¶
Return the session-scoped local cache folder, creating the directory on first access.
The folder lives under ~/.cache/http/<host>/<path>
when :attr:base_url is set (so different APIs on the same machine
don't collide), or ~/.cache/http/default otherwise.
Thread-safe: the directory is created under the session lock and
the resulting :class:Folder is cached for the lifetime of
the singleton.
clear_connections ¶
Close every cached idle connection.
Lifecycle convenience — closes the per-host sockets the session accumulated. Not called automatically; explicit cleanup is the caller's responsibility (or rely on process exit).
fetch ¶
fetch(
method: str,
url: URL | str,
*,
headers: Optional[Mapping[str, str]] = None,
body: Any = None,
timeout: Any = None,
preload_content: bool = False,
decode_content: bool = True,
redirect: bool = True,
tags: Optional[Mapping[str, str]] = None
) -> HTTPResponse
Low-level wire fetch — bypasses the cache / auth-refresh pipeline.
Build a synthetic :class:HTTPRequest from method /
url / headers / body and run it through the same
retry + redirect machinery :meth:_send_http does for the
regular :meth:send path. Useful when the caller just wants a
raw byte stream off a URL — Databricks external-link readers
feeding :func:pa.input_stream are the canonical case.
insecure ¶
Return a session with SSL verification disabled.
If self already has verify=False, returns self.
Otherwise returns a new :class:HTTPSession with the same
base_url / headers / waiting / auth / proxy
/ no_proxy but verify=False.
Emits :class:InsecureRequestWarning on first construction.
send ¶
send(
request: HTTPRequest,
config: SendConfig | Mapping[str, Any] | None = None,
*,
wait: WaitingConfigArg = ...,
raise_error: bool = ...,
remote_cache: CacheConfig | Mapping[str, Any] | None = ...,
local_cache: CacheConfig | Mapping[str, Any] | None = ...,
cache_only: bool = ...,
spark_session: Optional["SparkSession"] = ...,
start: bool = True,
**options
) -> HTTPResponse
Prepare, dispatch, and (optionally) await the response.
Single-request entry point — always returns a :class:Response.
adds no value.
When spark_session is bound (or True / ... resolved
via :meth:PyEnv.spark_session), the wire send is fanned out
to an executor through the same mapInArrow path
:meth:send_many uses — the driver does not cross the wire.
Otherwise the call runs synchronously on the local pool.
start=True (default) fires the wire call. start=False
builds the prepared request + response shell without crossing
the wire — the :class:StatementExecutor override uses the
same knob to return an idled :class:StatementResult whose
backend submission is deferred until
:meth:StatementResult.start fires. Plain HTTP sessions don't
need an idle :class:Response (the network call is
synchronous), so the base raises a clean
NotImplementedError via :meth:_build_idle_response.
Per-request :attr:HTTPRequest.send_config is used as the
base when no explicit config is passed — explicit kwargs
still override individual fields.
refresh_auth ¶
Resolve the auth handler and stamp the Authorization header.
Called automatically by _local_send on 401/403 responses.
Per-request request.auth wins over session-wide self.auth.
Returns True when a handler ran and the header was stamped,
False when no handler is bound and force=False.
Override this method in your session subclass to implement custom auth flows (query-param tokens, API-key rotation, challenge- response, etc.)::
class MySession(HTTPSession):
def refresh_auth(self, request, force=True):
request.headers["X-API-Key"] = self._rotate_key()
return True
prepare_request_before_send ¶
Session-wide request hook fired once per outbound request.
Stamps the session reference, sent_at timestamp, merged
headers, and auth. When the request's local_cache has
received_from / received_to but no tabular, fills
in the session's default local-cache folder. A bare
session.get(url) with no cache config does no disk I/O.
prepare_response_after_received ¶
Session-wide response hook fired once per completed network send.
Default returns response unchanged. Subclasses override to log,
redact, enrich, or wrap responses returned from the wire. Runs in
:meth:_send after :meth:_local_send and before cache writeback,
so the persisted response reflects any post-processing. Cache hits
bypass it. Travels with the session into Spark workers via
__getstate__ / __setstate__.
send_many ¶
send_many(
requests: Iterator[HTTPRequest],
config: SendConfig | Mapping[str, Any] | None = None,
*,
wait: WaitingConfigArg = ...,
raise_error: bool = ...,
remote_cache: CacheConfig | Mapping[str, Any] | None = ...,
local_cache: CacheConfig | Mapping[str, Any] | None = ...,
cache_only: bool = ...,
spark_session: Optional["SparkSession"] = ...,
batch_size: int | None = None,
ordered: bool = False,
max_in_flight: int | None = None,
max_batch_ttl: float | None = None,
**options
) -> Iterator[HTTPResponse]
Stream responses for a batch of requests.
Batch orchestration kwargs (batch_size, ordered,
max_in_flight, max_batch_ttl) control chunking and
concurrency; everything else is folded into a :class:SendConfig
that gets stamped on each request.
Send-config kwargs default to ... (Ellipsis). When left
unset the per-request :attr:HTTPRequest.send_config is
preserved; an explicit value overrides that field on every
request in the batch.
send_many_batches ¶
send_many_batches(
requests: Iterator[HTTPRequest],
*,
wait: WaitingConfigArg = ...,
raise_error: bool = ...,
remote_cache: CacheConfig | Mapping[str, Any] | None = ...,
local_cache: CacheConfig | Mapping[str, Any] | None = ...,
cache_only: bool = ...,
spark_session: Optional["SparkSession"] = ...,
batch_size: int | None = None,
ordered: bool = False,
max_in_flight: int | None = None,
max_batch_ttl: float | None = None
) -> Iterator[HTTPResponseBatch]
Yield one :class:HTTPResponseBatch per processed chunk.
Send-config kwargs default to ... (Ellipsis). When left
unset the per-request :attr:HTTPRequest.send_config is
preserved; an explicit value overrides that field on every
request in the batch.
HTTPSession ¶
HTTPSession(
base_url: Optional[URL | str] = None,
verify: "bool | str | pathlib.Path" = True,
pool_maxsize: int = 10,
headers: "HTTPHeaders | Mapping[str, str] | None" = None,
waiting: WaitingConfig = DEFAULT_WAITING_CONFIG,
*,
auth: Optional[Authorization] = None,
proxy: Optional[URL | str] = None,
no_proxy: Optional[str] = None
)
Bases: Session
HTTP/HTTPS session — singleton-keyed by (base_url, verify, pool_maxsize, headers, waiting, auth).
Inherits the singleton + pickle + job_pool plumbing from
:class:~yggdrasil.io.session.Session and adds every HTTP-specific
concern: a stdlib-backed connection pool (HTTPSession owns the per-host
socket cache directly — no separate PoolManager indirection)
connection pool, the prepare → cache-lookup → wire-send → cache-writeback
pipeline (both single-request :meth:send and bulk :meth:send_many),
Spark fan-out via mapInArrow, the verb sugar
(:meth:get / :meth:post / :meth:put / :meth:patch /
:meth:delete / :meth:head / :meth:options / :meth:request),
cookie-header coercion, and the 403 auth-refresh retry loop.
No User-Agent generator, cookie jar, or browser-emulation layering is
built in — per-vendor integrations subclass this for their own auth /
pagination / rate-limit policy (see :class:yggdrasil.fxrate.FxRate for
a worked example).
to_singleton ¶
Promote this instance into the per-class _INSTANCES cache.
Hot listing paths (iterdir / _ls / glob) build
children with singleton_ttl=False so the bounded cache
doesn't fill up with thousands of short-lived entries. When a
caller decides one of those children is worth keeping around
(handing it to a long-running worker, returning it from an
API), :meth:to_singleton registers self into the cache
so the next constructor call with the same key collapses to
the same instance.
ttl defaults to the subclass's _SINGLETON_TTL
(... = no caching, None = process lifetime, or a
seconds count). When a different instance is already cached
under this key, that pre-existing one wins and is returned
unchanged — the cache is the source of truth.
invalidate_singleton ¶
Pop self from the per-class _INSTANCES cache.
Mutating ops on a Singleton-cached object (writes, deletes,
schema invalidations on a Databricks table, put_object on
an :class:S3Path) want to make sure the next caller asking
for the same key gets a fresh build rather than collapsing
onto this stale handle — that's what remove_global=True
(the default) does. The pop is :meth:identity-guarded:
only an entry that still points at self is removed, so
a concurrent re-construction that already raced past this
thread is left alone.
remove_global=False is a no-op. The keyword exists so
subclass invalidators (invalidate_singleton,
_invalidate_entity_tag_cache, …) can offer the same
switch without branching at the call site.
local_cache ¶
Return the session-scoped local cache folder, creating the directory on first access.
The folder lives under ~/.cache/http/<host>/<path>
when :attr:base_url is set (so different APIs on the same machine
don't collide), or ~/.cache/http/default otherwise.
Thread-safe: the directory is created under the session lock and
the resulting :class:Folder is cached for the lifetime of
the singleton.
clear_connections ¶
Close every cached idle connection.
Lifecycle convenience — closes the per-host sockets the session accumulated. Not called automatically; explicit cleanup is the caller's responsibility (or rely on process exit).
fetch ¶
fetch(
method: str,
url: URL | str,
*,
headers: Optional[Mapping[str, str]] = None,
body: Any = None,
timeout: Any = None,
preload_content: bool = False,
decode_content: bool = True,
redirect: bool = True,
tags: Optional[Mapping[str, str]] = None
) -> HTTPResponse
Low-level wire fetch — bypasses the cache / auth-refresh pipeline.
Build a synthetic :class:HTTPRequest from method /
url / headers / body and run it through the same
retry + redirect machinery :meth:_send_http does for the
regular :meth:send path. Useful when the caller just wants a
raw byte stream off a URL — Databricks external-link readers
feeding :func:pa.input_stream are the canonical case.
insecure ¶
Return a session with SSL verification disabled.
If self already has verify=False, returns self.
Otherwise returns a new :class:HTTPSession with the same
base_url / headers / waiting / auth / proxy
/ no_proxy but verify=False.
Emits :class:InsecureRequestWarning on first construction.
send ¶
send(
request: HTTPRequest,
config: SendConfig | Mapping[str, Any] | None = None,
*,
wait: WaitingConfigArg = ...,
raise_error: bool = ...,
remote_cache: CacheConfig | Mapping[str, Any] | None = ...,
local_cache: CacheConfig | Mapping[str, Any] | None = ...,
cache_only: bool = ...,
spark_session: Optional["SparkSession"] = ...,
start: bool = True,
**options
) -> HTTPResponse
Prepare, dispatch, and (optionally) await the response.
Single-request entry point — always returns a :class:Response.
adds no value.
When spark_session is bound (or True / ... resolved
via :meth:PyEnv.spark_session), the wire send is fanned out
to an executor through the same mapInArrow path
:meth:send_many uses — the driver does not cross the wire.
Otherwise the call runs synchronously on the local pool.
start=True (default) fires the wire call. start=False
builds the prepared request + response shell without crossing
the wire — the :class:StatementExecutor override uses the
same knob to return an idled :class:StatementResult whose
backend submission is deferred until
:meth:StatementResult.start fires. Plain HTTP sessions don't
need an idle :class:Response (the network call is
synchronous), so the base raises a clean
NotImplementedError via :meth:_build_idle_response.
Per-request :attr:HTTPRequest.send_config is used as the
base when no explicit config is passed — explicit kwargs
still override individual fields.
refresh_auth ¶
Resolve the auth handler and stamp the Authorization header.
Called automatically by _local_send on 401/403 responses.
Per-request request.auth wins over session-wide self.auth.
Returns True when a handler ran and the header was stamped,
False when no handler is bound and force=False.
Override this method in your session subclass to implement custom auth flows (query-param tokens, API-key rotation, challenge- response, etc.)::
class MySession(HTTPSession):
def refresh_auth(self, request, force=True):
request.headers["X-API-Key"] = self._rotate_key()
return True
prepare_request_before_send ¶
Session-wide request hook fired once per outbound request.
Stamps the session reference, sent_at timestamp, merged
headers, and auth. When the request's local_cache has
received_from / received_to but no tabular, fills
in the session's default local-cache folder. A bare
session.get(url) with no cache config does no disk I/O.
prepare_response_after_received ¶
Session-wide response hook fired once per completed network send.
Default returns response unchanged. Subclasses override to log,
redact, enrich, or wrap responses returned from the wire. Runs in
:meth:_send after :meth:_local_send and before cache writeback,
so the persisted response reflects any post-processing. Cache hits
bypass it. Travels with the session into Spark workers via
__getstate__ / __setstate__.
send_many ¶
send_many(
requests: Iterator[HTTPRequest],
config: SendConfig | Mapping[str, Any] | None = None,
*,
wait: WaitingConfigArg = ...,
raise_error: bool = ...,
remote_cache: CacheConfig | Mapping[str, Any] | None = ...,
local_cache: CacheConfig | Mapping[str, Any] | None = ...,
cache_only: bool = ...,
spark_session: Optional["SparkSession"] = ...,
batch_size: int | None = None,
ordered: bool = False,
max_in_flight: int | None = None,
max_batch_ttl: float | None = None,
**options
) -> Iterator[HTTPResponse]
Stream responses for a batch of requests.
Batch orchestration kwargs (batch_size, ordered,
max_in_flight, max_batch_ttl) control chunking and
concurrency; everything else is folded into a :class:SendConfig
that gets stamped on each request.
Send-config kwargs default to ... (Ellipsis). When left
unset the per-request :attr:HTTPRequest.send_config is
preserved; an explicit value overrides that field on every
request in the batch.
send_many_batches ¶
send_many_batches(
requests: Iterator[HTTPRequest],
*,
wait: WaitingConfigArg = ...,
raise_error: bool = ...,
remote_cache: CacheConfig | Mapping[str, Any] | None = ...,
local_cache: CacheConfig | Mapping[str, Any] | None = ...,
cache_only: bool = ...,
spark_session: Optional["SparkSession"] = ...,
batch_size: int | None = None,
ordered: bool = False,
max_in_flight: int | None = None,
max_batch_ttl: float | None = None
) -> Iterator[HTTPResponseBatch]
Yield one :class:HTTPResponseBatch per processed chunk.
Send-config kwargs default to ... (Ellipsis). When left
unset the per-request :attr:HTTPRequest.send_config is
preserved; an explicit value overrides that field on every
request in the batch.