`yggdrasil.dataclasses.safe_function`¶

safe_function ¶

Signature-driven argument coercion for Python callables.

Wraps :func:yggdrasil.data.cast.convert against the function's own signature so callers can pass string-shaped inputs (CLI argv, Databricks job parameters, HTTP form fields, environment variables) and have them reach the function as the annotated Python type — no manual int(os.environ[...]) / datetime.fromisoformat(...) sprinkled at every call site.

Two entry points:

:func:check_function_args is the pure utility: takes a function and a raw (args, kwargs) pair, returns coerced (args, kwargs) ready to splat back into the call. Use it when you have the raw inputs in hand and want to coerce once before dispatching.
:func:checkargs is the decorator built on top: wraps a function so every call is routed through :func:check_function_args before the real call. Use it on functions whose inputs cross a string boundary every time.

PEP 563 (from __future__ import annotations) and string-quoted annotations are resolved to real types via :func:inspect.get_annotations with eval_str=True, so the coercion sees int rather than the string literal "int". Unannotated parameters pass through untouched — there's nothing to target. *args and **kwargs annotations are honored per element.

Also bundles :func:describe_signature / :func:format_signature so callers that want to display or stamp signature metadata (job task descriptions, OpenAPI shapes, debug logs) reach for the same surface.

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.safe_function¶