Operation Io

class dffml.operation.io.AcceptUserInput(*args, **kwargs)

Accept input from stdin using python input()

Returns:

A dictionary containing user input.

Return type:

dict

Examples

The following example shows how to use AcceptUserInput. (Assumes that the input from stdio is “Data flow is awesome”!)

>>> import asyncio
>>> from dffml import *
>>>
>>> dataflow = DataFlow.auto(AcceptUserInput, GetSingle)
>>> dataflow.seed.append(
...     Input(
...         value=[AcceptUserInput.op.outputs["InputData"].name],
...         definition=GetSingle.op.inputs["spec"],
...     )
... )
>>>
>>> async def main():
...     async for ctx, results in MemoryOrchestrator.run(dataflow, {"input": []}):
...         print(results)
>>>
>>> asyncio.run(main())
Enter the value: {'UserInput': 'Data flow is awesome'}

CONTEXT

alias of AcceptUserInputContext

class dffml.operation.io.AcceptUserInputContext(parent: OperationImplementation, ctx: BaseInputSetContext, octx: BaseOrchestratorContext)

async run(inputs: Dict[str, Any]) → Dict[str, Any]

Implementation of the operation goes here. Should take and return a dict with keys matching the input and output parameters of the Operation object associated with this operation implementation context.

async dffml.operation.io.print_output(data: Any)

Print the output on stdout using python print()

Parameters:

data (Any) – A python literal to be printed.

Examples

The following example shows how to use print_output.

>>> import asyncio
>>> from dffml import *
>>>
>>> dataflow = DataFlow.auto(print_output)
>>> inputs = [
...     Input(
...         value="print_output example", definition=print_output.op.inputs["data"]
...     )
... ]
>>>
>>> async def main():
...     async for ctx, results in MemoryOrchestrator.run(dataflow, inputs):
...         pass
>>>
>>> asyncio.run(main())
print_output example