hyped.core package

Subpackages

Submodules

Module contents

The core package for defining and executing data flows as directed acyclic graphs (DAGs).

This package provides the necessary classes and methods to construct, manage, and execute complex data processing workflows. It uses a graph-based approach where each node represents a data processor, and edges represent the flow of data between processors.

Modules:

  • executor: Manages the execution of the data flow graph.

  • flow: Provides the high-level interface for defining data processing workflows.

  • graph: Defines the structure of the data flow graph and its components.

  • optim: Defines an Optimizer to optimize the graph of the data flow.

  • typing: Defines core type aliases including those used to define node interfaces.

  • nodes: Defines the base classes for nodes of the DAG.

  • features: The feature system.

  • ops: Implementation of core operations.

  • testing: Defines base classes to implement node tests.

While these modules are crucial for processor development, they are not intended for direct use by end users interacting with the high-level data flow interface.

class hyped.core.BaseDataAggregator(*args: Any, **kwargs: Any)[source]

Bases: BaseNode[C], ABC

Base class for data aggregators.

This class serves as the base for all data aggregators, defining the necessary interfaces and methods for implementing custom aggregators. Subclasses must implement the extract and update methods, which define the logic for retrieving and updating aggregated values in the data flow graph.

Extracted = ~Extract
State = ~State
Value = ~Value
abstractmethod async extract(ctx: RunContext, *args: Feature, **kwargs: Feature) → Extracted[source]

Extract necessary values from the inputs for aggregation.

Parameters:

  • ctx (RunContext) – The run context object.

  • *args (Feature) – Positional input arguments.

  • **kwargs (Feature) – Keyword input arguments.

Returns:

The extracted context values required for aggregation.

Return type:

Extract

abstractmethod seed(ctx: RunContext) → tuple[Value, State][source]

Compute the seed aggregation value and state.

Note that the seed() function is not part of the data flow execution but of the initialization process. Therefore, it is called before initialize() of the node and the ctx.session will be None.

Parameters:

ctx (RunContext) – The run context object with session=None.

Returns:

The initial value and state for the aggregator.

Return type:

tuple[Value, State]

property signature: Signature

Retrieve the signature for the aggregator node.

This method constructs a signature for the aggregator node by combining the parameters of the extract() method with the return annotation of the update() method. The ctx parameter is excluded from the parameter list, ensuring that the signature reflects only the feature inputs relevant to the aggregation process.

The return annotation of the signature is derived from the update() method, representing the aggregated feature type produced by the aggregator node.

Returns:

A constructed signature for the aggregator node, with parameters from extract() (excluding ctx) and a return annotation based on the feature type from update().

Return type:

inspect.Signature

Raises:

  • TypeError – If the return annotation of update() does not represent

  • a valid aggregated feature type. –

abstractmethod async update(ctx: RunContext, val: Value, state: State, extracted: Extracted) → tuple[Value, State][source]

Update the aggregation value and context.

Parameters:

  • ctx (RunContext) – The run context object.

  • val (Value) – The current aggregation value.

  • state (State) – The current aggregation state.

  • extracted (Extract) – The values extracted from the input batch.

Returns:

The updated aggregation value and state.

Return type:

tuple[Value, State]

class hyped.core.BaseDataAggregatorConfig[source]

Bases: BaseNodeConfig

Base configuration class for data aggregators.

This class serves as the base configuration class for data aggregators. It inherits from BaseConfig, providing basic configuration functionality for data aggregation tasks.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class hyped.core.BaseDataAugmentor(*args: Any, **kwargs: Any)[source]

Bases: BaseNode[C], ABC

Base class for data augmentors in a data flow graph.

This class represents a data augmentor node in a data flow graph. Data augmentors modify or generate new samples from existing ones, which can include filtering or creating new data points. Subclasses of BaseDataAugmentor must implement either the process or the batch_process method to define how the augmentation is applied to the input data.

infer_output_partition(ctx: RunContext, partition: str) → str[source]

Determine the output partition of the augmentater.

By default, data augmentors point to their own partition. This method reuses the node ID of the augmentor as the output partition ID.

