`extended_data.inputs.main`¶

Tier 3 directed input processing for the extended-data package.

This module manages inputs from environment variables, stdin, and explicit mappings. It can merge, replace, snapshot, freeze, and thaw input state while keeping public snapshots in Tier 2 containers. It also decodes inputs from JSON, YAML, and Base64 and coerces scalar values through Tier 1 type primitives.

Module Contents¶

Classes¶

InputProvider

Manage directed inputs from environment variables, stdin, or mappings.

Data¶

_MISSING

API¶

extended_data.inputs.__main__._MISSING = 'object(...)'¶

class extended_data.inputs.__main__.InputProvider(inputs: collections.abc.Mapping[str, Any] | None = None, from_environment: bool = True, from_stdin: bool = False, env_prefix: str | None = None, strip_env_prefix: bool = False)¶

Manage directed inputs from environment variables, stdin, or mappings.

Attributes: inputs (ExtendedDict): Dictionary to store inputs. frozen_inputs (ExtendedDict): Dictionary to store frozen inputs. from_stdin (bool): Flag indicating if inputs were read from stdin. merger (Merger): Object to manage deep merging of dictionaries.

Initialization

Initializes the InputProvider with the provided inputs.

Optionally loading additional inputs from environment variables and stdin.

Args: inputs (Mapping[str, Any] | None): Initial inputs to be processed. from_environment (bool): Whether to load inputs from environment variables. from_stdin (bool): Whether to load inputs from stdin. env_prefix (str | None): Optional prefix to filter environment variables. strip_env_prefix (bool): Whether to strip the prefix from environment keys.

static _normalize_inputs(inputs: collections.abc.Mapping[str, Any] | None) → dict[str, Any]¶

static _filtered_environment(env: collections.abc.Mapping[str, str], *, env_prefix: str | None, strip_prefix: bool) → dict[str, str]¶

_merge_inputs(base: collections.abc.Mapping[str, Any], incoming: collections.abc.Mapping[str, Any]) → dict[str, Any]¶

static _load_from_stdin() → dict[str, Any]¶

static _coerce_text(value: Any) → Any¶

static _format_available_keys(inputs: collections.abc.Mapping[str, Any]) → str¶: Format available input keys without exposing their values.

static _return_value(value: Any, *, as_extended: bool) → Any¶

get_input(k: str, default: Any | None = None, required: bool = False, is_bool: bool = False, is_integer: bool = False, is_float: bool = False, is_path: bool = False, is_datetime: bool = False, as_extended: bool = False) → Any¶

Retrieves an input by key, with options for type conversion and default values.

This method leverages Extended Data core utilities for robust type conversions, including support for Path objects, datetime parsing, and numeric conversions.

Args: k (str): The key for the input. default (Any | None): The default value if the key is not found. required (bool): Whether the input is required. Raises an error if required and not found. is_bool (bool): Whether to convert the input to a boolean. is_integer (bool): Whether to convert the input to an integer. is_float (bool): Whether to convert the input to a float. is_path (bool): Whether to convert the input to a Path object. is_datetime (bool): Whether to convert the input to a datetime object. as_extended (bool): Whether to wrap the returned value in Tier 2 containers.

Returns: Any: The retrieved input, potentially converted or defaulted.

decode_input(k: str, default: Any | None = None, required: bool = False, decode_from_json: bool = False, decode_from_yaml: bool = False, decode_from_base64: bool = False, allow_none: bool = True, as_extended: bool = False) → Any¶

Decodes an input value, optionally from Base64, JSON, or YAML.

Args: k (str): The key for the input. default (Any | None): The default value if the key is not found. required (bool): Whether the input is required. Raises an error if required and not found. decode_from_json (bool): Whether to decode the input from JSON format. decode_from_yaml (bool): Whether to decode the input from YAML format. decode_from_base64 (bool): Whether to decode the input from Base64. allow_none (bool): Whether to allow None as a valid return value. as_extended (bool): Wrap decoded container values in Tier 2 Extended Data containers.

Returns: Any: The decoded input, potentially converted or defaulted.

freeze_inputs() → extended_data.containers.mappings.ExtendedDict¶

Freezes the current inputs, preventing further modifications until thawed.

Returns: ExtendedDict: The frozen inputs.

thaw_inputs() → extended_data.containers.mappings.ExtendedDict¶

Thaws the inputs, merging the frozen inputs back into the current inputs.

Returns: ExtendedDict: The thawed inputs.

snapshot_inputs(*, frozen: bool = False) → extended_data.containers.mappings.ExtendedDict¶