Utilities¶

concurrency ¶

Concurrency utility helpers.

RLock ¶

RLock()

A re-entrant lock that copies and pickles as a fresh, unheld lock.

A drop-in for threading.RLock on models that get deep-copied, as a destination does with the source it is bound to: a bare lock cannot be copied. A copy never shares the original's lock, so it starts unheld.

Create the underlying lock.

acquire ¶

acquire(blocking: bool = True, timeout: float = -1) -> bool

Acquire the lock, re-entrantly for the thread already holding it.

Parameters:

Name	Type	Description	Default
`blocking`	`bool`	Whether to wait for the lock when another thread holds it.	`True`
`timeout`	`float`	Seconds to wait at most when blocking; `-1` waits forever.	`-1`

Returns:

Type	Description
`bool`	Whether the lock was acquired.

release ¶

release() -> None

Release one level of the lock held by the calling thread.

ThreadLocal ¶

Bases: _thread._local

Per-thread attributes that copy and pickle as a fresh, empty namespace.

A drop-in for threading.local on models that get deep-copied: a bare one cannot be copied, and a copy has no claim on the original's threads.

run ¶

run(coro: Coroutine[Any, Any, _T]) -> _T

Run a coroutine to completion from synchronous code.

The sync bridge to the async-native engine. It backs the sync entrypoints — asset.run(), asset.materialize(), dag.materialize() — and can be called directly to drive any other framework coroutine (e.g. a configured runner) from a script, a REPL, or a notebook cell without touching asyncio:

import interloper as il

result = il.run(il.AsyncRunner(max_workers=8).run(dag))

Unlike asyncio.run, this works where an event loop is already running (Jupyter) and reuses one persistent background loop across calls, so loop-bound clients cached on connections stay valid from one invocation to the next. Async code should await the coroutine directly instead.

Ctrl-C cancels the coroutine before re-raising KeyboardInterrupt.

Parameters:

Name	Type	Description	Default
`coro`	`Coroutine[Any, Any, _T]`	The coroutine to execute.	required

Returns:

Type	Description
`_T`	The coroutine's result.

Raises:

Type	Description
`RuntimeError`	If called from code already running on the bridge's own loop (`await` instead — blocking would deadlock).
`KeyboardInterrupt`	Re-raised after cancelling the coroutine.

bounded_gather ¶

bounded_gather(
    coros: Iterable[Coroutine[Any, Any, _T]], *, limit: int
) -> list[_T]

Await coroutines concurrently, capped at limit in flight at once.

Results are returned in the order the coroutines were given (like asyncio.gather), but at most limit run concurrently — the bound is what keeps fan-out (paginated pages, per-entity requests) from stampeding an API into rate limits. If any coroutine raises, the exception propagates and the rest are cancelled.

Parameters:

Name	Type	Description	Default
`coros`	`Iterable[Coroutine[Any, Any, _T]]`	The coroutines to run.	required
`limit`	`int`	Maximum number of coroutines in flight at once (must be >= 1).	required

Returns:

Type	Description
`list[_T]`	The results, ordered to match `coros`.

Raises:

Type	Description
`ValueError`	If `limit` is less than 1.

invoke ¶

invoke(fn: Any, *args: Any, **kwargs: Any) -> Any

Call a sync or async callable uniformly on the event loop.

Async callables are awaited natively; sync ones are offloaded to a worker thread via asyncio.to_thread so they never block the loop. This is how the framework lets data(), destination read/write and fetch-field providers be written as either sync or async while the engine stays async-native.

Parameters:

Name	Type	Description	Default
`fn`	`Any`	The callable to invoke.	required
`*args`	`Any`	Positional arguments forwarded to `fn`.	required
`**kwargs`	`Any`	Keyword arguments forwarded to `fn`.	required

Returns:

Type	Description
`Any`	The callable's result.

data ¶

Data utility helpers.

is_empty ¶

is_empty(data: Any) -> bool

Return whether a value carries no data.

Kept deliberately conservative: only values we can positively confirm are empty count as empty, so callers never skip work on a value they don't understand. None is empty; objects exposing a boolean empty (pandas / polars DataFrames and Series — whose own bool() raises) defer to it; sized containers (lists, dicts, ...) are empty when their length is zero. Anything else (e.g. a lazy generator we must not consume) is treated as non-empty.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The value to inspect; any type, never consumed.	required

Returns:

Type	Description
`bool`	True only when the value is positively confirmed to carry no data.

text ¶

Text processing utilities for naming, casing, and labeling.

validate_key ¶

validate_key(key: str) -> None

