queenbee.recipe.recipe module

Queenbee recipe class.

Recipe is a collection of plugins and inter-related tasks that describes an end to end steps for the recipe.

class queenbee.recipe.recipe.BakedRecipe(*, type: ~typing.Literal['BakedRecipe'] = 'BakedRecipe', annotations: ~typing.Dict[str, ~typing.Any] = <factory>, api_version: ~typing.Literal['v1beta1'] = 'v1beta1', metadata: ~queenbee.base.metadata.MetaData | None = None, dependencies: ~typing.List[~queenbee.recipe.dependency.Dependency] | None = None, flow: ~typing.List[~queenbee.recipe.dag.DAG], digest: str, templates: ~typing.List[~queenbee.recipe.recipe.TemplateFunction | ~queenbee.recipe.dag.DAG])

Bases: Recipe

Baked Recipe.

A Baked Recipe contains all the templates referred to in the DAG within a templates list.

check_inputs()

Check DAG Task inputs match template inputs

classmethod check_template_dependency_names(v: List[DAG], info: ValidationInfo) → List[DAG]

Overwrite test function from inherited class

digest: str

classmethod from_folder(folder_path: str, refresh_deps: bool = True, config: ~queenbee.config.Config = type: Config annotations: {} auth: [] repositories: [])

Generate a baked recipe from a recipe folder

Note

Here is an example of a Recipe folder:

.
├── .dependencies
│   ├── plugins
│   │   └── <sha-256>.yaml
│   └── recipes
│       └── <sha-256>.yaml
├── flow
|   ├── sub-dag.yaml
│   └── main.yaml
├── dependencies.yaml
└── package.yaml

Parameters:

read (folder_path {str} -- The path to the folder to)

Keyword Arguments:

• of (refresh_deps {bool} -- Fetch the dependencies from their source instead) – the .dependencies folder (default: {True})

• (default (config {Config} -- A queenbee config object) – {Config()})

Returns:

BakedRecipe – A baked recipe

classmethod from_recipe(recipe: ~queenbee.recipe.recipe.Recipe, config: ~queenbee.config.Config = type: Config annotations: {} auth: [] repositories: [])

Bake a recipe

Parameters:

recipe (recipe {Recipe} -- A Queenbee)

Keyword Arguments:

(default (config {Config} -- A queenbee config object) – {Config()})

Raises:

ValueError – The dependencies or templates do not match the flow

Returns:

BakedRecipe – A baked recipe

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod remove_duplicates(v: List[TemplateFunction | DAG]) → List[TemplateFunction | DAG]

Remove duplicated templates by name

classmethod replace_template_refs(dependencies: List[Dependency], dags: List[DAG], digest_dict: Dict[str, str]) → List[DAG]

Replace DAG Task template names with unique dependency digest based names

Parameters:

• dependencies (dependencies {List[Dependency]} -- A list of locked)

• DAGs (dags {List[DAG]} -- A list of)

• Recipe (digest {str} -- The digest of the Recipe used to create the Baked)

Raises:

ValueError – Unresolvable dependency in a task

Returns:

List[DAG] – A list of DAGs with updated template names

property root_dag: DAG

static template_by_name(templates: List[TemplateFunction | DAG], name: str) → DAG | TemplateFunction

Retrieve a template from a list by name

Parameters:

• {List[Union[DAG (templates)

• templates (TemplateFunction]]} -- A list of)

• byt (name {str} -- The name to retrieve a template)

Raises:

ValueError – Template not found

Returns:

Union[DAG, TemplateFunction] – A template

templates: List[TemplateFunction | DAG]

type: Literal['BakedRecipe']

class queenbee.recipe.recipe.Recipe(*, type: ~typing.Literal['Recipe'] = 'Recipe', annotations: ~typing.Dict[str, ~typing.Any] = <factory>, api_version: ~typing.Literal['v1beta1'] = 'v1beta1', metadata: ~queenbee.base.metadata.MetaData | None = None, dependencies: ~typing.List[~queenbee.recipe.dependency.Dependency] | None = None, flow: ~typing.List[~queenbee.recipe.dag.DAG])

Bases: BaseModel

A Queenbee Recipe

api_version: Literal['v1beta1']

classmethod check_dag_names(v: List[DAG], info: ValidationInfo) → List[DAG]

Check DAG names do not overlap with dependency names

classmethod check_entrypoint(v: List[DAG]) → List[DAG]

