queenbee.io.outputs.step module¶

Output types for a Queenbee Job steps.

For more information on plugins see plugin module.

class queenbee.io.outputs.step.StepArrayOutput(*, type: ~typing.Literal['StepArrayOutput'] = 'StepArrayOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, items_type: ~queenbee.io.common.ItemType = ItemType.String, value: ~typing.List)[source]¶

Bases: FunctionArrayOutput

A JSON array output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['StepArrayOutput']¶

value: List¶

class queenbee.io.outputs.step.StepBooleanOutput(*, type: ~typing.Literal['StepBooleanOutput'] = 'StepBooleanOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, value: bool)[source]¶

Bases: FunctionBooleanOutput

The boolean type matches only two special values: True and False.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['StepBooleanOutput']¶

value: bool¶

class queenbee.io.outputs.step.StepFileOutput(*, type: ~typing.Literal['StepFileOutput'] = 'StepFileOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, source: ~queenbee.io.artifact_source.HTTP | ~queenbee.io.artifact_source.S3 | ~queenbee.io.artifact_source.ProjectFolder)[source]¶

Bases: FunctionFileOutput

A file output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source: HTTP | S3 | ProjectFolder¶

type: Literal['StepFileOutput']¶

class queenbee.io.outputs.step.StepFolderOutput(*, type: ~typing.Literal['StepFolderOutput'] = 'StepFolderOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, source: ~queenbee.io.artifact_source.HTTP | ~queenbee.io.artifact_source.S3 | ~queenbee.io.artifact_source.ProjectFolder)[source]¶

Bases: FunctionFolderOutput

A folder output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source: HTTP | S3 | ProjectFolder¶

type: Literal['StepFolderOutput']¶

class queenbee.io.outputs.step.StepIntegerOutput(*, type: ~typing.Literal['StepIntegerOutput'] = 'StepIntegerOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, value: int)[source]¶

Bases: FunctionIntegerOutput

An integer output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['StepIntegerOutput']¶

value: int¶

class queenbee.io.outputs.step.StepJSONObjectOutput(*, type: ~typing.Literal['StepJSONObjectOutput'] = 'StepJSONObjectOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, value: ~typing.Dict)[source]¶

Bases: FunctionJSONObjectOutput

A JSON object output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['StepJSONObjectOutput']¶

value: Dict¶

class queenbee.io.outputs.step.StepNumberOutput(*, type: ~typing.Literal['StepNumberOutput'] = 'StepNumberOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, value: float)[source]¶

Bases: FunctionNumberOutput

A number output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['StepNumberOutput']¶

value: float¶

class queenbee.io.outputs.step.StepPathOutput(*, type: ~typing.Literal['StepPathOutput'] = 'StepPathOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, source: ~queenbee.io.artifact_source.HTTP | ~queenbee.io.artifact_source.S3 | ~queenbee.io.artifact_source.ProjectFolder)[source]¶

Bases: FunctionPathOutput

A file or a folder output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source: HTTP | S3 | ProjectFolder¶

type: Literal['StepPathOutput']¶

class queenbee.io.outputs.step.StepStringOutput(*, type: ~typing.Literal['StepStringOutput'] = 'StepStringOutput', annotations: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str, description: str | None = None, path: str, required: bool = True, value: str)[source]¶

Bases: FunctionStringOutput

A String output.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['StepStringOutput']¶

value: str¶

Generate a step output from a template output type and a value

Parameters:

{Union[DAGOutputs (template) – template (DAG or Function)
a (FunctionOutputs]} -- An output from) – template (DAG or Function)
in (value {Any} -- The output value calculated for this template) – the workflow step

Returns:

StepOutputs – A Step Output object