Operation Mapping

dffml.operation.mapping.create_mapping(key: str, value: Any)

Creates a mapping of a given key and value.

Parameters:

• key (str) – The key for the mapping.

• value (Any) – The value for the mapping.

Returns:

A dictionary containing the mapping created.

Return type:

dict

Examples

>>> import asyncio
>>> from dffml import *
>>>
>>> dataflow = DataFlow.auto(create_mapping, GetSingle)
>>> dataflow.seed.append(
...     Input(
...         value=[create_mapping.op.outputs["mapping"].name],
...         definition=GetSingle.op.inputs["spec"],
...     )
... )
>>> inputs = [
...     Input(
...         value="key1", definition=create_mapping.op.inputs["key"],
...     ),
...     Input(
...         value=42, definition=create_mapping.op.inputs["value"],
...     ),
... ]
>>>
>>> async def main():
...     async for ctx, result in MemoryOrchestrator.run(dataflow, inputs):
...         print(result)
>>>
>>> asyncio.run(main())
{'mapping': {'key1': 42}}

dffml.operation.mapping.mapping_extract_value(mapping: Dict[str, Any], traverse: List[str])

Extracts value from a given mapping.

Parameters:

• mapping (dict) – The mapping to extract the value from.

• traverse (list[str]) – A list of keys to traverse through the mapping dictionary and extract the values.

Returns:

A dictionary containing the value of the keys.

Return type:

dict

Examples

>>> import asyncio
>>> from dffml import *
>>>
>>> dataflow = DataFlow.auto(mapping_extract_value, GetSingle)
>>>
>>> dataflow.seed.append(
...     Input(
...         value=[mapping_extract_value.op.outputs["value"].name],
...         definition=GetSingle.op.inputs["spec"],
...     )
... )
>>> inputs = [
...     Input(
...         value={"key1": {"key2": 42}},
...         definition=mapping_extract_value.op.inputs["mapping"],
...     ),
...     Input(
...         value=["key1", "key2"],
...         definition=mapping_extract_value.op.inputs["traverse"],
...     ),
... ]
>>>
>>> async def main():
...     async for ctx, result in MemoryOrchestrator.run(dataflow, inputs):
...         print(result)
>>>
>>> asyncio.run(main())
{'value': 42}