arcaflow_plugin_sdk.plugin module

arcaflow_plugin_sdk.plugin.build_schema(*args: StepType) → SchemaType[source]

This function takes functions annotated with @plugin.step and creates a schema from them.

Parameters:

args – the steps to be added to the schema

Returns:

a callable schema

Example

Imports:

>>> from arcaflow_plugin_sdk import plugin
>>> from dataclasses import dataclass

Create an input dataclass:

>>> @dataclass
... class InputData:
...    name: str

Create an output dataclass:

>>> @dataclass
... class OutputData:
...    message: str

Create the plugin:

>>> @plugin.step(
...     id="hello-world",
...     name="Hello world!",
...     description="Says hello :)",
...     outputs={"success": OutputData},
... )
... def hello_world(params: InputData) -> typing.Tuple[str, typing.Union[OutputData]]:
...     return "success", OutputData("Hello, {}!".format(params.name))

Create the schema from one or more step functions:

>>> plugin_schema = plugin.build_schema(hello_world)

You can now call the step schema directly with data validation:

>>> plugin_schema("hello-world", {"name": "Arca Lot"})
('success', {'message': 'Hello, Arca Lot!'})
arcaflow_plugin_sdk.plugin.run(s: ~arcaflow_plugin_sdk.schema.SchemaType, argv: ~typing.List[str] = ('/home/docs/checkouts/readthedocs.org/user_builds/arcaflow-plugin-sdk-python/envs/latest/lib/python3.9/site-packages/sphinx/__main__.py', '-T', '-E', '-b', 'html', '-d', '_build/doctrees', '-D', 'language=en', '.', '/home/docs/checkouts/readthedocs.org/user_builds/arcaflow-plugin-sdk-python/checkouts/latest/_readthedocs//html'), stdin: ~_io.TextIOWrapper = <_io.TextIOWrapper name='<stdin>' mode='r' encoding='utf-8'>, stdout: ~_io.TextIOWrapper = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, stderr: ~_io.TextIOWrapper = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>) → int[source]

Run takes a schema and runs it as a command line utility. It returns the exit code of the program. It is intended to be used as an entry point for your plugin.

Parameters:

  • s – the schema to run

  • argv – command line arguments

  • stdin – standard input

  • stdout – standard output

  • stderr – standard error

Returns:

exit code

arcaflow_plugin_sdk.plugin.signal_handler(id: str, name: str, description: str, icon: str | None = None) → Callable[[Callable[[InputT], None]], SignalHandlerType][source]

@plugin.signal_handler is a decorator that takes a function with a single parameter and creates a schema for it that you can use with plugin.step.

Parameters:

  • id – The identifier for the signal handler.

  • name – The human-readable name for the signal handler.

  • description – The human-readable description for the signal handler.

  • icon – SVG icon for this signal.

Returns:

A schema for the signal.

arcaflow_plugin_sdk.plugin.step(id: str, name: str, description: str, outputs: Dict[str, Type], icon: str | None = None) → Callable[[Callable[[InputT], OutputT]], StepType][source]

@plugin.step is a decorator that takes a function with a single parameter and creates a schema for it that you can use with plugin.build_schema.

Parameters:

  • id – The identifier for the step.

  • name – The human-readable name for the step.

  • description – The human-readable description for the step.

  • outputs – A dict linking response IDs to response object types.

  • icon – SVG icon for this step.

Returns:

A schema for the step.

arcaflow_plugin_sdk.plugin.step_with_signals(id: str, name: str, description: str, outputs: Dict[str, Type], signal_handler_method_names: List[str], signal_emitters: List[SignalSchema], step_object_constructor: Callable[[], StepObjectT], icon: str | None = None) → Callable[[Callable[[StepObjectT, InputT], OutputT]], StepType][source]

@plugin.step_with_signals is a decorator that takes a class and a method with a single parameter (plus self) and creates a schema for it that you can use with plugin.build_schema.

Parameters:

  • id – The identifier for the step.

  • name – The human-readable name for the step.

  • description – The human-readable description for the step.

  • outputs – A dict linking response IDs to response object types.

  • signal_handler_method_names – A list of methods for all signal handlers.

  • signal_emitters – A list of signal schemas for signal emitters.

  • step_object_constructor – A constructor lambda for the object with the step and signal methods.

  • icon – SVG icon for this step.

Returns:

A schema for the step.