arcaflow_plugin_sdk.schema module

This module provides the tools to build a schema, either by hand, or automatically from type hints and annotations.

Using

For using this module please check the documentation located at https://arcalot.github.io/arcaflow/creating-plugins/python/

Contributing

This module is structured by using region folding delimiters (e.g. # region SomeRegion). You can use an editor that supports these regions to collapse them together for easier access. The module has the following regions:

Exceptions

This region defines a number of exceptions that can be raised either while building a schema or while working with input and output.

Annotations

Annotations can be applied to dataclass fields to convey extra information (e.g. minimum or maximum values). These annotations are used when building a schema. Note that, unlike tools like Pydantic we do not validate the dataclass fields in-situ. Instead, we only add metadata information which is verified when the data is serialized or unserialized. This late verification is done to avoid confusing error messages when using third party code that may be constructing the dataclasses with intermediate states that are temporarily invalid.

Type aliases

This section defines a number of type aliases for use in the schema itself.

Schema

This section contains the data model for the Arcaflow schema definition itself. The schema is self-describing and can be serialized and unserialized with the same tools that you use to serialize or unserialize data that conforms to the schema. The plugin system uses these dataclasses to construct the SCOPE_SCHEMA and SCHEMA_SCHEMA variables, which you can use to serialize and unserialize a scope definition or an entire schema.

Types

Types are the implementations of a schema. They add serialization, validation, and unserialization to the schema itself. When working with schemas in practice you will want to work with types unless you don’t need to unserialize data.

Build

This region holds the tools to automatically build a schema from annotations. Use build_object_schema to build a schema from a dataclass.

Note that the built schema will be a type, not a schema, so it is possible to use these schemas for serializing and unserializing data.

Schema schemas

This region holds two variables, SCOPE_SCHEMA and SCHEMA_SCHEMA. You can use these variables to work with schemas themselves. For example, you can create a client that calls Arcaflow plugins and use these classes to unserialize schemas that the plugins send.

class arcaflow_plugin_sdk.schema.AbstractType

Bases: Generic[TypeT]

This class is an abstract class describing the methods needed to implement a type.

abstract serialize(data: TypeT, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

abstract unserialize(data: Any, path: Tuple[str] = ()) → TypeT

This function takes the underlying raw data and decodes it into the underlying advanced data type (e.g. dataclass) for usage.

Parameters:

• data – the raw data.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the advanced datatype.

Raises:

ConstraintException – if the passed data was not valid.

abstract validate(data: TypeT, path: Tuple[str] = ())

This function validates an already unserialized data type and raises an exception if it does not match the type definition.

Parameters:

• data – the unserialized data.

• path – the path that lead to this validation call, in order to produce a nice error message

Raises:

ConstraintException – if the passed data was not valid.

class arcaflow_plugin_sdk.schema.AnySchema

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class stores the details of the “any” type, which allows all lists, dicts, integers, floats, strings, and bools in a value.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create an any schema:

>>> ref = schema.AnySchema()

class arcaflow_plugin_sdk.schema.AnyType

Bases: AnySchema, AbstractType

This type is a serializable version of the “any” type.

Examples:

Import:

>>> from arcaflow_plugin_sdk import schema

Create an AnyType: >>> t = schema.AnyType()

Unserialize a value: >>> t.unserialize(“Hello world!”) ‘Hello world!’ >>> t.unserialize(1) 1 >>> t.unserialize(1.0) 1.0 >>> t.unserialize(True) True >>> t.unserialize({“message”: “Hello world!”}) {‘message’: ‘Hello world!’} >>> t.unserialize([“Hello world!”]) [‘Hello world!’]

Validate a value: >>> t.validate(“Hello world!”) >>> t.validate(1) >>> t.validate(1.0) >>> t.validate(True) >>> t.validate({“message”: “Hello world!”}) >>> t.validate([“Hello world!”])

Serialize a value: >>> t.serialize(“Hello world!”) ‘Hello world!’ >>> t.serialize(1) 1 >>> t.serialize(1.0) 1.0 >>> t.serialize(True) True >>> t.serialize({“message”: “Hello world!”}) {‘message’: ‘Hello world!’} >>> t.serialize([“Hello world!”]) [‘Hello world!’]

Not everything is accepted: >>> from dataclasses import dataclass >>> @dataclass … class TestClass: … pass >>> t.unserialize(TestClass()) Traceback (most recent call last): … arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Unsupported data type for ‘any’ type: TestClass

serialize(data: Any, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

unserialize(data: Any, path: Tuple[str] = ()) → Any

This function takes the underlying raw data and decodes it into the underlying advanced data type (e.g. dataclass) for usage.

Parameters:

• data – the raw data.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the advanced datatype.

Raises:

ConstraintException – if the passed data was not valid.

validate(data: Any, path: Tuple[str] = ())

This function validates an already unserialized data type and raises an exception if it does not match the type definition.

Parameters:

• data – the unserialized data.

• path – the path that lead to this validation call, in order to produce a nice error message

Raises:

ConstraintException – if the passed data was not valid.

exception arcaflow_plugin_sdk.schema.BadArgumentException(msg: str)

Bases: Exception

BadArgumentException indicates that an invalid configuration was passed to a schema component. The message will explain what exactly the problem is, but may not be able to locate the exact error as the schema may be manually built.

msg: str

class arcaflow_plugin_sdk.schema.BoolSchema

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema information for boolean types. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use BoolType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.FloatSchema(
...     min=0,
...     max=100,
...     units=schema.UNIT_PERCENT
... )

class arcaflow_plugin_sdk.schema.BoolType

Bases: BoolSchema, AbstractType

This type represents a boolean value with a multitude of unserialization options.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create bool type:

>>> bool_type = schema.BoolType()

Now you can use the type to unseralize, validate, or serialize values.

serialize(data: TypeT, path: Tuple[str] = ()) → Any

This function serializes a bool value.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create bool type:

>>> bool_type = schema.BoolType()

Validate:

>>> bool_type.serialize(True)
True
>>> bool_type.serialize(False)
False

This will throw an error:

>>> bool_type.serialize(1)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Boolean value expected, <class 'int'> found

unserialize(data: Any, path: Tuple[str] = ()) → TypeT

This function unserializes a bool value from a variety of types.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create bool type:

>>> bool_type = schema.BoolType()

Unserialize a bool iun various ways:

>>> bool_type.unserialize(True)
True
>>> bool_type.unserialize(1)
True
>>> bool_type.unserialize("true")
True
>>> bool_type.unserialize("yes")
True
>>> bool_type.unserialize("y")
True
>>> bool_type.unserialize("on")
True
>>> bool_type.unserialize("enable")
True
>>> bool_type.unserialize("enabled")
True
>>> bool_type.unserialize(False)
False
>>> bool_type.unserialize(0)
False
>>> bool_type.unserialize("false")
False
>>> bool_type.unserialize("no")
False
>>> bool_type.unserialize("n")
False
>>> bool_type.unserialize("off")
False
>>> bool_type.unserialize("disable")
False
>>> bool_type.unserialize("disabled")
False

This will throw an error:

>>> bool_type.unserialize("")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Boolean value expected, string found ()

validate(data: TypeT, path: Tuple[str] = ())

This function validates a bool value.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create bool type:

>>> bool_type = schema.BoolType()

Validate:

>>> bool_type.validate(True)
>>> bool_type.validate(False)

This will throw an error:

>>> bool_type.validate(1)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Boolean value expected, <class 'int'> found

exception arcaflow_plugin_sdk.schema.ConstraintException(path: Tuple[str] = (), msg: str = '')

Bases: Exception

ConstraintException indicates that the passed data violated one or more constraints defined in the schema. The message holds the exact path of the problematic field, as well as a message explaining the error. If this error is not easily understood, please open an issue on the Arcaflow plugin SDK.

msg: str = ''

path: Tuple[str] = ()

class arcaflow_plugin_sdk.schema.DisplayValue(name: str | None = None, description: str | None = None, icon: str | None = None)

Bases: object

This class holds the fields related to displaying an item in a user interface.

Example:

>>> d = DisplayValue(
...     name="Foo",
...     description="This is a foo",
...     icon="<svg></svg>"
... )

description: str | None = None

icon: str | None = None

name: str | None = None

class arcaflow_plugin_sdk.schema.FloatSchema(min: float | None = None, max: float | None = None, units: Units | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema information for 64-bit floating point numbers. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use FloatType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.FloatSchema(
...     min=0,
...     max=100,
...     units=schema.UNIT_PERCENT
... )

max: float | None = None

min: float | None = None

units: Units | None = None

class arcaflow_plugin_sdk.schema.FloatType(min: float | None = None, max: float | None = None, units: Units | None = None)

Bases: FloatSchema, AbstractType

This type represents a 64-bit floating point / real number.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize float type:

>>> float_type = schema.FloatType(
...     min=3.0,
...     max=5.0,
...     units=schema.UNIT_TIME,
... )

Now you can use this type to unserialize, validate, or serialize.

serialize(data: float, path: Tuple[str] = ()) → Any

This function will return a float for the base unit of this value.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize float type:

>>> float_type = schema.FloatType(
...     min=1000000.0,
...     max=5000000.0,
...     units=schema.UNIT_TIME,
... )

This will work:

>>> float_type.serialize(3000000.0)
3000000.0

These will fail:

>>> float_type.serialize("4s")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be an float, str given
>>> float_type.serialize(6000000.0)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at most 5s (5000000.0ns)
>>> float_type.serialize(500000.0)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000.0ns)

unserialize(data: Any, path: Tuple[str] = ()) → float

This function can unserialize a number for a integers or strings. If the passed data is a string, it can take the unit of the current type into account.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize float type:

>>> float_type = schema.FloatType(
...     min=1000000.0,
...     max=5000000.0,
...     units=schema.UNIT_TIME,
... )

Unserialize:

>>> float_type.unserialize("2s30ms")
2030000.0
>>> float_type.unserialize("2s30ms1.1ns")
2030001.1
>>> float_type.unserialize(3000000)
3000000.0
>>> float_type.unserialize(3000000.0)
3000000.0

These will fail:

>>> float_type.unserialize("4k")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Cannot parse '4k' as 'nanoseconds': invalid format, valid unit types are: 'nanoseconds', 'nanosecond', 'ns', 'microseconds', 'microsecond', 'ms', 'seconds', 'second', 's', 'minutes', 'minute', 'm', 'hours', 'hour', 'H', 'days', 'day', 'd'
>>> float_type.unserialize("6s")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at most 5s (5000000.0ns)
>>> float_type.unserialize("500ms")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000.0ns)
>>> float_type.unserialize(500000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000.0ns)

validate(data: float, path: Tuple[str] = ())

This function validates the passed number for conformity with the schema.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize float type:

>>> float_type = schema.FloatType(
...     min=1000000.0,
...     max=5000000.0,
...     units=schema.UNIT_TIME,
... )

This will work:

>>> float_type.validate(3000000.0)

These will fail:

>>> float_type.validate(3000000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be an float, int given
>>> float_type.validate("4s")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be an float, str given
>>> float_type.validate(6000000.0)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at most 5s (5000000.0ns)
>>> float_type.validate(500000.0)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000.0ns)

class arcaflow_plugin_sdk.schema.IntEnumSchema(values: Dict[int, DisplayValue], units: Units | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class specifically holds an enum that has integer values.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use IntEnumType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.IntEnumSchema({
...     1: schema.DisplayValue("Apple"),
...     2: schema.DisplayValue("Banana"),
...     3: schema.DisplayValue("Orange"),
... })
>>> s.valid_values()
[1, 2, 3]

units: Units | None = None

valid_values() → List[int]

values: Dict[int, DisplayValue]

class arcaflow_plugin_sdk.schema.IntEnumType(t: Type[EnumT])

Bases: _EnumType, IntEnumSchema

This class represents an enum type that is an integer.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from enum import Enum

Create an enum:

>>> class PrimeNumbers(Enum):
...    FIRST=2
...    SECOND=3
...    THIRD=5

Create the type:

>>> prime_numbers_type = schema.IntEnumType(PrimeNumbers)

Unserialize a value:

>>> prime_numbers_type.unserialize(2)
<PrimeNumbers.FIRST: 2>

Serialize a value:

>>> prime_numbers_type.serialize(PrimeNumbers.FIRST)
2

Validate a value:

>>> prime_numbers_type.validate(4)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: '4' is not a valid value for 'PrimeNumbers'

class arcaflow_plugin_sdk.schema.IntSchema(min: int | None = None, max: int | None = None, units: Units | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema information for 64-bit integers. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use IntType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.IntSchema(
...     min=3,
...     max=5,
...     units=schema.UNIT_BYTE
... )

max: int | None = None

min: int | None = None

units: Units | None = None

class arcaflow_plugin_sdk.schema.IntType(min: int | None = None, max: int | None = None, units: Units | None = None)

Bases: IntSchema, AbstractType

IntType represents an integer type, both positive or negative. It is designed to take a 64 bit value.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize int type:

>>> int_type = schema.IntType(
...     min=3,
...     max=5,
...     units=schema.UNIT_TIME,
... )

Now you can use this type to unserialize, validate, or serialize.

serialize(data: int, path: Tuple[str] = ()) → Any

This function will return an integer for the base unit of this value.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize int type:

>>> int_type = schema.IntType(
...     min=1000000,
...     max=5000000,
...     units=schema.UNIT_TIME,
... )

This will work:

>>> int_type.serialize(3000000)
3000000

These will fail:

>>> int_type.serialize("4s")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be an integer, str given
>>> int_type.serialize(6000000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at most 5s (5000000ns)
>>> int_type.serialize(500000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000ns)

unserialize(data: Any, path: Tuple[str] = ()) → int

This function can unserialize a number for a integers or strings. If the passed data is a string, it can take the unit of the current type into account.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize int type:

>>> int_type = schema.IntType(
...     min=1000000,
...     max=5000000,
...     units=schema.UNIT_TIME,
... )

Unserialize:

>>> int_type.unserialize("2s30ms")
2030000
>>> int_type.unserialize(3000000)
3000000

These will fail:

>>> int_type.unserialize("4k")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Cannot parse '4k' as 'nanoseconds': invalid format, valid unit types are: 'nanoseconds', 'nanosecond', 'ns', 'microseconds', 'microsecond', 'ms', 'seconds', 'second', 's', 'minutes', 'minute', 'm', 'hours', 'hour', 'H', 'days', 'day', 'd'
>>> int_type.unserialize("6s")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at most 5s (5000000ns)
>>> int_type.unserialize("500ms")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000ns)
>>> int_type.unserialize(500000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000ns)

validate(data: int, path: Tuple[str] = ())

This function validates the passed number for conformity with the schema.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Initialize int type:

>>> int_type = schema.IntType(
...     min=1000000,
...     max=5000000,
...     units=schema.UNIT_TIME,
... )

This will work:

>>> int_type.validate(3000000)

These will fail:

>>> int_type.validate("4s")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be an integer, str given
>>> int_type.validate(6000000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at most 5s (5000000ns)
>>> int_type.validate(500000)
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must be at least 1s (1000000ns)

exception arcaflow_plugin_sdk.schema.InvalidAnnotationException(annotation: str, msg: str)

Bases: Exception

InvalidAnnotationException indicates that an annotation was used on a type it does not support.

annotation: str

msg: str

exception arcaflow_plugin_sdk.schema.InvalidInputException(cause: ConstraintException)

Bases: Exception

This exception indicates that the input data for a given step didn’t match the schema. The embedded ConstraintException holds the details of this failure.

constraint: ConstraintException

exception arcaflow_plugin_sdk.schema.InvalidOutputException(cause: ConstraintException)

Bases: Exception

This exception indicates that the output of a schema was invalid. This is always a bug in the plugin and should be reported to the plugin author.

constraint: ConstraintException

class arcaflow_plugin_sdk.schema.ListSchema(items: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema, min: int | None = None, max: int | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema definition for lists. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use ListType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.ListSchema(
...     items=schema.StringSchema(),
...     min=2,
...     max=3,
... )

items: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema

max: int | None = None

min: int | None = None

class arcaflow_plugin_sdk.schema.ListType(items: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema, min: int | None = None, max: int | None = None)

Bases: ListSchema, AbstractType, Generic[ListT]

ListType is a strongly typed list that can have elements of only one type. The typical Python equivalent would be typing.List[sometype].

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a list type:

>>> list_type = schema.ListType(
...     schema.StringType(),
...     min=1,
...     max=5,
... )

Now you can use the list type to unserialize, validate, and serialize.

items: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema

serialize(data: ListT, path: Tuple[str] = ()) → Any

This function serializes the list elements into a list for transport.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a list type:

>>> list_type = schema.ListType(
...     schema.StringType(min=1),
...     min=1,
...     max=3,
... )

Serialize the data:

>>> list_type.serialize(["a"])
['a']

unserialize(data: Any, path: Tuple[str] = ()) → ListT

This function unserializes the list itself, and also unserializes the underlying type.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a list type:

>>> list_type = schema.ListType(
...     schema.StringType(min=1),
...     min=1,
...     max=3,
... )

Unserialize data:

>>> list_type.unserialize(["Hello world!"])
['Hello world!']

These will fail:

>>> list_type.unserialize([])
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at least 1 items, 0 given
>>> list_type.unserialize(["a","b","c","d"])
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at most 3 items, 4 given

Underlying types are also validated:

>>> list_type.unserialize([""])
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'item 0': String must be at least 1 characters, 0 given

validate(data: TypeT, path: Tuple[str] = ())

This function validates the data type. It also validates the underlying data type.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a list type:

>>> list_type = schema.ListType(
...     schema.StringType(min=1),
...     min=1,
...     max=3,
... )

Validate the data:

>>> list_type.validate(["Hello world!"])

These will fail:

>>> list_type.validate([])
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at least 1 items, 0 given
>>> list_type.validate(["a","b","c","d"])
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at most 3 items, 4 given

Underlying types are also validated:

>>> list_type.validate([""])
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'item 0': String must be at least 1 characters, 0 given

class arcaflow_plugin_sdk.schema.MapSchema(keys: StringEnumSchema | IntEnumSchema | StringSchema | IntSchema, values: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema, min: int | None = None, max: int | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema definition for key-value associations. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use MapType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.MapSchema(
...     keys=schema.StringSchema(),
...     values=schema.IntSchema(),
...     min=2,
...     max=3,
... )

keys: StringEnumSchema | IntEnumSchema | StringSchema | IntSchema

max: int | None = None

min: int | None = None

values: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema

class arcaflow_plugin_sdk.schema.MapType(keys: StringEnumSchema | IntEnumSchema | StringSchema | IntSchema, values: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema, min: int | None = None, max: int | None = None)

Bases: MapSchema, AbstractType, Generic[MapT]

MapType is a key-value dict with fixed types for both.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a map type:

>>> map_type = schema.MapType(
...     keys=schema.StringType(min=1),
...     values=schema.IntType(),
...     min=1,
...     max=2,
... )

Now you can use the map type to unserialize, validate, or serialize data.

keys: StringEnumSchema | IntEnumSchema | StringSchema | IntSchema

serialize(data: MapT, path: Tuple[str] = ()) → Any

This function serializes the data into the transportable system.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a map type:

>>> map_type = schema.MapType(
...     keys=schema.StringType(min=2),
...     values=schema.IntType(min=3),
...     min=1,
...     max=2,
... )

This is valid:

>>> map_type.serialize({"foo": 5})
{'foo': 5}

This will not work due to underlying types failing validation:

>>> map_type.serialize({"a": 5})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'a -> key': String must be at least 2 characters, 1 given
>>> map_type.serialize({"foo":1})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'foo -> value': Must be at least 3

This will also fail because the map does not have enough elements:

>>> map_type.serialize({})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at least 1 elements, 0 given.

unserialize(data: Any, path: Tuple[str] = ()) → MapT

Unserialize a map (dict) type as defined with the underlying types.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a map type:

>>> map_type = schema.MapType(
...     keys=schema.StringType(min=2),
...     values=schema.IntType(min=3),
...     min=1,
...     max=2,
... )

Unserialize data:

>>> map_type.unserialize({"foo": 5})
{'foo': 5}

This will not work due to underlying types failing validation:

>>> map_type.unserialize({"a": 5})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'a -> key': String must be at least 2 characters, 1 given
>>> map_type.unserialize({"foo":1})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'foo -> value': Must be at least 3

This will also fail because the map does not have enough elements:

>>> map_type.unserialize({})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at least 1 elements, 0 given.

validate(data: TypeT, path: Tuple[str] = ())

This function validates the map and its underlying types.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a map type:

>>> map_type = schema.MapType(
...     keys=schema.StringType(min=2),
...     values=schema.IntType(min=3),
...     min=1,
...     max=2,
... )

This is valid:

>>> map_type.validate({"foo": 5})

This will not work due to underlying types failing validation:

>>> map_type.validate({"a": 5})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'a -> key': String must be at least 2 characters, 1 given
>>> map_type.validate({"foo":1})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'foo -> value': Must be at least 3

This will also fail because the map does not have enough elements:

>>> map_type.validate({})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Must have at least 1 elements, 0 given.

values: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema

exception arcaflow_plugin_sdk.schema.NoSuchSignalException(step: str, signal: str)

Bases: Exception

NoSuchSignalException indicates that the given signal is not supported by the plugin’s step.

signal: str

step: str

exception arcaflow_plugin_sdk.schema.NoSuchStepException(step: str)

Bases: Exception

NoSuchStepException indicates that the given step is not supported by the plugin.

step: str

class arcaflow_plugin_sdk.schema.ObjectSchema(id: str, properties: Dict[str, PropertySchema])

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the definition for objects comprised of defined fields. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use PropertyType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.ObjectSchema(
...     "MyObject",
...     {
...         "foo": schema.PropertySchema(
...             type=schema.StringSchema(),
...             # Set the display settings for the property.
...             display=schema.DisplayValue(
...                 name="Foo",
...             ),
...             # Set the default value in JSON-encoded format.
...             default="'bar'",
...             # Add examples in JSON-encoded format.
...             examples=['baz'],
...             # Mark the field as optional.
...             required=False,
...             # ...
...         )
...     }
... )

id: str

properties: Dict[str, PropertySchema]

class arcaflow_plugin_sdk.schema.ObjectType(cls: Type[ObjectT], properties: Dict[str, PropertyType])

Bases: ObjectSchema, AbstractType, Generic[ObjectT]

ObjectType represents an object with predefined fields. The property declaration must match the fields in the class. The type currently does not validate if the properties match the provided class.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass

Define your dataclass

>>> @dataclass
... class ExampleData:
...    a: str

Create a schema:

>>> object_type = schema.build_object_schema(ExampleData)

This will result in a scope object, containing the dataclass with the property as a root object:

>>> object_type.properties["a"].type
StringType(min=None, max=None, pattern=None)

Alternatively, you can construct the type by hand:

>>> object_type = schema.ObjectType(
...     ExampleData,
...     {
...         "a": schema.PropertyType(schema.StringType())
...     }
... )

Now you can query the type as before:

>>> object_type.properties["a"].type
StringType(min=None, max=None, pattern=None)

You can now use the object_type to unserialize, validate, and serialize properties.

property cls: Type[ObjectT]

properties: Dict[str, PropertyType]

serialize(data: ObjectT, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

unserialize(data: Any, path: Tuple[str] = ()) → ObjectT

This function unserializes a dict into a dataclass.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass

Define a dataclass:

>>> @dataclass
... class TestData:
...     a: str

Build a schema:

>>> object_type = schema.build_object_schema(TestData)

Unserialize data:

>>> object_type.unserialize({"a":"Hello world!"})
TestData(a='Hello world!')

This will fail:

>>> object_type.unserialize("Hello world!")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'TestData': Must be a dict, got str

validate(data: TypeT, path: Tuple[str] = ())

This function will validate the dataclass and all underlying types as well.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass

Define a dataclass:

>>> @dataclass
... class TestData:
...     a: typing.Annotated[str, schema.min(1)]

Build a schema:

>>> object_type = schema.build_object_schema(TestData)

Validate a data class:

>>> object_type.validate(TestData("Hello world!"))

These will fail:

>>> object_type.validate(TestData(""))
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'TestData -> a': String must be at least 1 characters, 0 given

class arcaflow_plugin_sdk.schema.OneOfIntSchema(types: Dict[int, RefSchema | ScopeSchema | ObjectSchema], discriminator_field_name: str = '_type')

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the definition of variable types with an integer discriminator. This type acts as a split for a case where multiple possible object types can be present in a field. This type requires that there be a common field (the discriminator) which tells a parsing party which type it is. The field type in this case is a string.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use OneOfIntType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> a = schema.ObjectSchema(
...     "A",
...     {
...         "a": schema.PropertySchema(
...             type=schema.StringSchema(),
...         )
...     }
... )
>>> b = schema.ObjectSchema(
...     "B",
...     {
...         "b": schema.PropertySchema(
...             type=schema.IntSchema(),
...         )
...     }
... )
>>> one_of = schema.OneOfIntSchema(
...     {
...         1: schema.RefSchema("A_ref"),
...         2: schema.RefSchema("B_ref")
...     }
... )
>>> c = schema.ObjectSchema(
...     "C",
...     {
...         "o": schema.PropertySchema(
...             one_of,
...         )
...     }
... )
>>> s = schema.ScopeSchema(
...     {
...         "A_ref": a,
...         "B_ref": b,
...         "C_ref": c,
...     },
...     "C",
... )

discriminator_field_name: str = '_type'

types: Dict[int, RefSchema | ScopeSchema | ObjectSchema]

class arcaflow_plugin_sdk.schema.OneOfIntType(types: Dict[int, RefSchema | ScopeSchema | ObjectSchema], scope: ScopeType, discriminator_field_name: str = '_type')

Bases: OneOfIntSchema, _OneOfType[OneOfT, int], Generic[OneOfT]

types: Dict[int, RefSchema | ScopeSchema | ObjectSchema]

class arcaflow_plugin_sdk.schema.OneOfStringSchema(types: Dict[str, RefSchema | ScopeSchema | ObjectSchema], discriminator_field_name: str = '_type')

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the definition of variable types with a string discriminator. This type acts as a split for a case where multiple possible object types can be present in a field. This type requires that there be a common field (the discriminator) which tells a parsing party which type it is. The field type in this case is a string.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use OneOfStringType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> a = schema.ObjectSchema(
...     "A",
...     {
...         "a": schema.PropertySchema(
...             type=schema.StringSchema(),
...         )
...     }
... )
>>> b = schema.ObjectSchema(
...     "B",
...     {
...         "b": schema.PropertySchema(
...             type=schema.IntSchema(),
...         )
...     }
... )
>>> one_of = schema.OneOfStringSchema(
...     {
...         "a": schema.RefSchema("A_ref"),
...         "b": schema.RefSchema("B_ref")
...     }
... )
>>> c = schema.ObjectSchema(
...     "C",
...     {
...         "o": schema.PropertySchema(
...             one_of,
...         )
...     }
... )
>>> s = schema.ScopeSchema(
...     {
...         "A_ref": a,
...         "B_ref": b,
...         "C_ref": c,
...     },
...     "C",
... )

Instead of RefSchema, you can also add other object-like types, such as the ObjectSchema or the ScopeSchema directly: >>> one_of = schema.OneOfStringSchema( … { … “a”: schema.RefSchema(“A_ref”), … “b”: schema.ObjectSchema( … “B”, … { … “b”: schema.PropertySchema( … type=schema.IntSchema(), … ) … } … ) … } … )

discriminator_field_name: str = '_type'

types: Dict[str, RefSchema | ScopeSchema | ObjectSchema]

class arcaflow_plugin_sdk.schema.OneOfStringType(types: Dict[str, RefSchema | ScopeSchema | ObjectSchema], scope: ScopeType, discriminator_field_name: str = '_type')

Bases: OneOfStringSchema, _OneOfType[OneOfT, str], Generic[OneOfT]

types: Dict[str, RefSchema | ScopeSchema | ObjectSchema]

class arcaflow_plugin_sdk.schema.PatternSchema

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema information for regular expression patterns. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use PatternType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.PatternSchema()

class arcaflow_plugin_sdk.schema.PatternType

Bases: PatternSchema, AbstractType

PatternType represents a regular expression.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> pattern_type = schema.PatternType()

You can now unserialize, validate, or serialize the data.

serialize(data: Pattern, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

unserialize(data: Any, path: Tuple[str] = ()) → Pattern

This function unserializes a regular expression from a string.

Example:

Initialize:

>>> from arcaflow_plugin_sdk import schema
>>> pattern_type = schema.PatternType()

Unserialize:

>>> regexp = pattern_type.unserialize("^[a-z]+$")
>>> regexp.pattern
'^[a-z]+$'

This will throw an error:

>>> regexp = pattern_type.unserialize("[")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Invalid regular expression (unterminated character set at position 0)

validate(data: Pattern, path: Tuple[str] = ())

This function validates a regular expression as such.

Example:

Initialize:

>>> from arcaflow_plugin_sdk import schema
>>> from re import compile
>>> pattern_type = schema.PatternType()

Validate:

>>> pattern_type.validate(compile("^[a-z]+$"))

This will fail:

>>> pattern_type.validate("^[a-z]+$")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: Not a regular expression

class arcaflow_plugin_sdk.schema.PropertySchema(type: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema, display: DisplayValue | None = None, default: str | None = None, examples: List[str] | None = None, required: bool = True, required_if: List[str] | None = None, required_if_not: List[str] | None = None, conflicts: List[str] | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the schema definition for a single object property. It is usable in conjunction with ObjectSchema. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use PropertyType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.ObjectSchema(
...     "MyObject",
...     {
...         "foo": schema.PropertySchema(
...             type=schema.StringSchema(),
...             # Set the display settings for the property.
...             display=schema.DisplayValue(
...                 name="Foo",
...             ),
...             # Set the default value in JSON-encoded format.
...             default="'bar'",
...             # Add examples in JSON-encoded format.
...             examples=['baz'],
...             # Mark the field as optional.
...             required=False,
...             # ...
...         )
...     }
... )

conflicts: List[str] | None = None

default: str | None = None

display: DisplayValue | None = None

examples: List[str] = None

required: bool = True

required_if: List[str] | None = None

required_if_not: List[str] | None = None

type: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema

class arcaflow_plugin_sdk.schema.PropertyType(type: StringEnumSchema | IntEnumSchema | StringSchema | PatternSchema | IntSchema | FloatSchema | BoolSchema | ListSchema | MapSchema | ScopeSchema | ObjectSchema | OneOfStringSchema | OneOfIntSchema | RefSchema | AnySchema, display: DisplayValue | None = None, default: str | None = None, examples: List[str] | None = None, required: bool | None = True, required_if: List[str] | None = None, required_if_not: List[str] | None = None, conflicts: List[str] | None = None, field_override: str = '')

Bases: PropertySchema, Generic[PropertyT]

This class holds the schema definition for a single object property . It is usable in conjunction with ObjectType.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass

Define your dataclass

>>> @dataclass
... class ExampleData:
...    a: str

Create a schema:

>>> object_type = schema.build_object_schema(ExampleData)

This will result in a scope object, containing the dataclass with the property as a root object:

>>> object_type.properties["a"].type
StringType(min=None, max=None, pattern=None)

Alternatively, you can construct the type by hand:

>>> object_type = schema.ObjectType(
...     ExampleData,
...     {
...         "a": schema.PropertyType(schema.StringType())
...     }
... )

Now you can query the type as before:

>>> object_type.properties["a"].type
StringType(min=None, max=None, pattern=None)

field_override: str = ''

class arcaflow_plugin_sdk.schema.RefSchema(id: str, display: DisplayValue | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the definition of a reference to a scope-wide object. The ref must always be inside a scope, either directly or indirectly. If several scopes are embedded within each other, the Ref references the object in the current scope.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use RefType.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a scope with an object:

>>> s = schema.ScopeSchema(
...     {
...         "a": schema.ObjectSchema("A", {})
...     },
...     "a"
... )

Create a ref:

>>> ref = schema.RefSchema("a")

Use ref in an object:

>>> s.objects["b"] = schema.ObjectSchema(
...     "B",
...    {
...        "a": schema.PropertySchema(ref)
...    }
... )

display: DisplayValue | None = None

id: str

class arcaflow_plugin_sdk.schema.RefType(id: str, scope: ScopeType)

Bases: RefSchema, AbstractType

A ref is a reference to an object in a Scope.

property properties

serialize(data: TypeT, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

unserialize(data: Any, path: Tuple[str] = ()) → TypeT

This function takes the underlying raw data and decodes it into the underlying advanced data type (e.g. dataclass) for usage.

Parameters:

• data – the raw data.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the advanced datatype.

Raises:

ConstraintException – if the passed data was not valid.

validate(data: TypeT, path: Tuple[str] = ())

This function validates an already unserialized data type and raises an exception if it does not match the type definition.

Parameters:

• data – the unserialized data.

• path – the path that lead to this validation call, in order to produce a nice error message

Raises:

ConstraintException – if the passed data was not valid.

arcaflow_plugin_sdk.schema.SCHEMA_SCHEMA = ScopeType(objects={'Schema': ObjectType(id='Schema', properties={'steps': PropertyType(type=MapType(keys=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), values=RefType(id='StepSchema', display=None), min=None, max=None), display=DisplayValue(name='Steps', description='Steps this schema supports.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.Schema'>), 'StepSchema': ObjectType(id='StepSchema', properties={'id': PropertyType(type=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), display=DisplayValue(name='ID', description='Machine identifier for this step.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'input': PropertyType(type=RefType(id='ScopeSchema', display=None), display=DisplayValue(name='Input', description='Input data schema', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'outputs': PropertyType(type=MapType(keys=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), values=RefType(id='StepOutputSchema', display=None), min=None, max=None), display=DisplayValue(name='Outputs', description='Possible outputs from this step.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'signal_handlers': PropertyType(type=MapType(keys=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), values=RefType(id='SignalSchema', display=None), min=None, max=None), display=DisplayValue(name='Signal handlers', description='Signals that are input by the step.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'signal_emitters': PropertyType(type=MapType(keys=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), values=RefType(id='SignalSchema', display=None), min=None, max=None), display=DisplayValue(name='Signal emitters', description='Signals that are output by the step.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.StepSchema'>), 'ScopeSchema': ObjectType(id='ScopeSchema', properties={'objects': PropertyType(type=MapType(keys=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), values=RefType(id='ObjectSchema', display=None), min=None, max=None), display=DisplayValue(name='Objects', description='A set of referenceable objects. These objects may contain references themselves.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'root': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Root object', description='Reference to the root object of this scope', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.ScopeSchema'>), 'ObjectSchema': ObjectType(id='ObjectSchema', properties={'id': PropertyType(type=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), display=DisplayValue(name='ID', description='Unique identifier for this object within the current scope.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'properties': PropertyType(type=MapType(keys=StringType(min=None, max=None, pattern=None), values=RefType(id='PropertySchema', display=None), min=None, max=None), display=DisplayValue(name='Properties', description='Properties of this object.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.ObjectSchema'>), 'PropertySchema': ObjectType(id='PropertySchema', properties={'type': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'pattern': RefType(id='PatternSchema', display=DisplayValue(name='Pattern', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None)), 'float': RefType(id='FloatSchema', display=DisplayValue(name='Float', description=None, icon=None)), 'bool': RefType(id='BoolSchema', display=DisplayValue(name='Bool', description=None, icon=None)), 'list': RefType(id='ListSchema', display=DisplayValue(name='List', description=None, icon=None)), 'map': RefType(id='MapSchema', display=DisplayValue(name='Map', description=None, icon=None)), 'scope': RefType(id='ScopeSchema', display=DisplayValue(name='Scope', description=None, icon=None)), 'object': RefType(id='ObjectSchema', display=DisplayValue(name='Object', description=None, icon=None)), 'one_of_string': RefType(id='OneOfStringSchema', display=DisplayValue(name='Multiple with string key', description=None, icon=None)), 'one_of_int': RefType(id='OneOfIntSchema', display=DisplayValue(name='Multiple with int key', description=None, icon=None)), 'ref': RefType(id='RefSchema', display=DisplayValue(name='Object reference', description=None, icon=None)), 'any': RefType(id='AnySchema', display=DisplayValue(name='Any', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Type', description='Type definition for this field.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'default': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Default', description='Default value for this property in JSON encoding. The value must be unserializable by the type specified in the type field. ', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'examples': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Examples', description='Example values for this property, encoded as JSON.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'required': PropertyType(type=BoolType(), display=DisplayValue(name='Required', description='When set to true, the value for this field must be provided under all circumstances.', icon=None), default='true', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'required_if': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Required if', description='Sets the current property to required if any of the properties in this list are set.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'required_if_not': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Required if not', description='Sets the current property to be required if none of the properties in this list are set.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'conflicts': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Conflicts', description='The current property cannot be set if any of the listed properties are set.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.PropertySchema'>), 'StringEnumSchema': ObjectType(id='StringEnumSchema', properties={'values': PropertyType(type=MapType(keys=StringType(min=None, max=None, pattern=None), values=RefType(id='DisplayValue', display=None), min=1, max=None), display=DisplayValue(name='Values', description='Mapping where the left side of the map holds the possible value and the right side holds the display value for forms, etc.', icon=None), default=None, examples=['{"apple": {"name": "Apple"}, "orange": {"name": "Orange"}}'], required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.StringEnumSchema'>), 'DisplayValue': ObjectType(id='DisplayValue', properties={'name': PropertyType(type=StringType(min=1, max=None, pattern=None), display=DisplayValue(name='Name', description='Short text serving as a name or title for this item.', icon=None), default=None, examples=['"Fruit"'], required=False, required_if=None, required_if_not=None, conflicts=None), 'description': PropertyType(type=StringType(min=1, max=None, pattern=None), display=DisplayValue(name='Description', description='Description for this item if needed.', icon=None), default=None, examples=['"Please select the fruit you would like."'], required=False, required_if=None, required_if_not=None, conflicts=None), 'icon': PropertyType(type=StringType(min=1, max=None, pattern=None), display=DisplayValue(name='Icon', description='SVG icon for this item. Must have the declared size of 64x64, must not include additional namespaces, and must not reference external resources.', icon=None), default=None, examples=['"<svg ...></svg>"'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.DisplayValue'>), 'IntEnumSchema': ObjectType(id='IntEnumSchema', properties={'values': PropertyType(type=MapType(keys=IntType(min=None, max=None, units=None), values=RefType(id='DisplayValue', display=None), min=1, max=None), display=DisplayValue(name='Values', description='Possible values for this field.', icon=None), default=None, examples=['{"1024": {"name": "kB"}, "1048576": {"name": "MB"}}'], required=True, required_if=None, required_if_not=None, conflicts=None), 'units': PropertyType(type=RefType(id='Units', display=None), display=DisplayValue(name=None, description=None, icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.IntEnumSchema'>), 'Units': ObjectType(id='Units', properties={'base_unit': PropertyType(type=RefType(id='Unit', display=None), display=DisplayValue(name='Base unit', description='The base unit is the smallest unit of scale for this set of units.', icon=None), default=None, examples=['{"name_short_singular": "B", "name_short_plural": "B", "name_long_singular": "byte", "name_long_plural": "bytes"}'], required=True, required_if=None, required_if_not=None, conflicts=None), 'multipliers': PropertyType(type=MapType(keys=IntType(min=None, max=None, units=None), values=RefType(id='Unit', display=None), min=None, max=None), display=DisplayValue(name='Multipliers', description='A set of multiplies that describe multiple units of scale.', icon=None), default=None, examples=['{"1024": {"name_short_singular": "kB", "name_short_plural": "kB", "name_long_singular": "kilobyte", "name_long_plural": "kilobytes"}, "1048576": {"name_short_singular": "MB", "name_short_plural": "MB", "name_long_singular": "megabyte", "name_long_plural": "megabytes"}}'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.Units'>), 'Unit': ObjectType(id='Unit', properties={'name_short_singular': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Short name (singular)', description='Short name that can be printed in a few characters, singular form.', icon=None), default=None, examples=['"B"', '"char"'], required=True, required_if=None, required_if_not=None, conflicts=None), 'name_short_plural': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Short name (plural)', description='Short name that can be printed in a few characters, plural form.', icon=None), default=None, examples=['"B"', '"chars"'], required=True, required_if=None, required_if_not=None, conflicts=None), 'name_long_singular': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Long name (singular)', description='Longer name for this unit in singular form.', icon=None), default=None, examples=['"byte"', '"character"'], required=True, required_if=None, required_if_not=None, conflicts=None), 'name_long_plural': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Long name (plural)', description='Longer name for this unit in plural form.', icon=None), default=None, examples=['"bytes"', '"characters"'], required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.Unit'>), 'StringSchema': ObjectType(id='StringSchema', properties={'min': PropertyType(type=IntType(min=0, max=None, units=Units(base_unit=Unit(name_short_singular='char', name_short_plural='chars', name_long_singular='character', name_long_plural='characters'), multipliers=None)), display=DisplayValue(name='Minimum length', description='Minimum length for this string (inclusive).', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=0, max=None, units=Units(base_unit=Unit(name_short_singular='char', name_short_plural='chars', name_long_singular='character', name_long_plural='characters'), multipliers=None)), display=DisplayValue(name='Maximum length', description='Maximum length for this string (inclusive).', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None), 'pattern': PropertyType(type=PatternType(), display=DisplayValue(name='Pattern', description='Regular expression this string must match.', icon=None), default=None, examples=['"^[a-zA-Z]+$"'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.StringSchema'>), 'PatternSchema': ObjectType(id='PatternSchema', properties={}, _cls=<class 'arcaflow_plugin_sdk.schema.PatternSchema'>), 'IntSchema': ObjectType(id='IntSchema', properties={'min': PropertyType(type=IntType(min=None, max=None, units=None), display=DisplayValue(name='Minimum value', description='Minimum value for this int (inclusive).', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=None, max=None, units=None), display=DisplayValue(name='Maximum value', description='Maximum value for this int (inclusive).', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None), 'units': PropertyType(type=RefType(id='Units', display=None), display=DisplayValue(name='Units', description='Units this number represents.', icon=None), default=None, examples=['{"base_unit": {"name_short_singular": "char", "name_short_plural": "chars", "name_long_singular": "character", "name_long_plural": "characters"}}'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.IntSchema'>), 'FloatSchema': ObjectType(id='FloatSchema', properties={'min': PropertyType(type=FloatType(min=None, max=None, units=None), display=DisplayValue(name='Minimum value', description='Minimum value for this float (inclusive).', icon=None), default=None, examples=['5.0'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=FloatType(min=None, max=None, units=None), display=DisplayValue(name='Maximum value', description='Maximum value for this float (inclusive).', icon=None), default=None, examples=['16.0'], required=False, required_if=None, required_if_not=None, conflicts=None), 'units': PropertyType(type=RefType(id='Units', display=None), display=DisplayValue(name='Units', description='Units this number represents.', icon=None), default=None, examples=['{"base_unit": {"name_short_singular": "%", "name_short_plural": "%", "name_long_singular": "percent", "name_long_plural": "percent"}}'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.FloatSchema'>), 'BoolSchema': ObjectType(id='BoolSchema', properties={}, _cls=<class 'arcaflow_plugin_sdk.schema.BoolSchema'>), 'ListSchema': ObjectType(id='ListSchema', properties={'items': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'pattern': RefType(id='PatternSchema', display=DisplayValue(name='Pattern', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None)), 'float': RefType(id='FloatSchema', display=DisplayValue(name='Float', description=None, icon=None)), 'bool': RefType(id='BoolSchema', display=DisplayValue(name='Bool', description=None, icon=None)), 'list': RefType(id='ListSchema', display=DisplayValue(name='List', description=None, icon=None)), 'map': RefType(id='MapSchema', display=DisplayValue(name='Map', description=None, icon=None)), 'scope': RefType(id='ScopeSchema', display=DisplayValue(name='Scope', description=None, icon=None)), 'object': RefType(id='ObjectSchema', display=DisplayValue(name='Object', description=None, icon=None)), 'one_of_string': RefType(id='OneOfStringSchema', display=DisplayValue(name='Multiple with string key', description=None, icon=None)), 'one_of_int': RefType(id='OneOfIntSchema', display=DisplayValue(name='Multiple with int key', description=None, icon=None)), 'ref': RefType(id='RefSchema', display=DisplayValue(name='Object reference', description=None, icon=None)), 'any': RefType(id='AnySchema', display=DisplayValue(name='Any', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Items', description='Type definition for items in this list.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'min': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Minimum items', description='Minimum number of items in this list.', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Maximum items', description='Maximum number of items in this list.', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.ListSchema'>), 'MapSchema': ObjectType(id='MapSchema', properties={'keys': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Keys', description='Type definition for map keys.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'values': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'pattern': RefType(id='PatternSchema', display=DisplayValue(name='Pattern', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None)), 'float': RefType(id='FloatSchema', display=DisplayValue(name='Float', description=None, icon=None)), 'bool': RefType(id='BoolSchema', display=DisplayValue(name='Bool', description=None, icon=None)), 'list': RefType(id='ListSchema', display=DisplayValue(name='List', description=None, icon=None)), 'map': RefType(id='MapSchema', display=DisplayValue(name='Map', description=None, icon=None)), 'scope': RefType(id='ScopeSchema', display=DisplayValue(name='Scope', description=None, icon=None)), 'object': RefType(id='ObjectSchema', display=DisplayValue(name='Object', description=None, icon=None)), 'one_of_string': RefType(id='OneOfStringSchema', display=DisplayValue(name='Multiple with string key', description=None, icon=None)), 'one_of_int': RefType(id='OneOfIntSchema', display=DisplayValue(name='Multiple with int key', description=None, icon=None)), 'ref': RefType(id='RefSchema', display=DisplayValue(name='Object reference', description=None, icon=None)), 'any': RefType(id='AnySchema', display=DisplayValue(name='Any', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Values', description='Type definition for map values.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'min': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Minimum items', description='Minimum number of items in this list.', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Maximum items', description='Maximum number of items in this list.', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.MapSchema'>), 'OneOfStringSchema': ObjectType(id='OneOfStringSchema', properties={'types': PropertyType(type=MapType(keys=StringType(min=None, max=None, pattern=None), values=OneOfStringType(types={'ref': RefType(id='RefSchema', display=None), 'scope': RefType(id='ScopeSchema', display=None), 'object': RefType(id='ObjectSchema', display=None)}, discriminator_field_name='type_id'), min=None, max=None), display=DisplayValue(name=None, description=None, icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'discriminator_field_name': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Discriminator field name', description='Name of the field used to discriminate between possible values. If this field ispresent on any of the component objects it must also be a string.', icon=None), default='"_type"', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.OneOfStringSchema'>), 'RefSchema': ObjectType(id='RefSchema', properties={'id': PropertyType(type=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), display=DisplayValue(name='ID', description='Referenced object ID.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.RefSchema'>), 'OneOfIntSchema': ObjectType(id='OneOfIntSchema', properties={'types': PropertyType(type=MapType(keys=IntType(min=None, max=None, units=None), values=OneOfStringType(types={'ref': RefType(id='RefSchema', display=None), 'scope': RefType(id='ScopeSchema', display=None), 'object': RefType(id='ObjectSchema', display=None)}, discriminator_field_name='type_id'), min=None, max=None), display=DisplayValue(name=None, description=None, icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'discriminator_field_name': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Discriminator field name', description='Name of the field used to discriminate between possible values. If this field ispresent on any of the component objects it must also be an int.', icon=None), default='"_type"', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.OneOfIntSchema'>), 'AnySchema': ObjectType(id='AnySchema', properties={}, _cls=<class 'arcaflow_plugin_sdk.schema.AnySchema'>), 'StepOutputSchema': ObjectType(id='StepOutputSchema', properties={'schema': PropertyType(type=RefType(id='ScopeSchema', display=None), display=DisplayValue(name='Schema', description='Data schema for this particular output.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'error': PropertyType(type=BoolType(), display=DisplayValue(name='Error', description='If set to true, this output will be treated as an error output.', icon=None), default='false', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.StepOutputSchema'>), 'SignalSchema': ObjectType(id='SignalSchema', properties={'id': PropertyType(type=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), display=DisplayValue(name='ID', description='Machine identifier for this step.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'data_schema': PropertyType(type=RefType(id='ScopeSchema', display=None), display=DisplayValue(name='Data', description='Input or output data schema', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.SignalSchema'>)}, root='Schema')

This variable holds a constructed, serializable/unserializable schema for an entire schema. You can use it to send schemas to the engine, or to build an engine replacement. (This is normally handled by the plugin module.)

arcaflow_plugin_sdk.schema.SCOPE_SCHEMA = ScopeType(objects={'ScopeSchema': ObjectType(id='ScopeSchema', properties={'objects': PropertyType(type=MapType(keys=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), values=RefType(id='ObjectSchema', display=None), min=None, max=None), display=DisplayValue(name='Objects', description='A set of referenceable objects. These objects may contain references themselves.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'root': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Root object', description='Reference to the root object of this scope', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.ScopeSchema'>), 'ObjectSchema': ObjectType(id='ObjectSchema', properties={'id': PropertyType(type=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), display=DisplayValue(name='ID', description='Unique identifier for this object within the current scope.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'properties': PropertyType(type=MapType(keys=StringType(min=None, max=None, pattern=None), values=RefType(id='PropertySchema', display=None), min=None, max=None), display=DisplayValue(name='Properties', description='Properties of this object.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.ObjectSchema'>), 'PropertySchema': ObjectType(id='PropertySchema', properties={'type': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'pattern': RefType(id='PatternSchema', display=DisplayValue(name='Pattern', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None)), 'float': RefType(id='FloatSchema', display=DisplayValue(name='Float', description=None, icon=None)), 'bool': RefType(id='BoolSchema', display=DisplayValue(name='Bool', description=None, icon=None)), 'list': RefType(id='ListSchema', display=DisplayValue(name='List', description=None, icon=None)), 'map': RefType(id='MapSchema', display=DisplayValue(name='Map', description=None, icon=None)), 'scope': RefType(id='ScopeSchema', display=DisplayValue(name='Scope', description=None, icon=None)), 'object': RefType(id='ObjectSchema', display=DisplayValue(name='Object', description=None, icon=None)), 'one_of_string': RefType(id='OneOfStringSchema', display=DisplayValue(name='Multiple with string key', description=None, icon=None)), 'one_of_int': RefType(id='OneOfIntSchema', display=DisplayValue(name='Multiple with int key', description=None, icon=None)), 'ref': RefType(id='RefSchema', display=DisplayValue(name='Object reference', description=None, icon=None)), 'any': RefType(id='AnySchema', display=DisplayValue(name='Any', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Type', description='Type definition for this field.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'default': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Default', description='Default value for this property in JSON encoding. The value must be unserializable by the type specified in the type field. ', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'examples': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Examples', description='Example values for this property, encoded as JSON.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'required': PropertyType(type=BoolType(), display=DisplayValue(name='Required', description='When set to true, the value for this field must be provided under all circumstances.', icon=None), default='true', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'required_if': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Required if', description='Sets the current property to required if any of the properties in this list are set.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'required_if_not': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Required if not', description='Sets the current property to be required if none of the properties in this list are set.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None), 'conflicts': PropertyType(type=ListType(items=StringType(min=None, max=None, pattern=None), min=None, max=None), display=DisplayValue(name='Conflicts', description='The current property cannot be set if any of the listed properties are set.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.PropertySchema'>), 'StringEnumSchema': ObjectType(id='StringEnumSchema', properties={'values': PropertyType(type=MapType(keys=StringType(min=None, max=None, pattern=None), values=RefType(id='DisplayValue', display=None), min=1, max=None), display=DisplayValue(name='Values', description='Mapping where the left side of the map holds the possible value and the right side holds the display value for forms, etc.', icon=None), default=None, examples=['{"apple": {"name": "Apple"}, "orange": {"name": "Orange"}}'], required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.StringEnumSchema'>), 'DisplayValue': ObjectType(id='DisplayValue', properties={'name': PropertyType(type=StringType(min=1, max=None, pattern=None), display=DisplayValue(name='Name', description='Short text serving as a name or title for this item.', icon=None), default=None, examples=['"Fruit"'], required=False, required_if=None, required_if_not=None, conflicts=None), 'description': PropertyType(type=StringType(min=1, max=None, pattern=None), display=DisplayValue(name='Description', description='Description for this item if needed.', icon=None), default=None, examples=['"Please select the fruit you would like."'], required=False, required_if=None, required_if_not=None, conflicts=None), 'icon': PropertyType(type=StringType(min=1, max=None, pattern=None), display=DisplayValue(name='Icon', description='SVG icon for this item. Must have the declared size of 64x64, must not include additional namespaces, and must not reference external resources.', icon=None), default=None, examples=['"<svg ...></svg>"'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.DisplayValue'>), 'IntEnumSchema': ObjectType(id='IntEnumSchema', properties={'values': PropertyType(type=MapType(keys=IntType(min=None, max=None, units=None), values=RefType(id='DisplayValue', display=None), min=1, max=None), display=DisplayValue(name='Values', description='Possible values for this field.', icon=None), default=None, examples=['{"1024": {"name": "kB"}, "1048576": {"name": "MB"}}'], required=True, required_if=None, required_if_not=None, conflicts=None), 'units': PropertyType(type=RefType(id='Units', display=None), display=DisplayValue(name=None, description=None, icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.IntEnumSchema'>), 'Units': ObjectType(id='Units', properties={'base_unit': PropertyType(type=RefType(id='Unit', display=None), display=DisplayValue(name='Base unit', description='The base unit is the smallest unit of scale for this set of units.', icon=None), default=None, examples=['{"name_short_singular": "B", "name_short_plural": "B", "name_long_singular": "byte", "name_long_plural": "bytes"}'], required=True, required_if=None, required_if_not=None, conflicts=None), 'multipliers': PropertyType(type=MapType(keys=IntType(min=None, max=None, units=None), values=RefType(id='Unit', display=None), min=None, max=None), display=DisplayValue(name='Multipliers', description='A set of multiplies that describe multiple units of scale.', icon=None), default=None, examples=['{"1024": {"name_short_singular": "kB", "name_short_plural": "kB", "name_long_singular": "kilobyte", "name_long_plural": "kilobytes"}, "1048576": {"name_short_singular": "MB", "name_short_plural": "MB", "name_long_singular": "megabyte", "name_long_plural": "megabytes"}}'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.Units'>), 'Unit': ObjectType(id='Unit', properties={'name_short_singular': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Short name (singular)', description='Short name that can be printed in a few characters, singular form.', icon=None), default=None, examples=['"B"', '"char"'], required=True, required_if=None, required_if_not=None, conflicts=None), 'name_short_plural': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Short name (plural)', description='Short name that can be printed in a few characters, plural form.', icon=None), default=None, examples=['"B"', '"chars"'], required=True, required_if=None, required_if_not=None, conflicts=None), 'name_long_singular': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Long name (singular)', description='Longer name for this unit in singular form.', icon=None), default=None, examples=['"byte"', '"character"'], required=True, required_if=None, required_if_not=None, conflicts=None), 'name_long_plural': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Long name (plural)', description='Longer name for this unit in plural form.', icon=None), default=None, examples=['"bytes"', '"characters"'], required=True, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.Unit'>), 'StringSchema': ObjectType(id='StringSchema', properties={'min': PropertyType(type=IntType(min=0, max=None, units=Units(base_unit=Unit(name_short_singular='char', name_short_plural='chars', name_long_singular='character', name_long_plural='characters'), multipliers=None)), display=DisplayValue(name='Minimum length', description='Minimum length for this string (inclusive).', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=0, max=None, units=Units(base_unit=Unit(name_short_singular='char', name_short_plural='chars', name_long_singular='character', name_long_plural='characters'), multipliers=None)), display=DisplayValue(name='Maximum length', description='Maximum length for this string (inclusive).', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None), 'pattern': PropertyType(type=PatternType(), display=DisplayValue(name='Pattern', description='Regular expression this string must match.', icon=None), default=None, examples=['"^[a-zA-Z]+$"'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.StringSchema'>), 'PatternSchema': ObjectType(id='PatternSchema', properties={}, _cls=<class 'arcaflow_plugin_sdk.schema.PatternSchema'>), 'IntSchema': ObjectType(id='IntSchema', properties={'min': PropertyType(type=IntType(min=None, max=None, units=None), display=DisplayValue(name='Minimum value', description='Minimum value for this int (inclusive).', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=None, max=None, units=None), display=DisplayValue(name='Maximum value', description='Maximum value for this int (inclusive).', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None), 'units': PropertyType(type=RefType(id='Units', display=None), display=DisplayValue(name='Units', description='Units this number represents.', icon=None), default=None, examples=['{"base_unit": {"name_short_singular": "char", "name_short_plural": "chars", "name_long_singular": "character", "name_long_plural": "characters"}}'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.IntSchema'>), 'FloatSchema': ObjectType(id='FloatSchema', properties={'min': PropertyType(type=FloatType(min=None, max=None, units=None), display=DisplayValue(name='Minimum value', description='Minimum value for this float (inclusive).', icon=None), default=None, examples=['5.0'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=FloatType(min=None, max=None, units=None), display=DisplayValue(name='Maximum value', description='Maximum value for this float (inclusive).', icon=None), default=None, examples=['16.0'], required=False, required_if=None, required_if_not=None, conflicts=None), 'units': PropertyType(type=RefType(id='Units', display=None), display=DisplayValue(name='Units', description='Units this number represents.', icon=None), default=None, examples=['{"base_unit": {"name_short_singular": "%", "name_short_plural": "%", "name_long_singular": "percent", "name_long_plural": "percent"}}'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.FloatSchema'>), 'BoolSchema': ObjectType(id='BoolSchema', properties={}, _cls=<class 'arcaflow_plugin_sdk.schema.BoolSchema'>), 'ListSchema': ObjectType(id='ListSchema', properties={'items': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'pattern': RefType(id='PatternSchema', display=DisplayValue(name='Pattern', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None)), 'float': RefType(id='FloatSchema', display=DisplayValue(name='Float', description=None, icon=None)), 'bool': RefType(id='BoolSchema', display=DisplayValue(name='Bool', description=None, icon=None)), 'list': RefType(id='ListSchema', display=DisplayValue(name='List', description=None, icon=None)), 'map': RefType(id='MapSchema', display=DisplayValue(name='Map', description=None, icon=None)), 'scope': RefType(id='ScopeSchema', display=DisplayValue(name='Scope', description=None, icon=None)), 'object': RefType(id='ObjectSchema', display=DisplayValue(name='Object', description=None, icon=None)), 'one_of_string': RefType(id='OneOfStringSchema', display=DisplayValue(name='Multiple with string key', description=None, icon=None)), 'one_of_int': RefType(id='OneOfIntSchema', display=DisplayValue(name='Multiple with int key', description=None, icon=None)), 'ref': RefType(id='RefSchema', display=DisplayValue(name='Object reference', description=None, icon=None)), 'any': RefType(id='AnySchema', display=DisplayValue(name='Any', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Items', description='Type definition for items in this list.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'min': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Minimum items', description='Minimum number of items in this list.', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Maximum items', description='Maximum number of items in this list.', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.ListSchema'>), 'MapSchema': ObjectType(id='MapSchema', properties={'keys': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Keys', description='Type definition for map keys.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'values': PropertyType(type=OneOfStringType(types={'enum_string': RefType(id='StringEnumSchema', display=DisplayValue(name='String enum', description=None, icon=None)), 'enum_integer': RefType(id='IntEnumSchema', display=DisplayValue(name='Integer enum', description=None, icon=None)), 'string': RefType(id='StringSchema', display=DisplayValue(name='String', description=None, icon=None)), 'pattern': RefType(id='PatternSchema', display=DisplayValue(name='Pattern', description=None, icon=None)), 'integer': RefType(id='IntSchema', display=DisplayValue(name='Integer', description=None, icon=None)), 'float': RefType(id='FloatSchema', display=DisplayValue(name='Float', description=None, icon=None)), 'bool': RefType(id='BoolSchema', display=DisplayValue(name='Bool', description=None, icon=None)), 'list': RefType(id='ListSchema', display=DisplayValue(name='List', description=None, icon=None)), 'map': RefType(id='MapSchema', display=DisplayValue(name='Map', description=None, icon=None)), 'scope': RefType(id='ScopeSchema', display=DisplayValue(name='Scope', description=None, icon=None)), 'object': RefType(id='ObjectSchema', display=DisplayValue(name='Object', description=None, icon=None)), 'one_of_string': RefType(id='OneOfStringSchema', display=DisplayValue(name='Multiple with string key', description=None, icon=None)), 'one_of_int': RefType(id='OneOfIntSchema', display=DisplayValue(name='Multiple with int key', description=None, icon=None)), 'ref': RefType(id='RefSchema', display=DisplayValue(name='Object reference', description=None, icon=None)), 'any': RefType(id='AnySchema', display=DisplayValue(name='Any', description=None, icon=None))}, discriminator_field_name='type_id'), display=DisplayValue(name='Values', description='Type definition for map values.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'min': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Minimum items', description='Minimum number of items in this list.', icon=None), default=None, examples=['5'], required=False, required_if=None, required_if_not=None, conflicts=None), 'max': PropertyType(type=IntType(min=0, max=None, units=None), display=DisplayValue(name='Maximum items', description='Maximum number of items in this list.', icon=None), default=None, examples=['16'], required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.MapSchema'>), 'OneOfStringSchema': ObjectType(id='OneOfStringSchema', properties={'types': PropertyType(type=MapType(keys=StringType(min=None, max=None, pattern=None), values=OneOfStringType(types={'ref': RefType(id='RefSchema', display=None), 'scope': RefType(id='ScopeSchema', display=None), 'object': RefType(id='ObjectSchema', display=None)}, discriminator_field_name='type_id'), min=None, max=None), display=DisplayValue(name=None, description=None, icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'discriminator_field_name': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Discriminator field name', description='Name of the field used to discriminate between possible values. If this field ispresent on any of the component objects it must also be a string.', icon=None), default='"_type"', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.OneOfStringSchema'>), 'RefSchema': ObjectType(id='RefSchema', properties={'id': PropertyType(type=StringType(min=1, max=255, pattern=re.compile('^[$@a-zA-Z0-9-_]+$')), display=DisplayValue(name='ID', description='Referenced object ID.', icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'display': PropertyType(type=RefType(id='DisplayValue', display=None), display=DisplayValue(name='Display options', description='Name, description and icon.', icon=None), default=None, examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.RefSchema'>), 'OneOfIntSchema': ObjectType(id='OneOfIntSchema', properties={'types': PropertyType(type=MapType(keys=IntType(min=None, max=None, units=None), values=OneOfStringType(types={'ref': RefType(id='RefSchema', display=None), 'scope': RefType(id='ScopeSchema', display=None), 'object': RefType(id='ObjectSchema', display=None)}, discriminator_field_name='type_id'), min=None, max=None), display=DisplayValue(name=None, description=None, icon=None), default=None, examples=None, required=True, required_if=None, required_if_not=None, conflicts=None), 'discriminator_field_name': PropertyType(type=StringType(min=None, max=None, pattern=None), display=DisplayValue(name='Discriminator field name', description='Name of the field used to discriminate between possible values. If this field ispresent on any of the component objects it must also be an int.', icon=None), default='"_type"', examples=None, required=False, required_if=None, required_if_not=None, conflicts=None)}, _cls=<class 'arcaflow_plugin_sdk.schema.OneOfIntSchema'>), 'AnySchema': ObjectType(id='AnySchema', properties={}, _cls=<class 'arcaflow_plugin_sdk.schema.AnySchema'>)}, root='ScopeSchema')

This variable holds a constructed, serializable/unserializable schema for a scope. You can use it to send schemas to the engine, or to build an engine replacement. (This is normally handled by the plugin module.)

class arcaflow_plugin_sdk.schema.Schema(steps: Dict[str, StepSchema])

Bases: object

This is a collection of steps supported by a plugin.

steps: Dict[str, StepSchema]

exception arcaflow_plugin_sdk.schema.SchemaBuildException(path: Tuple[str], msg: str)

Bases: Exception

SchemaBuildException indicates an error while building the schema using type inspection. This exception holds the path to the parameter that caused the problem. The message should be easily understood, if you are having trouble with an error message, please open a ticket on the Arcaflow plugin SDK.

class arcaflow_plugin_sdk.schema.SchemaType(steps: Dict[str, StepSchema])

Bases: Schema

A schema is a definition of one or more steps that can be executed. The step has a defined input and output

call_step(step_id: str, input_param: Any) → Tuple[str, Any]

This function calls a specific step with the input parameter that has already been unserialized. It expects the data to be already valid, use unserialize_step_input to produce a valid input. This function is automatically called by __call__ after unserializing the input.

Parameters:

• step_id – The ID of the input step to run.

• input_param – The unserialized data structure the step expects.

Returns:

The ID of the output, and the data structure returned from the step.

call_step_signal(step_id: str, signal_id: str, unserialized_input_param: Any)

This function calls a specific step’s signal with the input parameter that has already been unserialized. It expects the data to be already valid, use unserialize_signal_input to produce a valid input.

Parameters:

• step_id – The ID of the input step to run.

• unserialized_input_param – The unserialized data structure the step expects.

Returns:

The ID of the output, and the data structure returned from the step.

get_signal(step_id: str, signal_id: str)

get_step(step_id: str)

serialize_output(step_id: str, output_id: str, output_data: Any) → Any

This function takes an output ID (e.g. “error”) and structured output_data and serializes them into a format suitable for wire transport. This function is automatically called by __call__ after the step is run.

Parameters:

• step_id – The step ID to use to look up the schema for serialization.

• output_id – The string identifier for the output data structure.

• output_data – The data structure returned from the step.

Returns:

steps: Dict[str, StepType]

unserialize_signal_handler_input(step_id: str, signal_id: str, serialized_data: Any) → Any

This function unserializes the input from a raw data to data structures, such as dataclasses. This function is automatically called by __call__ before running the step with the unserialized input.

Parameters:

• step_id – The step ID to use to look up the schema for unserialization.

• serialized_data – The raw data to unserialize.

Returns:

The unserialized data in the structure the step expects it.

unserialize_step_input(step_id: str, serialized_data: Any) → Any

This function unserializes the input from a raw data to data structures, such as dataclasses. This function is automatically called by __call__ before running the step with the unserialized input.

Parameters:

• step_id – The step ID to use to look up the schema for unserialization.

• serialized_data – The raw data to unserialize.

Returns:

The unserialized data in the structure the step expects it.

class arcaflow_plugin_sdk.schema.ScopeSchema(objects: Dict[str, ObjectSchema], root: str)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

A scope is a container for holding objects that can be referenced. It also optionally holds a reference to the root object of the current scope. References within the scope must always reference IDs in a scope. Scopes can be embedded into other objects, and scopes can have subscopes. Each RefSchema will reference objects in its current scope.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use ScopeType.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create a scope with an object:

>>> s = schema.ScopeSchema(
...     {
...         "a": schema.ObjectSchema("A", {})
...     },
...     "a",
... )

Create a ref:

>>> ref = schema.RefSchema("a")

Use ref in an object:

>>> s.objects["b"] = schema.ObjectSchema(
...     "B",
...    {
...        "a": schema.PropertySchema(ref)
...    }
... )

objects: Dict[str, ObjectSchema]

root: str

to_jsonschema()

to_openapi()

class arcaflow_plugin_sdk.schema.ScopeType(objects: Dict[str, ObjectType], root: str | None)

Bases: ScopeSchema, AbstractType

A scope is a container object for an object structure. Its main purpose is to hold objects that can be referenced, even in cases where circular references are desired. It mimics the ObjectType and provides several properties that proxy through to the underlying root object if set.

property cls: Type[ObjectT]

property id: str

objects: Dict[str, ObjectType]

property properties: Dict[str, PropertyType]

serialize(data: TypeT, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

unserialize(data: Any, path: Tuple[str] = ()) → TypeT

This function takes the underlying raw data and decodes it into the underlying advanced data type (e.g. dataclass) for usage.

Parameters:

• data – the raw data.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the advanced datatype.

Raises:

ConstraintException – if the passed data was not valid.

validate(data: TypeT, path: Tuple[str] = ())

This function validates an already unserialized data type and raises an exception if it does not match the type definition.

Parameters:

• data – the unserialized data.

• path – the path that lead to this validation call, in order to produce a nice error message

Raises:

ConstraintException – if the passed data was not valid.

class arcaflow_plugin_sdk.schema.SignalHandlerType(id: str, handler: Callable[[SignalDataT], None], data_schema: ScopeType, display: DisplayValue | None = None)

Bases: SignalSchema

SignalHandlerType describes a callable signal type.

class arcaflow_plugin_sdk.schema.SignalSchema(id: str, data_schema: ScopeSchema, display: DisplayValue | None = None)

Bases: object

Holds the definition for a single signal. This can be used for input or output signals.

To create, set the ID, and create a scope for the data input or output.

data_schema: ScopeSchema

display: DisplayValue | None = None

id: str

class arcaflow_plugin_sdk.schema.StepOutputSchema(schema: ScopeSchema, display: DisplayValue | None = None, error: bool = False)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds the possible outputs of a step and the metadata information related to these outputs.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use StepOutputType.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create scope:

>>> scope = schema.ScopeSchema(
...     {
...         "a": schema.ObjectSchema(
...             "a",
...             {},
...         )
...     },
...     "a",
... )

Create output schema:

>>> output = schema.StepOutputSchema(
...     scope,
...     schema.DisplayValue("Test output"),
...     error=False,
... )

display: DisplayValue | None = None

error: bool = False

schema: ScopeSchema

to_jsonschema()

Creates a JSON schema fragment for this output. This is useful when combining the output with other outputs into a comprehensive JSON schema output.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create scope:

>>> scope = schema.ScopeSchema(
...     {
...         "a": schema.ObjectSchema(
...             "a",
...             {
...                 "foo": schema.PropertySchema(schema.StringSchema())
...             },
...         )
...     },
...     "a",
... )

Create output schema:

>>> output = schema.StepOutputSchema(
...     scope,
...     schema.DisplayValue("Test output"),
...     error=False,
... )

Dump JSON schema:

>>> json_schema = output.to_jsonschema()
>>> json_schema["type"]
'object'
>>> json_schema["properties"]["foo"]["type"]
'string'

to_openapi()

Creates a OpenAPI fragment for this output. This is useful when combining the output with other outputs into a comprehensive OpenAPI document.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Create scope:

>>> scope = schema.ScopeSchema(
...     {
...         "a": schema.ObjectSchema(
...             "a",
...             {
...                 "foo": schema.PropertySchema(schema.StringSchema())
...             },
...         )
...     },
...     "a",
... )

Create output schema:

>>> output = schema.StepOutputSchema(
...     scope,
...     schema.DisplayValue("Test output"),
...     error=False,
... )

Dump OpenAPI schema:

>>> openapi = output.to_openapi()
>>> openapi["$ref"]
'#/components/schemas/a'
>>> openapi["components"]["schemas"]["a"]["properties"]["foo"]["type"]
'string'

class arcaflow_plugin_sdk.schema.StepOutputType(schema: ScopeSchema, display: DisplayValue | None = None, error: bool = False)

Bases: StepOutputSchema, AbstractType

This class holds the possible outputs of a step and the metadata information related to these outputs.

display: DisplayValue | None = None

error: bool = False

schema: ScopeType

serialize(data: TypeT, path: Tuple[str] = ()) → Any

This function serializes the passed data into it’s raw form for transport, e.g. string, int, dicts, list.

Parameters:

• data – the underlying data type to be serialized.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the raw datatype.

Raises:

ConstraintException – if the passed data was not valid.

unserialize(data: Any, path: Tuple[str] = ()) → TypeT

This function takes the underlying raw data and decodes it into the underlying advanced data type (e.g. dataclass) for usage.

Parameters:

• data – the raw data.

• path – the list of structural elements that lead to this point for error messages.

Returns:

the advanced datatype.

Raises:

ConstraintException – if the passed data was not valid.

validate(data: TypeT, path: Tuple[str] = ())

This function validates an already unserialized data type and raises an exception if it does not match the type definition.

Parameters:

• data – the unserialized data.

• path – the path that lead to this validation call, in order to produce a nice error message

Raises:

ConstraintException – if the passed data was not valid.

class arcaflow_plugin_sdk.schema.StepSchema(id: str, input: ScopeSchema, outputs: Dict[str, StepOutputSchema], signal_handlers: Dict[str, SignalSchema] | None = None, signal_emitters: Dict[str, SignalSchema] | None = None, display: DisplayValue | None = None)

Bases: object

This class holds the definition for a single step, it’s input and output definitions.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Define input:

>>> input_scope = schema.ScopeSchema(
...     {
...         "input": schema.ObjectSchema(
...             "input",
...             {
...                 "name": schema.PropertySchema(schema.StringSchema())
...             },
...         )
...     },
...     "input",
... )

Create output schema:

>>> output_scope = schema.ScopeSchema(
...     {
...         "greeting": schema.ObjectSchema(
...             "greeting",
...             {
...                 "text": schema.PropertySchema(schema.StringSchema())
...             },
...         )
...     },
...     "greeting",
... )
>>> output = schema.StepOutputSchema(
...     output_scope,
...     schema.DisplayValue("Test output"),
...     error=False,
... )

Create step schema:

>>> step_schema = schema.StepSchema(
...     "hello_world",
...     input_scope,
...     {"success": output},
...     schema.DisplayValue("Hello world!")
... )

display: DisplayValue | None = None

id: str

input: ScopeSchema

outputs: Dict[str, StepOutputSchema]

signal_emitters: Dict[str, SignalSchema] = None

signal_handlers: Dict[str, SignalSchema] = None

class arcaflow_plugin_sdk.schema.StepType(id: str, handler: Callable[[StepObjectT, StepInputT], Tuple[str, StepOutputT]], step_object_constructor: Callable[[], StepObjectT], input: ScopeType, outputs: Dict[str, StepOutputType], signal_handler_method_names: List[str], signal_emitters: Dict[str, SignalSchema] | None = None, display: DisplayValue | None = None)

Bases: StepSchema

StepSchema describes the schema for a single step. The input is always one ObjectType, while there are multiple possible outputs identified by a string.

initialized_object_data: StepObjectT

input: ScopeType

inspect_methods()

Retrieves the schemas from the method names given in the constructor.

object_cv: Condition = <Condition(<unlocked _thread.RLock object owner=0 count=0>, 0)>

object_data_ready: bool = False

outputs: Dict[str, StepOutputType]

signal_handler_method_names: List[str]

class arcaflow_plugin_sdk.schema.StringEnumSchema(values: Dict[str, DisplayValue])

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class specifically holds an enum that has string values. The values field maps the underlying values to display values for easy presentation.

This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use StringEnumType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> s = schema.StringEnumSchema({
...     "apple": schema.DisplayValue("Apple"),
...     "banana": schema.DisplayValue("Banana"),
...     "orange": schema.DisplayValue("Orange"),
... })
>>> s.valid_values()
['apple', 'banana', 'orange']

valid_values() → List[str]

values: Dict[str, DisplayValue]

class arcaflow_plugin_sdk.schema.StringEnumType(t: Type[EnumT])

Bases: _EnumType, StringEnumSchema

This class represents an enum type that is a string.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from enum import Enum

Create an enum:

>>> class Fruits(Enum):
...    APPLE="apple"
...    ORANGE="orange"

Create the type:

>>> fruit_type = schema.StringEnumType(Fruits)

Unserialize a value:

>>> fruit_type.unserialize("apple")
<Fruits.APPLE: 'apple'>

Serialize a value:

>>> fruit_type.serialize(Fruits.ORANGE)
'orange'

Validate a value:

>>> fruit_type.validate("plum")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: 'plum' is not a valid value for 'Fruits'

class arcaflow_plugin_sdk.schema.StringSchema(min: int | None = None, max: int | None = None, pattern: Pattern | None = None)

Bases: _JSONSchemaGenerator, _OpenAPIGenerator

This class holds schema information for strings. This dataclass only has the ability to hold the configuration but cannot serialize, unserialize or validate. For that functionality please use StringType.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> import re
>>> s = schema.StringSchema(
...     min=3,
...     max=5,
...     pattern=re.compile("^[a-z]+$")
... )

max: int | None = None

min: int | None = None

pattern: Pattern | None = None

class arcaflow_plugin_sdk.schema.StringType(min: int | None = None, max: int | None = None, pattern: Pattern | None = None)

Bases: StringSchema, AbstractType

StringType represents a string of characters for human consumption.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> import re
>>> s = schema.StringType(
...     min=3,
...     max=5,
...     pattern=re.compile("^[a-z]+$")
... )

You can now unserialize, validate, or serialize the data.

serialize(data: str, path: Tuple[str] = ()) → any

This function returns the string as-is after validating its contents.

unserialize(data: Any, path: Tuple[str] = ()) → str

Unserializes the current string from an integer or string.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> import re
>>> s = schema.StringType(
...     min=3,
...     max=5,
...     pattern=re.compile("^[0-9]+$")
... )
>>> s.unserialize(123)
'123'

validate(data: str, path: Tuple[str] = ())

Validates the given string for conformance with the rules set up in this object.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> import re
>>> s = schema.StringType(
...     min=3,
...     max=5,
...     pattern=re.compile("^[a-z]+$")
... )
>>> s.validate("as")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: String must be at least 3 characters, 2 given

>>> s.validate("asd")

>>> s.validate("123")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: String must match the pattern ^[a-z]+$

>>> s.validate("asdfgh")
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed: String must be at most 5 characters, 6 given

class arcaflow_plugin_sdk.schema.Unit(name_short_singular: str, name_short_plural: str, name_long_singular: str, name_long_plural: str)

Bases: object

A unit is a description of a single scale of measurement, such as a “second”. If there are multiple scales, such as “minute”, “second”, etc. then multiple of these unit classes can be composed into units.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> u = schema.Unit(
...     "ns",
...     "ns",
...     "nanosecond",
...     "nanoseconds"
... )

format_long(amount: int | float, display_zero: bool = True) → str

This function formats an amount according to this unit.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> u = schema.Unit(
...     "ns",
...     "ns",
...     "nanosecond",
...     "nanoseconds"
... )
>>> u.format_long(42)
'42 nanoseconds'
>>> u.format_long(1)
'1 nanosecond'
>>> u.format_long(0)
'0 nanoseconds'

format_short(amount: int | float, display_zero: bool = True) → str

This function formats an amount according to this unit.

Example:

>>> from arcaflow_plugin_sdk import schema
>>> u = schema.Unit(
...     "ns",
...     "ns",
...     "nanosecond",
...     "nanoseconds"
... )
>>> u.format_short(42)
'42ns'

name_long_plural: str

name_long_singular: str

name_short_plural: str

name_short_singular: str

exception arcaflow_plugin_sdk.schema.UnitParseException(msg: str)

Bases: Exception

This exception indicates that it failed to parse a unit string.

msg: str

class arcaflow_plugin_sdk.schema.Units(base_unit: Unit, multipliers: Dict[int, Unit] | None = None)

Bases: object

Units holds several scales of magnitude of the same unit, for example 5m30s.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Define your unit:

>>> t = schema.Units(
...     schema.Unit(
...         "ns",
...         "ns",
...         "nanosecond",
...         "nanoseconds"
...     ),
...     {
...         1000: schema.Unit(
...             "ms",
...             "ms",
...             "microsecond",
...             "microseconds"
...         ),
...         1000000: schema.Unit(
...             "s",
...             "s",
...             "second",
...             "seconds"
...         ),
...         60000000: schema.Unit(
...             "m",
...             "m",
...             "minute",
...             "minutes"
...         ),
...         3600000000: schema.Unit(
...             "H",
...             "H",
...             "hour",
...             "hours"
...         )
...     }
... )

Format your time:

>>> t.format_short(305000000)
'5m5s'

base_unit: Unit

format_long(data: int | float) → str

This function takes an integer and formats it so that it is the most readable based on the current set of units.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Use the pre-defined units to format a number:

>>> schema.UNIT_BYTE.format_long(1025)
'1 kilobyte 1 byte'

You can also format floats:

>>> schema.UNIT_PERCENT.format_long(1.1)
'1.1 percent'

format_short(data: int | float) → str

This function takes an integer and formats it so that it is the most readable based on the current set of units.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Use the pre-defined units to format a number:

>>> schema.UNIT_BYTE.format_short(1025)
'1kB1B'

You can also format floats:

>>> schema.UNIT_PERCENT.format_short(1.1)
'1.1%'

multipliers: Dict[int, Unit] | None = None

parse(data: str) → int | float

This function parses a string of units into the base number representation.

Raises:

UnitParseException – if the data string fails to parse.

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema

Parse a time string:

>>> schema.UNIT_TIME.parse("5m5s")
305000000

Parse percentages:

>>> schema.UNIT_PERCENT.parse("1.1%")
1.1

Parse bytes:

>>> schema.UNIT_BYTE.parse("5MB")
5242880

arcaflow_plugin_sdk.schema.build_object_schema(t, _skip_validation: bool = False) → ScopeType

This function builds a schema for a single object. This is useful when serializing input parameters into a file for underlying tools to use, or unserializing responses from underlying tools into output data types.

Parameters:

• t – the type to build a schema for.

• _skip_validation – Skip schema validation. For internal use only when constructing SCOPE_SCHEMA.

Returns:

the built object schema

arcaflow_plugin_sdk.schema.conflicts(conflicts: str) → Callable[[ValidatorT], ValidatorT]

This decorator creates a validation that triggers if the current field on an object is set in parallel with the specified field.

Parameters:

conflicts – The field to conflict with.

Returns:

the validator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         typing.Optional[str],
...         schema.conflicts("some_other_field")
...     ] = None
...     some_other_field: typing.Optional[str] = None

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

This is valid because neither field is set:

>>> unserialized_data = s.unserialize({})

This is not valid because both fields are set:

>>> unserialized_data = s.unserialize({"some_field": "bar", "some_other_field":"foo"})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'YourDataClass -> some_field': Field conflicts 'some_other_field', set one of the two, not both

This is valid because only the some_other_field is set:

>>> unserialized_data = s.unserialize({"some_other_field":"foo"})
>>> unserialized_data.some_other_field
'foo'

This is also valid because some_field is set: >>> unserialized_data = s.unserialize({“some_field”: “bar”}) >>> unserialized_data.some_field ‘bar’

arcaflow_plugin_sdk.schema.description(description: str)

The description annotation can be applied on any dataclass field, or on Union types to add a human-readable description to the field. It can contain line breaks and links for formatting. It is used as a form field description text or as part of a dropdown box in a form.

Example:

Imports:

>>> from dataclasses import dataclass
>>> from arcaflow_plugin_sdk import schema

Define your dataclass:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[str, schema.description("This is a string field")]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

Print field description:

>>> s["YourDataClass"].properties["some_field"].display.description
'This is a string field'

Parameters:

description – The description to apply

Returns:

Callable

arcaflow_plugin_sdk.schema.discriminator(discriminator_field_name: str) → Callable[[OneOfStringSchema | OneOfIntSchema], OneOfStringSchema | OneOfIntSchema]

This annotation is used to manually set the discriminator field on a Union type. :param discriminator_field_name: the name of the discriminator field. :return: the callable decorator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class A:
...     a: str
>>> @dataclass
... class B:
...     b: str
>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         typing.Union[A, B],
...         schema.discriminator("foo")
...     ]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

You can now deserialize a dataset:

>>> unserialized_data = s.unserialize({"some_field": {"foo": "A", "a": "Hello world!"}})
>>> unserialized_data.some_field.a
'Hello world!'

arcaflow_plugin_sdk.schema.discriminator_value(discriminator_value: str | int | Enum)

This annotation adds a custom value for an instance of a discriminator. The value must match the discriminator field

This annotation works only when used in conjunction with discriminator().

Parameters:

discriminator_value – The value for the discriminator field.

Returns:

The callable decorator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class A:
...     a: str
>>> @dataclass
... class B:
...     b: str
>>> @dataclass
... class YourDataClass:
...     some_field: typing.Union[
...         typing.Annotated[A, schema.discriminator_value("Foo")],
...         typing.Annotated[B, schema.discriminator_value("Bar")]
...     ]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

You can now deserialize a dataset:

>>> unserialized_data = s.unserialize({"some_field": {"_type": "Foo", "a": "Hello world!"}})
>>> unserialized_data.some_field.a
'Hello world!'

arcaflow_plugin_sdk.schema.example(example: Any) → Callable[[PropertySchema], PropertySchema]

This annotation provides the option to add an example to a type. :param example: the example as raw type, serializable by json.dumps. Do not use dataclasses

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclass:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         typing.Dict[str, str],
...         schema.example({"foo":"bar"})
...     ]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

Print field description:

>>> s["YourDataClass"].properties["some_field"].examples
['{"foo": "bar"}']

arcaflow_plugin_sdk.schema.icon(icon: str)

The icon annotation can be applied to any dataclass field, or on Union types to add a 64x64 pixel SVG icon to the item on display. However, the SVG must not contain any external sources or namespaces in order to work correctly.

Example:

Imports:

>>> from dataclasses import dataclass
>>> from arcaflow_plugin_sdk import schema

Define your dataclass:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[str, schema.icon("<svg></svg>")]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

Print field icon:

>>> s["YourDataClass"].properties["some_field"].display.icon
'<svg></svg>'

arcaflow_plugin_sdk.schema.id(id: str)

The id annotation can be used to change the serialized name of an object field. This is useful when a field must be serialized to a name that is not a valid Python field.

Example:

Imports:

>>> from dataclasses import dataclass
>>> from arcaflow_plugin_sdk import schema

Define your dataclass:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[str, schema.id("some-field")]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

Now unserialize your data:

>>> unserialized_data = s.unserialize({"some-field": "foo"})

This should now print “foo”:

>>> print(unserialized_data.some_field)
foo

Parameters:

id – The field to use

Returns:

Callable

arcaflow_plugin_sdk.schema.max(param: int) → Callable[[ValidatorT], ValidatorT]

This decorator creates a maximum length (strings), maximum number (int, float), or maximum element count (lists and maps) validation.

Parameters:

param – The maximum number

Returns:

the validator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         int,
...         schema.max(5)
...     ]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

The unserialization will validate the field:

>>> unserialized_data = s.unserialize({"some_field": 6})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'YourDataClass -> some_field': Must be at most 5
>>> unserialized_data = s.unserialize({"some_field": 4})
>>> unserialized_data.some_field
4

arcaflow_plugin_sdk.schema.min(param: int | float) → Callable[[ValidatorT], ValidatorT]

This decorator creates a minimum length (strings), minimum number (int, float), or minimum element count (lists and maps) validation.

Param:

The minimum number

Returns:

the validator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         int,
...         schema.min(5)
...     ]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

The unserialization will validate the field:

>>> unserialized_data = s.unserialize({"some_field": 4})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'YourDataClass -> some_field': Must be at least 5
>>> unserialized_data = s.unserialize({"some_field": 42})
>>> unserialized_data.some_field
42

arcaflow_plugin_sdk.schema.name(name: str)

The name annotation can be applied on any dataclass field, or on Union types to add a human-readable name to the field. It is used as a form field or as part of a dropdown box in a form.

Example:

Imports:

>>> from dataclasses import dataclass
>>> from arcaflow_plugin_sdk import schema

Define your dataclass:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[str, schema.name("Some field")]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

Print field name:

>>> s["YourDataClass"].properties["some_field"].display.name
'Some field'

Parameters:

name – The name to apply

Returns:

Callable

arcaflow_plugin_sdk.schema.pattern(pattern: Pattern) → Callable[[ValidatorT], ValidatorT]

This decorator creates a regular expression pattern validation for strings.

Parameters:

pattern – The regular expression.

Returns:

the validator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         str,
...         schema.pattern(re.compile("^[a-z]+$"))
...     ]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

The unserialization will validate the field:

>>> unserialized_data = s.unserialize({"some_field": "asdf1234"})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'YourDataClass -> some_field': String must match the pattern ^[a-z]+$
>>> unserialized_data = s.unserialize({"some_field": "asdf"})
>>> unserialized_data.some_field
'asdf'

arcaflow_plugin_sdk.schema.required_if(required_if: str) → Callable[[ValidatorT], ValidatorT]

This decorator creates a that marks the current field as required if the specified field is set.

Parameters:

required_if – The other field to use.

Returns:

the validator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         typing.Optional[str],
...         schema.required_if("some_other_field")
...     ] = None
...     some_other_field: typing.Optional[str] = None

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

This is a valid unserialization because both fields are unset:

>>> unserialized_data = s.unserialize({})
>>> unserialized_data.some_field
>>> unserialized_data.some_other_field

This is not valid because only some_other_field is set:

>>> unserialized_data = s.unserialize({"some_other_field":"foo"})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'YourDataClass -> some_field': This field is required because 'some_other_field' is set

This is valid again:

>>> unserialized_data = s.unserialize({"some_other_field":"foo", "some_field": "bar"})
>>> unserialized_data.some_field
'bar'
>>> unserialized_data.some_other_field
'foo'

arcaflow_plugin_sdk.schema.required_if_not(required_if_not: str) → Callable[[ValidatorT], ValidatorT]

This decorator creates a validation that marks the current field as required if the specified field is not set. If there are multiple of these validators, the current field is only marked as required if none of the specified fields are provided.

Parameters:

required_if_not – The other field to use.

Returns:

the validator

Example:

Imports:

>>> from arcaflow_plugin_sdk import schema
>>> from dataclasses import dataclass
>>> import typing

Your dataclasses:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[
...         typing.Optional[str],
...         schema.required_if_not("some_other_field")
...     ] = None
...     some_other_field: typing.Optional[str] = None

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

This is not valid because neither of the fields are set:

>>> unserialized_data = s.unserialize({})
Traceback (most recent call last):
...
arcaflow_plugin_sdk.schema.ConstraintException: Validation failed for 'YourDataClass -> some_field': This field is required because 'some_other_field' is not set

This is valid because the some_other_field is set:

>>> unserialized_data = s.unserialize({"some_other_field":"foo"})
>>> unserialized_data.some_other_field
'foo'

This is also valid because some_field is set:

>>> unserialized_data = s.unserialize({"some_field": "bar"})
>>> unserialized_data.some_field
'bar'

Both fields can also be set:

>>> unserialized_data = s.unserialize({"some_field": "bar", "some_other_field":"foo"})
>>> unserialized_data.some_other_field
'foo'
>>> unserialized_data.some_field
'bar'

arcaflow_plugin_sdk.schema.test_object_serialization(dc, fail: Callable[[str], None] | None = None, t: ObjectType | None = None)

This function aids serialization by first serializing, then unserializing the passed parameter according to the passed schema. It then compares that the two objects are equal.

Parameters:

• dc – the dataclass to use for tests.

• t – the schema for the dataclass. If none is passed, the schema is built automatically using schema.build_object_schema()

arcaflow_plugin_sdk.schema.units(units: Units)

This annotation lets you add unit definitions to int and float fields. This helps with determining how to treat that number, but also contains scaling information for creating a nicely formatted string, such as 5m30s.

Example:

Imports:

>>> from dataclasses import dataclass
>>> from arcaflow_plugin_sdk import schema

Define your dataclass:

>>> @dataclass
... class YourDataClass:
...     some_field: typing.Annotated[int, schema.units(schema.UNIT_BYTE)]

Build your schema:

>>> s = schema.build_object_schema(YourDataClass)

Print format data:

>>> s["YourDataClass"].properties["some_field"].type.units.format_short(5)
'5B'