`yggdrasil.dataclasses`¶

dataclasses ¶

Enhanced dataclass helpers with Arrow awareness.

WaitingConfig `dataclass` ¶

WaitingConfig(
    timeout: float = DEFAULT_TIMEOUT_TICKS,
    interval: float = 0.5,
    backoff: float = 1.5,
    max_interval: float = 10.0,
    retries: int = 4,
    max_attempts: Optional[int] = 4,
)

sleep ¶

sleep(
    iteration: int,
    start: float | None = None,
    max_interval: Optional[float] = None,
) -> None

iteration is 0-based (first wait => iteration=0)

Backoff sleep strategy: - interval == 0 => no sleep - backoff >= 1 => interval * backoff**iteration - max_interval == 0 => no cap, else cap sleep to max_interval - if start is provided and timeout > 0: * raise TimeoutError if already out of time * cap sleep so we don't oversleep past timeout

Awaitable ¶

Bases: ABC

progress ¶

progress() -> 'float | None'

A 0..1 completion fraction for a progress bar, or None if unknown.

A UI hook: a generic awaitable can't know its fraction, so the base returns None (drive a spinner, not a bar). Subclasses that do know — a batch's children done, a statement's rows fetched — override this. Consumed by :func:yggdrasil.cli.style.track.

watch ¶

watch(
    on_tick: "Any" = None, *, interval: float = 0.1, raise_error: bool = True
) -> "Awaitable"

Drive to completion, calling on_tick(self) each poll.

The hook a UI (spinner / progress bar) connects to without this trait importing any UI — keeping the layering clean. Starts the awaitable if it hasn't been, polls until done, then surfaces a failure (unless raise_error is False). Pairs with :func:yggdrasil.cli.style.track.

AwaitableBatch ¶

Bases: Awaitable

watch ¶

watch(
    on_tick: "Any" = None, *, interval: float = 0.1, raise_error: bool = True
) -> "Awaitable"

Drive to completion, calling on_tick(self) each poll.

The hook a UI (spinner / progress bar) connects to without this trait importing any UI — keeping the layering clean. Starts the awaitable if it hasn't been, polls until done, then surfaces a failure (unless raise_error is False). Pairs with :func:yggdrasil.cli.style.track.

progress ¶

progress() -> 'float | None'

Fraction of child awaitables finished — drives a real progress bar.

RefreshResult `dataclass` ¶

RefreshResult(
    value: Optional[T],
    created_at_ns: int | None = None,
    ttl_ns: int | None = None,
    expires_at_ns: int | None = None,
)

Bases: Generic[T]

Output of _refresh().

Provide either: - ttl_ns (duration) OR - expires_at_ns (absolute time)

If both are None → non-expiring.

make `classmethod` ¶

make(
    value: Optional[T] = None,
    *,
    created_at_ns: Optional[_IntLike] = None,
    ttl_ns: Optional[Union[_IntLike, timedelta]] = None,
    expires_at_ns: Optional[_IntLike] = None
) -> "RefreshResult[T]"

Convenience constructor with defaults + light casting.

Expiring `dataclass` ¶

Expiring(
    _value: Optional[T] = None,
    _created_at_ns: int = now_utc_ns(),
    _expires_at_ns: Optional[int] = None,
    _ttl_ns: Optional[int] = None,
)

Bases: Generic[T], ABC

Abstract thread-safe expiring cache holder.

Subclasses implement _refresh() which is called when .value is accessed and the cache is expired.

value `property` `writable` ¶

value: Optional[T]

Cached value with auto-refresh on expiry.

refresh ¶

refresh() -> None

Force refresh now (same as accessing .value and discarding it).

ExpiringDict ¶

ExpiringDict(
    default_ttl: Optional[Union[float, int, timedelta]] = 300.0,
    *,
    max_size: int | None = None,
    refresher: Optional[Callable[[K], RefreshResult[V]]] = None,
    on_evict: Optional[Callable[[K, V], None]] = None
)

Bases: Generic[K, V]

Thread-safe dictionary where every key carries an individual TTL.

Built on the same nanosecond time-utils as Expiring but fully lockless — every mutation rides the CPython GIL atomicity of dict.__setitem__ / dict.pop / list(dict.items()) with permissive race semantics, so the cache can never deadlock. No subclassing required.

Parameters¶

default_ttl : Default TTL as seconds (float), nanoseconds (int), timedelta, or None (keys never expire unless given a per-key TTL). max_size : Evict the soonest-to-expire key when capacity is reached. refresher : Optional Callable[[key], RefreshResult[V]]. When supplied, an expired get will call refresher(key) and atomically replace the entry rather than returning the default/raising. on_evict : Optional Callable[[key, value], None] invoked after every removal — TTL-driven sweeps, capacity evictions, explicit deletes, pop, clear, refresher replacements, __setitem__ overwrites of an existing key. The callback runs outside the cache's lock so it can do real work (close a file handle, unlink a temp file, decrement a refcount) without serializing other cache ops. Exceptions raised by the callback are swallowed so a bad callback can't poison the cache; if the callback's failure matters to the caller, they should log it themselves.