Validate that key starts with a letter and contains only letters, numbers, and underscores.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to validate.	required

Raises:

Type	Description
`ValueError`	If the key is invalid.

to_identifier ¶

to_identifier(text: str) -> str

Coerce text into a lowercase identifier (letters, digits, underscores).

Unlike to_snake_case, existing underscores are preserved verbatim (including __ runs used as name separators) — only invalid characters are replaced.

Parameters:

Name	Type	Description	Default
`text`	`str`	The text to coerce.	required

Returns:

Type	Description
`str`	The coerced text.

to_label ¶

to_label(text: str) -> str

Convert text to a human-readable title-cased label.

Parameters:

Name	Type	Description	Default
`text`	`str`	The text to convert.	required

Returns:

Type	Description
`str`	The converted text.

to_snake_case ¶

to_snake_case(text: str) -> str

Convert text to snake_case.

Parameters:

Name	Type	Description	Default
`text`	`str`	The text to convert.	required

Returns:

Type	Description
`str`	The converted text.

time ¶

Date and datetime helpers shared across the framework.

coerce_to_date ¶

coerce_to_date(value: object) -> date

Coerce a value to a datetime.date.

Accepts a date, a datetime (its date part is used), or an ISO-8601 date string. Anything else raises TypeError.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to coerce.	required

Returns:

Type	Description
`date`	The value as a `datetime.date`.

Raises:

Type	Description
`TypeError`	If the value cannot be interpreted as a date.

coerce_to_datetime ¶

coerce_to_datetime(value: object) -> datetime

Coerce a value to a naive-UTC datetime.datetime.

Accepts a datetime, a date (midnight is assumed), or an ISO-8601 datetime string. Anything else raises TypeError.

The result is a UTC label, not an instant: an aware datetime is converted to UTC and stripped of its tzinfo, so comparisons never mix aware and naive values.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to coerce.	required

Returns:

Type	Description
`datetime`	The value as a `datetime.datetime`.

Raises:

Type	Description
`TypeError`	If the value cannot be interpreted as a datetime.

add_months ¶

add_months(value: date, months: int) -> date

Shift a date by months, clamping the day to the target month's length.

Parameters:

Name	Type	Description	Default
`value`	`date`	The date to shift.	required
`months`	`int`	Number of months to add; negative shifts backwards.	required

Returns:

Type	Description
`date`	The shifted date.

assume_utc ¶

assume_utc(value: datetime) -> datetime

Treat a naive timestamp as UTC, leaving an aware one alone.

Parameters:

Name	Type	Description	Default
`value`	`datetime`	The timestamp to normalise, aware or naive.	required

Returns:

Type	Description
`datetime`	An aware datetime; an already-aware value is returned unchanged.

month_start ¶

month_start(value: datetime) -> date

The first day of the UTC calendar month a timestamp falls in.

Naive values are treated as UTC, so a timestamp read back from a database that drops tzinfo attributes to the same month it was written in.

Parameters:

Name	Type	Description	Default
`value`	`datetime`	The timestamp to attribute, aware or naive.	required

Returns:

Type	Description
`date`	The first day of that UTC month, as a date.

imports ¶

Dynamic import helpers.

import_from_path ¶

import_from_path(path: str) -> Any

Import an object from a path.

Two forms are supported:

Dotted path — "module.submodule.ClassName": imports the parent module, then does a single getattr for the final name.
Composite path — "module.submodule:ClassName.nested": the colon marks the module / attribute boundary explicitly. Everything before : is imported as a module, everything after is a chain of getattr calls on the imported object. This is the form emitted by Asset.classpath for source-owned assets so that the asset class can be reached via a class-level descriptor without instantiating its parent source.

Parameters:

Name	Type	Description	Default
`path`	`str`	Dotted or composite import path.	required

Returns:

Type	Description
`Any`	The imported object.

get_object_path ¶

get_object_path(target: Any) -> str

Return the dotted import path for a class or function.

Parameters:

Name	Type	Description	Default
`target`	`Any`	A class or function.	required

Returns:

Type	Description
`str`	Dotted path string like `"module.submodule.ClassName"`.

stats ¶

Statistics over plain samples.

percentile ¶

percentile(
    values: Sequence[float], q: float
) -> float | None

The nearest-rank q-th percentile of values.

Parameters:

Name	Type	Description	Default
`values`	`Sequence[float]`	The sample; order does not matter.	required
`q`	`float`	Percentile in `[0, 100]`.	required

Returns:

Type	Description
`float \| None`	The sample value at that rank, or `None` for an empty sample.