ogdc_runner.models.recipe_config module

class ogdc_runner.models.recipe_config.DataOneInput(**data: Any) → None

Bases: InputParam

DataOne input parameters.

Parameters:

• data (Any)

• type (Literal['dataone'])

• dataset_identifier (str)

• filename (str | None)

• dataset_pid (str | None)

• resolved_objects (list[dict[str, Any]])

dataset_identifier: str

dataset_pid: str | None

filename: str | None

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

resolve_dataone_inputs() → DataOneInput

Resolve DataONE dataset identifiers to data object URLs.

Return type:

DataOneInput

resolved_objects: list[dict[str, Any]]

type: Literal['dataone']

class ogdc_runner.models.recipe_config.DataOneRecipeOutput(**data: Any) → None

Bases: RecipeOutput

DataONE output configuration.

Recipe outputs will be published to a DataONE dataset.

Parameters:

• data (Any)

• type (Literal['dataone'])

• dataone_id (str)

dataone_id: str

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['dataone']

class ogdc_runner.models.recipe_config.InputParam(**data: Any) → None

Bases: OgdcBaseModel

Input parameter for a recipe.

Parameters:

• data (Any)

• type (Literal['url', 'pvc_mount', 'file_system', 'dataone'])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['url', 'pvc_mount', 'file_system', 'dataone']

class ogdc_runner.models.recipe_config.ParallelConfig(**data: Any) → None

Bases: OgdcBaseModel

Configuration for parallel execution behavior.

enabled

Whether parallel execution is enabled

partition_strategy

Strategy for dividing work (“files” or “file_chunks”)

partition_size

Number of files per partition; None means use the orchestrator default (typically 1 000)

max_parallelism

Maximum number of concurrent Argo pods across all parallel stages; None means no cap beyond the cluster default

Parameters:

• data (Any)

• enabled (bool)

• partition_strategy (Literal['files'])

• partition_size (int | None)

• max_parallelism (int | None)

enabled: bool

max_parallelism: int | None

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

classmethod must_be_positive(v: int | None) → int | None

Parameters:

v (int | None)

Return type:

int | None

partition_size: int | None

partition_strategy: Literal['files']

class ogdc_runner.models.recipe_config.PvcRecipeOutput(**data: Any) → None

Bases: RecipeOutput

PVC output configuration.

Recipe outputs will be stored on a specified PVC in kubernetes.

Parameters:

• data (Any)

• type (Literal['pvc'])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['pvc']

class ogdc_runner.models.recipe_config.RecipeConfig(**data: Any) → None

Bases: RecipeMeta

Model for a recipe’s configuration.

This includes the data in meta.yaml, plus some internal metadata/config that is generated dynamically at runtime (e.g., recipe_directory).

Parameters:

• data (Any)

• name (str)

• workflow (ShellWorkflow | VizWorkflow)

• input (RecipeInput)

• output (DataOneRecipeOutput | TemporaryRecipeOutput | PvcRecipeOutput)

• recipe_directory (Path)

property id: str

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

recipe_directory: Path

class ogdc_runner.models.recipe_config.RecipeImage(**data: Any) → None

Bases: OgdcBaseModel

Image configuration for the recipe.

Supports both local and hosted Docker images.

Parameters:

• data (Any)

• image (str)

• tag (str)

property full_image_path: str

Return the full image path including tag.

image: str

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

tag: str

class ogdc_runner.models.recipe_config.RecipeInput(**data: Any) → None

Bases: OgdcBaseModel

Parameters:

• data (Any)

• params (list[DataOneInput | UrlInput])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

params: list[DataOneInput | UrlInput]

classmethod validate_params(params: list[DataOneInput | UrlInput]) → list[DataOneInput | UrlInput]

Ensure there’s at least one input parameter.

Parameters:

params (list[DataOneInput | UrlInput])

Return type:

list[DataOneInput | UrlInput]

class ogdc_runner.models.recipe_config.RecipeMeta(**data: Any) → None

Bases: OgdcBaseModel

Model for a recipe’s metadata (meta.yaml).

Parameters:

• data (Any)

• name (str)

• workflow (ShellWorkflow | VizWorkflow)