Serialization¶

__getstate__ / __setstate__ are implemented: only live (non-expired) entries are persisted. The on_evict and refresher callbacks are NOT persisted — they're typically closures over runtime state. Compatible with pickle, copy.deepcopy, and joblib.

set ¶

set(key: K, value: V, ttl: Any = ...) -> None

Insert or overwrite key.

ttl accepts seconds (float), nanoseconds (int), timedelta, or None (no expiry). Omitting ttl uses the instance default. Schedules a background purge every 15 minutes. Fires on_evict for any entry that was capacity-evicted to make room or whose key got overwritten.

Lock-free. dict.__setitem__ / dict.pop / dict.get are GIL-atomic; the capacity check and overwrite-notify bookkeeping run under a permissive race regime (under heavy contention the cache may briefly hold max_size + 1 entries or notify a value that another thread already replaced — both harmless).

get ¶

get(key: K, default: Any = None) -> Optional[V]

Return value for key, or default if missing / expired.

set_many ¶

set_many(mapping: Dict[K, V], ttl: Any = ...) -> None

Insert multiple key-value pairs sharing a TTL.

Lock-free; each insert lands via GIL-atomic dict.__setitem__. The batch is no longer atomic end-to-end — a concurrent reader may observe a partially applied batch — but each individual entry's visibility is atomic, which is all a cache needs.

Fires on_evict (after the writes) for any entries that got capacity-evicted or overwritten.

update ¶

update(
    other: Union[Dict[K, V], "ExpiringDict[K, V]", None] = None,
    ttl: Any = ...,
    **kwargs: V
) -> None

Update the dict from a mapping and/or keyword arguments, mirroring the stdlib dict.update signature.

Parameters¶

other : A plain dict, another ExpiringDict, or any object with an .items() method. When the source is an ExpiringDict and no explicit ttl is given, each key's remaining TTL is preserved so expiry semantics survive a copy. Already-expired source keys are silently skipped. ttl : Override TTL for every key written. Accepts seconds (float), nanoseconds (int), timedelta, or None (no expiry). When omitted, the instance default is used for plain-dict sources, or the source key's remaining TTL is used for ExpiringDict sources. kwargs : Additional key-value pairs merged after *other, using *ttl / the instance default.

get_many ¶

get_many(keys: Iterable[K]) -> Dict[K, V]

Return {key: value} for all live keys in keys.

Pure read — no eviction side-effect. Lock-free; relies on the GIL atomicity of dict.get. Expired entries are skipped but left in place for the background purge / next get to evict (the previous pop-on-expire path raced with concurrent refreshers).

delete_many ¶

delete_many(keys: Iterable[K]) -> int

Delete keys; returns count of keys actually removed.

ttl ¶

ttl(key: K) -> Optional[float]

Remaining TTL in seconds for key, or None if gone/expired.

Pure inspection — no eviction side-effect. (The previous evict-on-expired variant raced with concurrent refresh and could drop a freshly-set value; lazy eviction happens in get and the background sweep.)

ttl_ns ¶

ttl_ns(key: K) -> Optional[int]

Remaining TTL in nanoseconds for key, or None.

snapshot ¶

snapshot() -> Dict[K, Tuple[V, Optional[int]]]

Shallow copy of live entries as {key: (value, expires_at_ns)}. Useful for debugging or external persistence.

clear ¶

clear() -> None

Drop every entry. Fires on_evict for each removed entry.

Lock-free. When on_evict is wired up we snapshot first, then call dict.clear() — entries inserted between the snapshot and the clear get cleared too but won't trigger the callback (permissive). When no callback is set, the fast path is a single GIL-atomic dict.clear() call.

purge_expired ¶

purge_expired() -> int

Explicitly evict all expired keys; returns count removed.

Fires on_evict for every entry that expired.

refresh_key ¶

refresh_key(key: K, ttl: Any = ...) -> bool

Reset the TTL of an existing, live key. Returns True if key existed and was refreshed, False otherwise. Fires on_evict ONLY when the call discovers a silently expired entry and drops it; never when the key is genuinely refreshed.

Lock-free pop+reinsert. Permissive race: a concurrent writer that lands between our pop and our reinsert will be overwritten by ours (we restore the previous value with a new TTL). Acceptable for a TTL-bump operation.

get_or_set ¶

get_or_set(key: K, default: Union[V, Callable[[], V]], ttl: Any = ...) -> V