Check the main DAG exists in the flow

classmethod check_template_dependency_names(v: List[DAG], info: ValidationInfo) → List[DAG]

Check all DAG template names exist in dependencies or other DAGs

static dag_by_name(flow: List[DAG], name: str) → DAG

Retrieve a DAG from a list by its name

Parameters:

• DAGs (flow {List[DAG]} -- A list of)

• by (name {str} -- The name to search)

Raises:

ValueError – No DAG found with the input name

Returns:

DAG – A DAG

dependencies: List[Dependency] | None

static dependency_by_name(dependencies: List[Dependency], name: str) → Dependency

Retrieve a dependency by its reference name (name or alias)

Parameters:

• dependencies (dependencies {List[Dependency]} -- A list of)

• by (name {str} -- The name or alias to search dependencies)

Raises:

ValueError – No dependency found with the input name

Returns:

Dependency – A dependency

flow: List[DAG]

classmethod from_folder(folder_path: str)

Generate a recipe from a folder

Note

Here is an example of a Recipe folder:

.
├── .dependencies
│   ├── plugins
│   │   └── <sha-256>.yaml
│   └── recipes
│       └── <sha-256>.yaml
├── flow
|   ├── sub-dag.yaml
│   └── main.yaml
├── dependencies.yaml
└── package.yaml

Parameters:

folder (folder_path {str} -- Path to the)

Returns:

Recipe – A Recipe

property inputs: List[DAGStringInput | DAGIntegerInput | DAGNumberInput | DAGBooleanInput | DAGFolderInput | DAGFileInput | DAGPathInput | DAGArrayInput | DAGJSONObjectInput | DAGGenericInput]

Get the Recipe’s inputs

Returns:

List[DAGInputs] – The inputs of the entrypoint DAG

property is_locked: bool

Indicates whether the Recipe dependencies are all locked

Returns:

bool – True if all dependencies are locked

lock_dependencies(config: ~queenbee.config.Config = type: Config annotations: {} auth: [] repositories: [])

Lock the dependencies by fetching them and storing their digest

metadata: MetaData | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property outputs: List[DAGGenericOutput | DAGStringOutput | DAGIntegerOutput | DAGNumberOutput | DAGBooleanOutput | DAGFolderOutput | DAGFileOutput | DAGPathOutput | DAGArrayOutput | DAGJSONObjectOutput]

Get the Recipe’s outputs

Returns:

List[DAGOutputs] – The outputs of the entrypoint DAG

property root_dag: DAG

classmethod sort_list(v: List[DAG]) → List[DAG]

Sort the list of DAGs by name

to_folder(folder_path: str, readme_string: str = None)

Write a Recipe to a folder

Note

Here is an example of a Recipe folder:

.
├── .dependencies
│   ├── plugin
│   │   └── plugin-dep-name
│   │       ├── functions
│   │       │   ├── func-1.yaml
│   │       │   ├── ...
│   │       │   └── func-n.yaml
│   │       ├── config.yaml
│   │       └── package.yaml
│   └── recipe
│       └── recipe-dep-name
│           ├── .dependencies
│           │   ├── plugin
│           │   └── recipe
│           ├── flow
│           │   └── main.yaml
│           ├── dependencies.yaml
│           └── package.yaml
├── flow
│   └── main.yaml
├── dependencies.yaml
└── package.yaml

Parameters:

folder (folder_path {str} -- The path to the Recipe)

Keyword Arguments:

(default (readme_string {str} -- The README file string) – {None})

type: Literal['Recipe']

write_dependencies(folder_path: str, config: ~queenbee.config.Config = type: Config annotations: {} auth: [] repositories: [])

Fetch dependencies manifests and write them to the .dependencies folder

Parameters:

folder (folder_path {str} -- Path to the recipe)

Keyword Arguments:

(default (config {Config} -- A queenbee config object) – {Config()})

write_dependency_file(folder_path: str)

Write the locked dependencies to a Recipe folder

Parameters:

folder (folder_path {str} -- The path to the recipe)