• input (RecipeInput)

• output (DataOneRecipeOutput | TemporaryRecipeOutput | PvcRecipeOutput)

input: RecipeInput

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

name: str

output: DataOneRecipeOutput | TemporaryRecipeOutput | PvcRecipeOutput

workflow: ShellWorkflow | VizWorkflow

class ogdc_runner.models.recipe_config.RecipeOutput(**data: Any) → None

Bases: OgdcBaseModel

Base model for recipe output configuration.

Parameters:

• data (Any)

• type (Literal['dataone', 'temporary', 'pvc'])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['dataone', 'temporary', 'pvc']

class ogdc_runner.models.recipe_config.ShellWorkflow(**data: Any) → None

Bases: Workflow

Model representing the shell workflow configuration.

Requires that the recipe_directory context be set when instantiating the class. E.g.,:

workflow = ShellWorkflow.model_validate(

{“sh_file”: “recipe.sh”}, context={“recipe_directory”: Path(“/path/to/recipe_directory”)},

),

Parameters:

• data (Any)

• type (Literal['shell'])

• sh_file (str | Path)

• parallel (ParallelConfig)

get_commands_from_sh_file() → list[str]

Returns a list of commands run from the workflow sh_file.

Return type:

list[str]

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

parallel: ParallelConfig

sh_file: str | Path

sh_file_path(info: ValidationInfo) → Self

Model-level validator that constructs a full path to sh_file.

Parameters:

info (ValidationInfo)

Return type:

Self

type: Literal['shell']

class ogdc_runner.models.recipe_config.TemporaryRecipeOutput(**data: Any) → None

Bases: RecipeOutput

Temporary output configuration.

Recipe outputs will be stored temporarily in a user-accessible location.

Parameters:

• data (Any)

• type (Literal['temporary'])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['temporary']

class ogdc_runner.models.recipe_config.UrlInput(**data: Any) → None

Bases: InputParam

Inpurt from URL.

When instantiated with context={“check_urls”: True}, URL-type parameters will be validated to ensure they are accessible via HTTP HEAD request.

Parameters:

• data (Any)

• type (Literal['url'])

• value (AnyUrl | str)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['url']

validate_url_accessible(info: ValidationInfo) → Self

Validate that URL-type parameters are accessible.

Parameters:

info (ValidationInfo)

Return type:

Self

value: AnyUrl | str

class ogdc_runner.models.recipe_config.VizWorkflow(**data: Any) → None

Bases: Workflow

Model representing the visualization workflow configuration.

Requires that the recipe_directory context be set when instantiating the class. E.g.,:

workflow = VizWorkflow.model_validate(

{“config_file”: “config.json”}, context={“recipe_directory”: Path(“/path/to/recipe_directory”)},

),

Parameters:

• data (Any)

• type (Literal['visualization'])

• config_file (str | Path | None)

• parallel (ParallelConfig)

config_file: str | Path | None

config_file_path(info: ValidationInfo) → Self

Model-level validator that constructs a full path to sh_file.

Parameters:

info (ValidationInfo)

Return type:

Self

get_config_file_json() → str

Get the viz workflow config as json.

If passed a JSON file, read the file content and return. Otherwise, an empty configuration will be returned (“{}”).

This configuration is used by the pdgworkflow for visualization workflows. When an empty config ({}) is returned, WorkflowManager will use its default behavior.

For documentation on available configuration options, see: - ConfigManager documentation: https://github.com/PermafrostDiscoveryGateway/viz-workflow/blob/feature-wf-k8s/pdgworkflow/ConfigManager.py - Example config: https://github.com/QGreenland-Net/ogdc-recipes/blob/main/recipes/viz-workflow/config.json

Return type:

str

Returns:

The content of the config.json file as a string, or empty JSON if file doesn’t exist. An empty config ({}) will cause ConfigManager to use default behavior.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

parallel: ParallelConfig

type: Literal['visualization']

class ogdc_runner.models.recipe_config.Workflow(**data: Any) → None

Bases: OgdcBaseModel

Parameters:

• data (Any)

• type (Literal['shell', 'visualization'])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

type: Literal['shell', 'visualization']