Return the live value for key; if missing/expired, store default (or the result of calling it) and return that.

Fires on_evict if a previously-stored expired entry got replaced — the old value is leaving the cache.

Note: the default callable is invoked OUTSIDE the internal lock. This rules out a class of deadlocks where default() takes another lock (or re-enters this cache from a different thread). The tradeoff is that under contention two threads may both invoke default() — the first writer wins, the loser silently discards its own computation.

apply_refresh_result ¶

apply_refresh_result(key: K, rr: RefreshResult[V]) -> None

Store the outcome of an external RefreshResult for key. Mirrors the contract from Expiring._apply_refresh_result.

Singleton ¶

Base class that caches one instance per hashable constructor key.

The cache is shared across every subclass (the default _singleton_key includes cls so different subclasses can coexist in one dict). A subclass that wants a private cache re-declares its own _INSTANCES ClassVar.

No mutex anywhere — :meth:__new__, :meth:to_singleton, and :meth:invalidate_singleton all ride :class:ExpiringDict's lockless GIL-atomic primitives (get_or_set for atomic check-and-insert, pop for atomic identity-guarded remove). Cannot deadlock by construction.

to_singleton ¶

to_singleton(ttl: Any = ...) -> '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 ¶

invalidate_singleton(remove_global: bool = True) -> None

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.

get_from_dict ¶

get_from_dict(
    obj: Mapping[str, Any], keys: Sequence[str], prefix: Optional[str]
) -> Any

Best-effort field lookup with optional prefix support.

Lookup order for each key

1) obj[key] 2) obj[prefix + key]

Returns:

Type	Description
`Any`	first non-MISSING value found
`Any`	MISSING if nothing matched

default_value ¶

default_value(f: Field[Any], with_factory: bool = True) -> Any

Return the effective default value for a dataclass field.

Returns:

Type	Description
`Any`	f.default when present
`Any`	f.default_factory() when present
`Any`	MISSING otherwise

serialize_dataclass_state ¶

serialize_dataclass_state(obj: Any) -> dict[str, Any]

Serialize constructor state for a dataclass instance.

Rules

only init=True fields are considered
private fields (name starts with "_") are skipped
None values are skipped
values equal to their effective default are skipped
output is a raw payload dict with no version envelope

restore_dataclass_state ¶

restore_dataclass_state(obj: Any, state: Any) -> None

Restore dataclass state from a raw payload dict.

Rules

None is treated as {}
unknown keys are ignored
missing init=True fields are filled from effective defaults
missing required init=True fields raise TypeError
non-init fields are reset to their effective defaults when available

Raises:

Type	Description
`TypeError`	If state is not a dict or a required field is missing.

now_utc_ns ¶

now_utc_ns() -> int

UTC epoch nanoseconds (monotonic-wall hybrid via time_ns).

datetime_to_epoch_ns ¶

datetime_to_epoch_ns(dt: datetime) -> int

datetime → epoch ns (microsecond precision ⇒ trailing *000).

timedelta_to_ns ¶

timedelta_to_ns(td: timedelta) -> int

timedelta → ns (microsecond precision ⇒ trailing *000).

describe_signature ¶

describe_signature(func: Callable[..., Any]) -> dict[str, Any]

Capture func's signature as a JSON-serializable dict.

Returns {"qualname", "module", "parameters": [...], "return"} where each parameter entry carries name, kind (the :class:inspect.Parameter.kind name), annotation (dotted path when the annotation is a class), and default (repr of the default) where present.

format_signature ¶

format_signature(sig_meta: Mapping[str, Any]) -> str

Render :func:describe_signature output as qualname(x: int = 5) -> str.

check_function_args ¶

check_function_args(
    func: Callable[..., Any],
    args: tuple = (),
    kwargs: Optional[Mapping[str, Any]] = None,
) -> tuple[tuple, dict[str, Any]]

Coerce args / kwargs to match func's annotated signature.

Walks :func:inspect.signature and routes each value whose parameter has a non-empty annotation through :func:yggdrasil.data.cast.convert. Returns coerced (args, kwargs) ready to splat back into the call — positional inputs stay positional, keyword inputs stay keyword (no pos-to-kw rewrite).

*args (VAR_POSITIONAL) and **kwargs (VAR_KEYWORD) annotations apply per element, so def f(*xs: int) coerces every element of xs to int. Excess positional or unknown keyword inputs (no matching parameter, no **kwargs catch-all) pass through untouched so the natural TypeError fires on call rather than being silently swallowed here.

Empty input short-circuits — no yggdrasil import fires.

Hot-path callers that invoke the same function repeatedly should reach for :func:build_row_invoker instead, which caches the :class:_SignaturePlan across calls.

build_row_invoker ¶