class queenbee.recipe.recipe.RecipeInterface(*, type: ~typing.Literal['RecipeInterface'] = 'RecipeInterface', annotations: ~typing.Dict[str, ~typing.Any] = <factory>, api_version: ~typing.Literal['v1beta1'] = 'v1beta1', metadata: ~queenbee.base.metadata.MetaData, source: str = None, inputs: ~typing.List[~queenbee.io.inputs.dag.DAGStringInput | ~queenbee.io.inputs.dag.DAGIntegerInput | ~queenbee.io.inputs.dag.DAGNumberInput | ~queenbee.io.inputs.dag.DAGBooleanInput | ~queenbee.io.inputs.dag.DAGFolderInput | ~queenbee.io.inputs.dag.DAGFileInput | ~queenbee.io.inputs.dag.DAGPathInput | ~queenbee.io.inputs.dag.DAGArrayInput | ~queenbee.io.inputs.dag.DAGJSONObjectInput | ~queenbee.io.inputs.dag.DAGGenericInput] = None, outputs: ~typing.List[~queenbee.io.outputs.dag.DAGGenericOutput | ~queenbee.io.outputs.dag.DAGStringOutput | ~queenbee.io.outputs.dag.DAGIntegerOutput | ~queenbee.io.outputs.dag.DAGNumberOutput | ~queenbee.io.outputs.dag.DAGBooleanOutput | ~queenbee.io.outputs.dag.DAGFolderOutput | ~queenbee.io.outputs.dag.DAGFileOutput | ~queenbee.io.outputs.dag.DAGPathOutput | ~queenbee.io.outputs.dag.DAGArrayOutput | ~queenbee.io.outputs.dag.DAGJSONObjectOutput] = None)

Bases: BaseModel

An interface object for creating a Recipe.

Recipe information only includes metadata, source, inputs and outputs of a Recipe. This object is useful for creating user interface for Recipes.

api_version: Literal['v1beta1']

classmethod create_empty_list(v: list) → list

classmethod from_recipe(recipe: Recipe | BakedRecipe, source: str = None)

Create a Recipe info from a recipe.

inputs: List[DAGStringInput | DAGIntegerInput | DAGNumberInput | DAGBooleanInput | DAGFolderInput | DAGFileInput | DAGPathInput | DAGArrayInput | DAGJSONObjectInput | DAGGenericInput]

metadata: MetaData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

outputs: List[DAGGenericOutput | DAGStringOutput | DAGIntegerOutput | DAGNumberOutput | DAGBooleanOutput | DAGFolderOutput | DAGFileOutput | DAGPathOutput | DAGArrayOutput | DAGJSONObjectOutput]

source: str

type: Literal['RecipeInterface']

class queenbee.recipe.recipe.TemplateFunction(*, type: ~typing.Literal['TemplateFunction'] = 'TemplateFunction', annotations: ~typing.Dict[str, ~typing.Any] = <factory>, inputs: ~typing.List[~typing.Annotated[~queenbee.io.inputs.function.FunctionStringInput | ~queenbee.io.inputs.function.FunctionIntegerInput | ~queenbee.io.inputs.function.FunctionNumberInput | ~queenbee.io.inputs.function.FunctionBooleanInput | ~queenbee.io.inputs.function.FunctionFolderInput | ~queenbee.io.inputs.function.FunctionFileInput | ~queenbee.io.inputs.function.FunctionPathInput | ~queenbee.io.inputs.function.FunctionArrayInput | ~queenbee.io.inputs.function.FunctionJSONObjectInput, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = <factory>, outputs: ~typing.List[~typing.Annotated[~queenbee.io.outputs.function.FunctionStringOutput | ~queenbee.io.outputs.function.FunctionIntegerOutput | ~queenbee.io.outputs.function.FunctionNumberOutput | ~queenbee.io.outputs.function.FunctionBooleanOutput | ~queenbee.io.outputs.function.FunctionFolderOutput | ~queenbee.io.outputs.function.FunctionFileOutput | ~queenbee.io.outputs.function.FunctionPathOutput | ~queenbee.io.outputs.function.FunctionArrayOutput | ~queenbee.io.outputs.function.FunctionJSONObjectOutput, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = <factory>, name: str, description: str | None = None, command: str | None = None, language: ~queenbee.plugin.function.ScriptingLanguages = ScriptingLanguages.Python, source: str | None = None, config: ~queenbee.plugin.plugin.PluginConfig)

Bases: Function

Function template.

config: PluginConfig

classmethod from_plugin(plugin: Plugin) → list

Generate a list of template functions from a plugin

Parameters:

plugin (plugin {Plugin} -- A)

Returns:

list – A list of template functions

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['TemplateFunction']