Note that this function is not part of the data flow execution but of the initialization process. Therefore, it is called before initialize() of the node and the ctx.session will be None.

Parameters:

  • ctx (RunContext) – Execution context for the node.

  • partition (PartitionId) – The ID of the input partition, i.e. the partition that the node is assigned to.

Returns:

The output partition ID, corresponding to the node ID of the augmentor.

Return type:

PartitionId

abstractmethod process(ctx: RunContext, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → AsyncIterable[Feature | Any | list[Any] | Scalar | Array][source]
abstractmethod process(ctx: RunContext, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Iterable[Feature | Any | list[Any] | Scalar | Array]
abstractmethod process(ctx: RunContext, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → tuple[Feature | Any | list[Any] | Scalar | Array, list[int]]
abstractmethod process(ctx: RunContext, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → tuple[Feature | Any | list[Any] | Scalar | Array, list[int]]

Defines the augmentation logic to be applied.

This method should be overridden by subclasses to define the augmentation logic. It may either be synchronous or asynchronous, depending on the subclass.

Parameters:

  • ctx (RunContext) – Context information for the data augmentor’s execution.

  • *args (Feature) – Positional input arguments.

  • **kwargs (Feature) – Keyword arguments.

Returns:

  • Iterable[Feature]: If the process function is synchronous, it returns an iterable of augmented output samples, potentially producing multiple outputs per input.

  • AsyncIterable[Feature]: If the process function is asynchronous, it returns an async iterable of augmented output samples, following the same logic as the synchronous mode.

  • tuple[Feature, TraceIndexList]: If the process function operates in batched mode, it returns:

    • Feature: A batch of augmented output samples.

    • TraceIndexList: A list of trace indices mapping each output sample to the corresponding source sample in the input batch. Specifically, the i-th output sample originates from the trace_index[i]-th input example.

Return type:

Union[Iterable[Feature], AsyncIterable[Feature], tuple[Feature, TraceIndexList]]

async run(ctx: RunContext, arrays: dict[str, Array]) → tuple[Array, list[int]][source]

Execute the main processing logic for the data augmentor.

This method serves as the primary entry point for processing data within a data flow graph, returning both the processed outputs and their associated trace indices. It orchestrates the execution of the process method according to the configured ProcessMode, handling input preparation, processing, and output finalization. In detail, the workflow is:

  1. Determine the processing mode (ProcessMode) based on the process method’s configuration.

  2. Prepare the input data using the ProcessMode.prepare method.

  3. Apply the process method to the prepared inputs. If the method is asynchronous, the outputs are awaited using asyncio.gather.

  4. Depending on whether the processing is batched: - If batched:

    1. Separate the outputs and their associated trace indices from the process method’s results.

    2. Concatenate the trace indices and finalize the outputs as a PyArrow array.

    • If non-batched: a. Collect all outputs and trace indices from the process method,

      consuming asynchronous or synchronous iterators as appropriate.

      1. Chain the outputs and finalize them as a PyArrow array.

      2. Build a trace index list that maps each output sample back to its corresponding input.

Parameters:

  • ctx (RunContext) – The execution context containing.

  • arrays (dict[str, pa.Array]) – A dictionary mapping input names to PyArrow arrays, representing the input data to be processed.

Returns:

A tuple containing the processed output as a PyArrow array and a list of trace indices mapping the output samples to their respective input sources.

Return type:

tuple[pa.Array, TraceIndexList]

property signature: Signature

Get the signature of the process() method.

Returns the signature of the process() method with the ctx parameter removed, keeping only the feature inputs modeled in the data flow graph.

Returns:

The signature of the process() method excluding the ctx parameter.

Return type:

inspect.Signature

Raises:

TypeError – If the return type of process() is not an iterable.

class hyped.core.BaseDataAugmentorConfig[source]

Bases: BaseNodeConfig

Base configuration class for data augmentors.

This class serves as the base configuration for data augmentors, inheriting from BaseNodeConfig to provide configuration functionality specifically for data augmentation tasks.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class hyped.core.BaseDataProcessor(*args: Any, **kwargs: Any)[source]

Bases: BaseNode[C], ABC

Base class for data processors in a data flow graph.

This class serves as the base for all data processors, representing nodes in a data flow graph. Subclasses of BaseDataProcessor implement specific process functions that map input features to output features. Custom data processors must either override the batch_process() method or the process() method to define their processing logic.

abstractmethod process(ctx: RunContext, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Feature | Any | list[Any] | Scalar | Array[source]

Process function.

This method should be implemented by subclasses to define the processing logic. It may either be synchronous or asynchronous, depending on the subclass.

Parameters:

  • ctx (RunContext) – The context for the current process call.

  • *args (Feature) – Positional feature arguments.

  • **kwargs (Feature) – Keyword feature arguments.

Returns:

The resulting processed feature.

Return type:

Feature

async run(ctx: RunContext, arrays: dict[str, Array]) → Array[source]

Execute the main processing logic for the data processor.

This method serves as the primary entry point for processing data within a data flow graph. It orchestrates the execution of the process method according to the configured ProcessMode, handling input preparation, processing, and output finalization. In detail the workflow is:

  1. Determine the processing mode (ProcessMode) based on the process method’s configuration.

  2. Prepare the input data using the ProcessMode.prepare method.

  3. Apply the process method to the prepared inputs. If the method is asynchronous, the outputs are awaited using asyncio.gather.

  4. Finalize the outputs using the ProcessMode.finalize method, which ensures that the results are correctly formatted as an PyArrow array.

Parameters:

  • ctx (RunContext) – The execution context containing.

  • arrays (dict[str, pa.Array]) – A dictionary mapping input names to PyArrow arrays, representing the input data to be processed.

Returns:

The processed output as a PyArrow array.

Return type:

pa.Array

property signature: Signature

Get the signature of the process() method.

Returns the signature of the process() method with the ctx parameter removed, keeping only the feature inputs modeled in the data flow graph.

Returns:

The signature of the process() method excluding the ctx parameter.

Return type:

inspect.Signature

class hyped.core.BaseDataProcessorConfig[source]

Bases: BaseNodeConfig

Base configuration class for data processors.

This class serves as the base configuration class for data processors. It inherits from BaseNodeConfig, a Pydantic model, providing basic configuration functionality for data processing tasks.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class hyped.core.DataFlow(features: None | Features = None)[source]

Bases: AbstractDataFlow, Generic[T]

Data Flow.

The DataFlow class allows users to create and manage directed acyclic graphs (DAGs) of data processors, facilitating complex data transformations and processing pipelines. Users can easily define source features, build sub-flows for specific outputs, and apply these workflows to batches of data or entire HuggingFace datasets.

This class integrates various components such as the data flow graph and the executor to provide a seamless experience for processing data. It handles the internal state management, execution scheduling, and data flow dependencies to ensure efficient and accurate data processing.

U = ~U
apply(ds: Dataset, collect: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False, keep_in_memory: bool = False, load_from_cache_file: bool = True, writer_batch_size: int = 1000, num_proc: None | int = None, desc: None | str = None) → Dataset[source]
apply(ds: Dataset, collect: Feature | dict[str, Any], aggregate: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False, keep_in_memory: bool = False, load_from_cache_file: bool = True, writer_batch_size: int = 1000, num_proc: None | int = None, desc: None | str = None) → tuple[Dataset, dict[str, Any]]
apply(ds: ItDataset, collect: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False) → ItDataset
apply(ds: ItDataset, collect: Feature | dict[str, Any], aggregate: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False) → tuple[ItDataset, dict[str, Any]]

Apply the data flow graph to a dataset.

This method processes a given dataset or iterable dataset using the data flow graph, executing transformations based on the collect and optional aggregate features. The behavior of the processing depends on whether the dataset is an in-memory or streamed dataset (i.e. datasets.Dataset or datasets.IterableDataset)).

Parameters:

  • ds (Dataset | IterableDataset) – The dataset to which the data flow graph will be applied.

  • collect (Feature | dict[str, Any]) – The collect feature, which defines the primary transformations applied to the dataset.

  • aggregate (None | Feature | dict[str, Any]) – An optional aggregate feature, which applies additional aggregate-level transformations. If not provided, aggregation is skipped.

  • debug (bool) – If True, executes debug nodes allowing for inspection of intermediate data. Defaults to False.

  • batch_size (int) – The number of samples to process in a batch. Defaults to 1000.

  • drop_last_batch (bool) – Whether to drop the last batch if it is smaller than the specified batch size. Defaults to False.

  • keep_in_memory (bool) – If True, the resulting dataset is kept in memory. Defaults to False. Only applies for datasets.Dataset.

  • load_from_cache_file (bool) – Whether to load the resulting dataset from cache files when possible. Defaults to True. Only applies for datasets.Dataset.

  • writer_batch_size (int) – Batch size for writing results to cache files. Defaults to 1000. Only applies for datasets.Dataset.

  • num_proc (None | int) – The number of processes to use for parallel processing. Defaults to None, which disables multiprocessing. Only applies for datasets.Dataset.

  • desc (Optional[str], optional, only for Dataset) – A description for the progress bar displayed during processing. Defaults to None. Only applies for datasets.Dataset.

Returns:

The transformed dataset matching the input dataset type and a dictionary containing the aggregation results when aggregation is applied, i.e. the aggregate input is specified.

Return type:

Dataset | ItDataset | tuple[Dataset, dict[str, Any]] | tuple[ItDataset, dict[str, Any]]

build(collect: Feature | dict[str, Any], aggregate: None | Feature | dict[str, Any] = None, *, aggregation_manager: DataAggregationManager | None = None, debug: bool = True) → ExecutableDataFlow[source]

Build an executable data flow for computing and collecting features.

This method constructs a read-only, executable representation of the data flow for a specified collect feature. Optionally, an aggregate feature can also be included for dataset-wide computations. The resulting data flow is optimized for execution.

Parameters:

  • collect (Feature | dict[str, Any]) – The feature to be computed and collected.

  • aggregate (None | Feature) – An optional feature for computing aggregated values across the dataset.

  • *

  • aggregation_manager (DataAggregationManager | None) – The data aggregation manager instance.

  • debug (bool) – If True, includes debug nodes in the built data flow, allowing for inspection of intermediate data. Defaults to True.

Returns:

An executable data flow that encapsulates the graph, collect feature, and optional aggregate feature.

Return type:

ExecutableDataFlow[T]

Raises:

  • RuntimeError – If the collect feature does not belong to the current data flow graph.

  • RuntimeError – If the aggregate feature does not belong to the current data flow graph.

collect(collect: dict[dict[str, dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | list[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | tuple[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | Any]) → _MappingFeature[source]
collect(collect: list[dict[str, dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | list[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | tuple[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | Any] | tuple[dict[str, dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | list[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | tuple[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | Any]) → SequenceFeature
collect(collect: Any) → Feature

Add a collect node to the data flow.

The collect() method processes a nested structure (e.g., dicts, lists, tuples) of constants and features and adds a corresponding collect node to the data flow graph. Constants are converted to constant nodes, while features are directly incorporated.

Parameters:

collect (NestedType[Any]) – A nested structure of constants, Feature objects, and Reference instances. The structure may include dictionaries, lists, and tuples, where constants are automatically added as constant nodes.

Returns:

A feature or nested structure of features representing

the collected data.

Return type:

Mapping | Sequence | Feature

Raises:

  • NotImplementedError – If an empty sequence (list or tuple) is encountered.

  • TypeError – If an unsupported type is encountered in the nested structure.

const(value: int) → Int64Feature[source]
const(value: float) → Float64Feature
const(value: bool) → BoolFeature
const(value: str) → StringFeature
const(value: dict[str, Any]) → _MappingFeature
const(value: list[Any] | tuple[Any]) → SequenceFeature
const(value: Any, feature_type: type[U]) → U

Add a constant value as a node to the data flow.

This method allows the user to add a constant value as a node in the data flow graph. The constant can either be associated with a specific feature type or have its data type inferred from the provided value.

Parameters:

  • value (Any) – The constant value to add to the data flow graph.

  • feature_type (None | type[Feature]) – The explicit feature type for the value. If provided, the data type will be validated and inferred from this type. If not provided, the data type is inferred from the value.

Returns:

The feature representation of the constant node added to the graph.

Return type:

Feature

Raises:

TypeError – If the provided feature_type is invalid or if the value cannot be validated against the feature_type.

property depth: int

Computes the total depth of the data flow graph.

The depth is defined as the maximum level of any node in the graph, where the root node has a depth of 0. This property calculates the depth by finding the maximum depth attribute among all nodes in the graph.

Returns:

The total depth of the graph.

Return type:

int

classmethod deserialize(data: str, debug: bool = True) → ExecutableDataFlow[source]

Deserializes a JSON string into an ExecutableDataFlow instance.

This method parses a JSON string into a dictionary, validates its structure, and uses the data to reconstruct the executable data flow, including references to the collection and aggregation nodes.

Parameters:

  • data (str) – The JSON string representing the serialized executable data flow.

  • debug (bool) – If True, the deserialized graph will include debug nodes present in the serialized flow. If False, debug nodes will be omitted from the deserialized graph. Defaults to True.

Returns:

The deserialized executable data flow.

Return type:

ExecutableDataFlow

Raises:

ValueError – If the input JSON string does not contain the required keys (“graph”, “collect”, and “aggregate”).

property source: T

Get the source features.

Returns:

The reference to the source features.

Return type:

T

property width: int

Computes the maximum width of the data flow graph.

The width is defined as the maximum number of nodes present at any single depth level in the graph. This property calculates the width by grouping nodes by their depth and finding the largest group.

Returns:

The maximum width of the graph.

Return type:

int

class hyped.core.DataFlowModule(debug: bool = True)[source]

Bases: ABC

Base class for defining reusable data processing modules.

A DataFlowModule encapsulates a data processing workflow. Subclasses must implement the call() method, which defines the core logic of the module.

abstractmethod call(*args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Feature | Any | list[Any] | Scalar | Array[source]

Defines the core logic of the data processing module.

Subclasses must implement this method to specify how input features are processed to produce output features. The signature of this method determines the input features of the data flow, and the return type determines the output feature.

Within this method, the flow property provides access to a DataFlow instance, giving access to functionality like ~.flow.DataFlow.collect.

Parameters:

  • *args (Feature) – Positional input features.

  • **kwargs (Feature) – Keyword input features.

Returns:

The output feature produced by the module.

Return type:

Feature

property flow: DataFlow | ExecutableDataFlow

The executable data flow for this module.

This property lazily builds and returns the data flow instance.

  • Within the call() method, this returns a mutable DataFlow instance, giving access to functionality like ~.flow.DataFlow.collect.

  • Outside of call(), this returns the build ExecutableDataFlow, which represents the fully constructed and optimized data flow, ready for execution.

Returns:

The data flow.

Return type:

DataFlow | ExecutableDataFlow

plot(node_format: str | Template = "[{{ node_id[:4] }}] {% if node_type == 'SOURCE_NODE' %}Source{% else %}{{ node_object }}{% endif %}", with_edge_labels: bool = True, edge_font_size: int = 6, node_font_size: int = 6, node_size: int = 5000, arrowsize: int = 25, color_map: dict[NodeType, str] = {}, legend: bool = True, legend_fontsize: int = 6, ax: None | Axes = None) → Axes[source]

Plot a data flow graph.

Parameters:

  • flow (DataFlow) – The data flow to plot.

  • node_format (str | Template) – The jinja template used to generate node labels.

  • with_edge_labels (bool) – Whether to include labels on the edges. Defaults to True.

  • edge_font_size (int) – The font size for edge labels. Defaults to 6.

  • node_font_size (int) – The font size for node labels. Defaults to 6.

  • node_size (int) – The size of the nodes. Defaults to 5_000.

  • arrowsize (int) – The size of the arrows on the edges. Defaults to 25.

  • color_map (dict[None | type, str]) – indicate custom color scheme based on the processor type. None refers to the source node.

  • legend (bool) – Whether to add a legend of the node types to the axes. Defaults to True.

  • legend_fontsize (int) – The font size for the legend. Defaults to 6.

  • ax (Optional[plt.Axes]) – Matplotlib axes object to draw the plot on. Defaults to None.

Returns:

The Matplotlib axes object with the plot.

Return type:

plt.Axes

class hyped.core.NodeProtocol(*args, **kwargs)[source]

Bases: Protocol, Generic[Params, Return]

Protocol for node-like objects in a data flow graph.

This protocol defines the interface for nodes within a data flow graph, including methods for calling nodes with or without an explicit data flow. The NodeProtocol is parameterized by Params, defining the arguments, and Return, defining the return type.

call(*args: ~Params, **kwargs: ~Params) → Return[source]
call(flow: AbstractDataFlow, *args: ~Params, **kwargs: ~Params) → Return
class hyped.core.RunContext(session: None | RunSession, node_id: str, index: int | list[int], rank: int, input_dtype: MappingType, output_dtype: None | DType, target_batch_size: None | int)[source]

Bases: object

Context information for the node execution.

Serves as an identifier for the specific call to the node within the data flow graph. This is particularly useful when a single node class is used multiple times in a data flow, as RunContext identifies the specific instance of the node call, i.e., the specific node in the flow graph.

index: int | list[int]

The index or list of indices associated with the processor execution.

This attribute is used to track the position or set of positions for processing data within a specific processor call. It could represent a single index or a list of indices, depending on how the data is partitioned or processed.

input_dtype: MappingType

The expected input data type for the processor.

This attribute defines the type of the data that the processor is designed to handle as input. It typically maps the input data’s structure or schema, providing context for how the data should be processed.

node_id: str

The id of the node in the data flow graph.

This attribute serves as a unique identifier for the context, ensuring each processor call can be distinctly recognized within the flow graph.

output_dtype: None | DType

The type of data the processor will produce as output.

This attribute defines the expected structure or type of the output that the processor will generate. It provides information about the transformation or processing that the input data undergoes and the format of the resulting data.

rank: int

The rank or position of the processor in a parallelized execution.

This attribute identifies the specific rank or position of the processor in a parallelized system (e.g., in distributed or multi-threaded processing). The rank determines the processor’s order or responsibility for a portion of the data during execution.

session: None | RunSession

The run session instance.

target_batch_size: None | int

The expected target batch size.

This attribute defines the batch size of the target partition of the node if set.

class hyped.core.ValidationSession[source]

Bases: object

A lightweight session context manager for managing key-value contextual data.

This class provides a simple way to store, retrieve, and clear context during a specific scope of execution. Contexts are stored as key-value pairs in an internal dictionary, which is cleared automatically when the top-level session ends.

clear_context(key: None | Hashable = None) → None[source]

Clear one or all contexts.

Parameters:

key (None | Hashable) – The key of the context entry to clear. If None, all contexts are cleared. Defaults to None.

get_context(key: Hashable) → Any[source]

Retrieve the context value associated with a specific key.

Parameters:

key (Hashable) – The key for the context entry to retrieve.

Returns:

The value associated with the key, or None if the key is not found.

Return type:

Any

property session_id: UUID

The unique session ID, generated on demand if not already initialized.

Returns:

A unique identifier for the session.

Return type:

UUID

set_context(key: Hashable, value: Any) → None[source]

Set a context value associated with a specific key.

Parameters:

  • key (Hashable) – The key for the context entry.

  • value (Any) – The value to associate with the key.

hyped.core.process_mode(batched: bool = False, backend: Literal['python', 'arrow'] = 'python') → Callable[[F], F][source]

Decorator to specify the processing mode of a data processing function.

This decorator associates a function with a ProcessMode, specifying how the function handles its inputs and outputs during execution. The mode defines whether the function operates in batched or non-batched mode and which backend is used for processing (e.g., “python” or “arrow”).

Use the @process_mode(...) decorator to annotate methods or functions that perform data processing. The decorator ensures the function is tagged with the appropriate processing mode, which can be validated or used during runtime.

Parameters:

  • batched (bool) – Indicates if the function processes data in batches.

  • backend (Backend) – Specifies the backend used for processing.

Returns:

The decorator function that associates a function with the specified ProcessMode instance.

Return type:

Callable[[F], F]