build_row_invoker(func: Callable[..., Any]) -> Callable[[Any], Any]

Return a per-row dispatcher that adapts row shape to func's signature.

Pre-computes the signature once and returns a callable invoker(row) -> result that:

Passes row directly when func has exactly one positional parameter and no **kwargs catch-all (the common def f(row): ... shape).
Spreads row as **kwargs when func has multiple named parameters or accepts **kwargs and row is a :class:Mapping. Keys missing from the declared parameter list are dropped unless a **kwargs catch-all is present; keys missing from the row stay unset (so defaults apply).
Spreads row as *args when func has only a *args catch-all (no other positional param) and row is a :class:tuple or :class:list.
Falls back to func(row) for anything else.

Coerces annotated arguments through the pre-built :class:_SignaturePlan coercer so a function annotated def f(id: int, name: str) called against a dict with string keys still gets its id coerced to int.

When the dict-spread call raises TypeError (e.g. the function rejected the spread shape), the invoker retries with func(row) once so a row that happens to be a dict but means "an opaque mapping value" still reaches the function. Other exceptions propagate.

Picklable: the returned closure references func and the plan by reference; both pickle through the standard cloudpickle path used elsewhere in :mod:yggdrasil.pickle.

build_batch_invoker ¶

build_batch_invoker(
    func: Callable[..., Any],
) -> Callable[["pa.RecordBatch"], list[Any]]

Return a per-batch dispatcher invoker(batch) -> list[Any].

Three dispatch shapes, picked from func's signature once:

Whole-batch tabular — when func has a single positional annotated parameter whose type is a recognised tabular shape (pa.RecordBatch / pa.Table / pl.DataFrame / pl.LazyFrame / pd.DataFrame), the entire incoming :class:pyarrow.RecordBatch is converted to that type via the :func:yggdrasil.data.cast.convert registry and handed to func in one call. The result is returned as a one-element list so the downstream chunker (e.g. _typed_cast) can fold it back into Arrow batches alongside results from other partitions.
Column-by-name + primitive target — when func has a single positional annotated parameter whose name matches a column in the incoming batch and whose annotation maps to a primitive Arrow type, the column is cast via :func:pyarrow.compute.cast (vectorised, one C++ kernel call) and func runs per cell — skipping the per-row dict reconstruction batch.to_pylist() would otherwise do.
Per-row fallback — for every other shape (multi-arg, **kwargs, no column name match, annotation that doesn't map to either category) the batch is materialised through batch.to_pylist() and dispatched via :func:build_row_invoker per row.

Per-row :func:yggdrasil.data.cast.convert calls collapse into one pa.compute.cast (column path) or one convert(batch, target) (whole-batch path) when either fast path applies.

checkargs ¶

checkargs(func: F) -> F

Wrap func so every call has its args coerced to the annotated types.

Built on :func:check_function_args — every invocation routes incoming args / kwargs through the coercion pass before the real call. Annotated parameters receive values converted via :func:yggdrasil.data.cast.convert; unannotated parameters pass through. :func:functools.wraps preserves __name__, __qualname__, __doc__, __annotations__, and the underlying __wrapped__ so :func:inspect.signature still reports the original signature.

Coroutine functions (async def) get an async wrapper that awaits the underlying call; sync functions get a plain wrapper. Re-wrapping is idempotent — applying @checkargs twice unwraps the inner __wrapped__ so the second decoration doesn't add a second coercion pass.

yggdrasil.dataclasses¶

dataclasses ¶

WaitingConfig dataclass ¶

sleep ¶

Awaitable ¶

progress ¶

watch ¶

AwaitableBatch ¶

watch ¶

progress ¶

RefreshResult dataclass ¶

make classmethod ¶

Expiring dataclass ¶

value property writable ¶

refresh ¶

ExpiringDict ¶

Parameters¶

Serialization¶

set ¶

get ¶

set_many ¶

update ¶

Parameters¶

get_many ¶

delete_many ¶

ttl ¶

ttl_ns ¶

snapshot ¶

clear ¶

purge_expired ¶

refresh_key ¶

get_or_set ¶

apply_refresh_result ¶

Singleton ¶

to_singleton ¶

invalidate_singleton ¶

get_from_dict ¶

default_value ¶

serialize_dataclass_state ¶

restore_dataclass_state ¶

now_utc_ns ¶

datetime_to_epoch_ns ¶

timedelta_to_ns ¶

describe_signature ¶

format_signature ¶

check_function_args ¶

build_row_invoker ¶

build_batch_invoker ¶

checkargs ¶

`yggdrasil.dataclasses`¶

WaitingConfig `dataclass` ¶

RefreshResult `dataclass` ¶

make `classmethod` ¶

Expiring `dataclass` ¶

value `property` `writable` ¶