Procodile API

The Procodile Python API is provided by the procodile package.

procodile.ProcessRegistry

ProcessRegistry()

Bases: Mapping[str, Process]

A registry for processes.

Processes are Python functions with extra metadata and can be extended to create workflows using steps decorator.

A Workflow consists of one or more Python functions with metadata, designed to execute sequentially by resolving dependencies and passing outputs to downstream steps.

This class provides a read-only mapping from unique identifiers to facade-like Process instances. While the user interacts with these processes, the registry internally manages full [Workflow][procodile.workflow.Workflow] instances.

The internal Workflow objects hold the source-of-truth metadata required for dependency resolution and execution, while the exposed Process objects serve as the public interface for client interaction.

main

main(
    function: Callable | None = None,
    /,
    *,
    id: Optional[str] = None,
    version: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    inputs: Optional[dict[str, FieldInfo | InputDescription]] = None,
    outputs: Optional[dict[str, FieldInfo | OutputDescription]] = None,
    inputs_arg: str | bool = False,
) -> Callable[[Callable], Workflow] | Callable

A decorator that can be applied to a user function in order to register it as a process in this registry.

Note:

- Use `main` decorator to express a process that comprises multiple steps
  that require a reference to the main entry point.

- Use `process` decorator to express a process that has no steps,
  hence requires no reference to a main step.

The decorator can be used with or without parameters.

Parameters:

Name	Type	Description	Default
`function`	`Callable \| None`	The decorated function that is passed automatically since `process()` is a decorator function.	`None`
`id`	`Optional[str]`	Optional process identifier. Must be unique within the registry. If not provided, the fully qualified function name will be used.	`None`
`version`	`Optional[str]`	Optional version identifier. If not provided, `"0.0.0"` will be used.	`None`
`title`	`Optional[str]`	Optional, short process title.	`None`
`description`	`Optional[str]`	Optional, detailed description of the process. If not provided, the function's docstring, if any, will be used.	`None`
`inputs`	`Optional[dict[str, FieldInfo \| InputDescription]]`	Optional mapping from function argument names to `pydantic.Field` or `InputDescription` instances. The preferred way is to annotate the arguments directly as described in The Annotated Pattern. Use `InputDescription` instances to pass extra information that cannot be represented by a `pydantic.Field`, e.g., `additionalParameters` or `keywords`.	`None`
`outputs`	`Optional[dict[str, FieldInfo \| OutputDescription]]`	Mapping from output names to `pydantic.Field` or `OutputDescription` instances. Required, if you have multiple outputs returned as a dictionary. In this case, the function must return a typed `tuple` and output names refer to the items of the tuple in given order.	`None`
`inputs_arg`	`str \| bool`	Specifies the use of an inputs argument. An inputs argument is a container for the actual process inputs. If specified, it must be the only function argument (besides an optional job context argument) and must be a subclass of `pydantic.BaseModel`. If `inputs_arg` is `True` the only argument will be the input argument, if `inputs_arg` is a `str` it must be the name of the only argument.	`False`

procodile.Process `dataclass`

Process(
    function: Callable,
    signature: inspect.Signature,
    model_class: type[BaseModel],
    description: ProcessDescription,
    inputs_arg: str | None,
    job_ctx_arg: str | None,
)

A process comprises a process description and executable code in form of a Python function.

Instances of this class are be managed by the ProcessRegistry.

Attributes:

Name	Type	Description
`function`	`Callable`	The user's Python function.
`signature`	`inspect.Signature`	The signature of `function`.
`job_ctx_arg`	`str \| None`	Names of `function` arguments of type `JobContext`.
`model_class`	`type[BaseModel]`	Pydantic model class for the arguments of `function`.
`description`	`ProcessDescription`	Process description modeled after OGC API - Processes - Part 1: Core.

create `classmethod`

create(
    function: Callable,
    id: Optional[str] = None,
    version: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    inputs: Optional[dict[str, FieldInfo | InputDescription]] = None,
    outputs: Optional[dict[str, FieldInfo | OutputDescription]] = None,
    inputs_arg: str | bool = False,
) -> Process

Create a new instance of this dataclass.

Called by the process_registry.main() and by the your_function.step() decorator, where your_function is the function decorated with main().

Not intended to be used by clients.

Parameters:

Name	Type	Description	Default
`function`	`Callable`	The decorated function that is passed automatically since `process()` is a decorator function.	required
`id`	`Optional[str]`	Optional process identifier. Must be unique within the registry. If not provided, the fully qualified function name will be used.	`None`
`version`	`Optional[str]`	Optional version identifier. If not provided, `"0.0.0"` will be used.	`None`
`title`	`Optional[str]`	Optional, short process title.	`None`
`description`	`Optional[str]`	Optional, detailed description of the process. If not provided, the function's docstring, if any, will be used.	`None`
`inputs`	`Optional[dict[str, FieldInfo \| InputDescription]]`	Optional mapping from function argument names to `pydantic.Field` or `InputDescription` instances. The preferred way is to annotate the arguments directly as described in The Annotated Pattern. Use `InputDescription` instances to pass extra information that cannot be represented by a `pydantic.Field`, e.g., `additionalParameters` or `keywords`.	`None`
`outputs`	`Optional[dict[str, FieldInfo \| OutputDescription]]`	Mapping from output names to `pydantic.Field` or `OutputDescription` instances. Required, if you have multiple outputs returned as a dictionary. In this case, the function must return a typed `tuple` and output names refer to the items of the tuple in given order.	`None`
`inputs_arg`	`str \| bool`	Specifies the use of an inputs argument. An inputs argument is a container for the actual process inputs. If specified, it must be the only function argument (besides an optional job context argument) and must be a subclass of `pydantic.BaseModel`. If `inputs_arg` is `True` the only argument will be the input argument, if `inputs_arg` is a `str` it must be the name of the only argument.	`False`

procodile.JobContext

Bases: ABC

Report process progress and check for task cancellation.

A process function can retrieve the current job context

via JobContext.get() from within a process function, or
as a function argument of type JobContext.

get `classmethod`

get() -> JobContext

Get the current job context.

Returns the current job context that can be used by process functions to report job progress in percent or via messages and to check whether cancellation has been requested. This function is intended to be called from within a process function executed as a job. If called as a usual Python function (without a job serving as context), the returned context will have no-op methods only.

Returns:

Type	Description
`JobContext`	An instance of the current job context.

report_progress `abstractmethod`

report_progress(
    progress: Optional[int] = None, message: Optional[str] = None
) -> None

Report task progress.

Parameters:

Name	Type	Description	Default
`progress`	`Optional[int]`	Progress in percent.	`None`
`message`	`Optional[str]`	Detail progress message.	`None`

Raises:

Type	Description
`JobCancellationException`	if an attempt has been made to cancel this job.

is_cancelled `abstractmethod`

is_cancelled() -> bool

Test whether an attempt has been made to cancel this job. It may still be running though.

Returns:

Type	Description
`bool`	`True` if so, `False` otherwise.

check_cancelled `abstractmethod`

check_cancelled() -> None

Raise a JobCancellationException, if an attempt has been made to cancel this job.

procodile.JobCancelledException

Bases: Exception

Raised if a job's cancellation has been requested.

procodile.cli.new_cli

new_cli(
    registry: Union[str, ProcessRegistry, Callable[[], ProcessRegistry]],
    name: str,
    version: str,
    help: str | None = None,
    summary: str | None = None,
    context: dict[str, Any] | None = None,
) -> typer.Typer

Get the CLI instance configured to use the process registry that is given either by

a reference of the form "path.to.module:attribute",
or process registry instance,
or as a no-arg process registry getter function.

The process registry is usually a singleton in your application.

The context object obj of the returned CLI object will be of type dict and will contain a process registry getter function using the key get_process_registry.

The function must be called before any CLI command or callback has been invoked. Otherwise, the provided get_process_registry getter will not be recognized and all commands that require the process registry will fail with an AssertionError.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the CLI application.	required
`registry`	`Union[str, ProcessRegistry, Callable[[], ProcessRegistry]]`	A registry reference string, or a registry instance, or a no-arg function that returns a registry instance.	required
`help`	`str \| None`	Optional CLI application help text. If not provided, the default `cuiman` help text will be used.	`None`
`summary`	`str \| None`	A one-sentence human-readable description of the tool that will be used by the default help text. Hence, used only, if `help` is not provided. Should end with a dot '.'.	`None`
`version`	`str`	Optional version string. If not provided, the `cuiman` version will be used.	required
`context`	`dict[str, Any] \| None`	Additional context values that will be registered with the CLI and can be accessed by commands that you add to the returned `typer.Typer` instance.	`None`

Return

a typer.Typer instance

Procodile API

procodile.ProcessRegistry

main

procodile.Process dataclass

create classmethod

procodile.JobContext

get classmethod

report_progress abstractmethod

is_cancelled abstractmethod

check_cancelled abstractmethod

procodile.JobCancelledException

procodile.cli.new_cli

procodile.Process `dataclass`

create `classmethod`

get `classmethod`

report_progress `abstractmethod`

is_cancelled `abstractmethod`

check_cancelled `abstractmethod`