cloacina-python::workflow

Rust

Structs

class WorkflowBuilder

Binding

Python API: cloaca.WorkflowBuilder

Python wrapper for WorkflowBuilder

Fields

Name	Type	Description
inner	cloacina::WorkflowBuilder
context	PyWorkflowContext

Methods

new

fn new(
    name: &str,
    tenant: Option<&str>,
    package: Option<&str>,
    workflow: Option<&str>,
) -> Self

Python API: cloaca.WorkflowBuilder.new

Create a new WorkflowBuilder with namespace context

Source

    pub fn new(
        name: &str,
        tenant: Option<&str>,
        package: Option<&str>,
        workflow: Option<&str>,
    ) -> Self {
        let context = PyWorkflowContext::new(
            tenant.unwrap_or("public"),
            package.unwrap_or("embedded"),
            workflow.unwrap_or(name),
        );

        let (tenant_id, _package_name, _workflow_id) = context.as_components();
        let workflow_builder = cloacina::Workflow::builder(name).tenant(tenant_id);

        PyWorkflowBuilder {
            inner: workflow_builder,
            context,
        }
    }

description

fn description(&mut self, description: &str)

Python API: cloaca.WorkflowBuilder.description

Set the workflow description

Source

    pub fn description(&mut self, description: &str) {
        self.inner = self.inner.clone().description(description);
    }

tag

fn tag(&mut self, key: &str, value: &str)

Python API: cloaca.WorkflowBuilder.tag

Add a tag to the workflow

Source

    pub fn tag(&mut self, key: &str, value: &str) {
        self.inner = self.inner.clone().tag(key, value);
    }

add_​task

fn add_task(&mut self, py: Python, task: Py<PyAny>) -> PyResult<()>

Python API: cloaca.WorkflowBuilder.add_task

Add a task to the workflow by ID or function reference

Source

    pub fn add_task(&mut self, py: Python, task: Py<PyAny>) -> PyResult<()> {
        match task.extract::<String>(py) {
            Ok(task_id) => {
                let rt = crate::runtime_scope::current_runtime().ok_or_else(|| {
                    PyValueError::new_err("WorkflowBuilder.add_task called outside a Runtime scope")
                })?;

                let (tenant_id, package_name, workflow_id) = self.context.as_components();
                let task_namespace =
                    cloacina::TaskNamespace::new(tenant_id, package_name, workflow_id, &task_id);

                let task_instance = rt.get_task(&task_namespace).ok_or_else(|| {
                    PyValueError::new_err(format!(
                        "Task '{}' not found in registry. Make sure it was decorated with @task.",
                        task_id
                    ))
                })?;

                self.inner = self
                    .inner
                    .clone()
                    .add_task(task_instance)
                    .map_err(|e| PyValueError::new_err(format!("Failed to add task: {}", e)))?;

                Ok(())
            }
            _ => match task.bind(py).hasattr("__name__") {
                Ok(true) => match task.getattr(py, "__name__") {
                    Ok(name_obj) => match name_obj.extract::<String>(py) {
                        Ok(func_name) => {
                            let rt = crate::runtime_scope::current_runtime().ok_or_else(|| {
                                PyValueError::new_err(
                                    "WorkflowBuilder.add_task called outside a Runtime scope",
                                )
                            })?;

                            let (tenant_id, package_name, workflow_id) =
                                self.context.as_components();
                            let task_namespace = cloacina::TaskNamespace::new(
                                tenant_id,
                                package_name,
                                workflow_id,
                                &func_name,
                            );

                            let task_instance = rt.get_task(&task_namespace).ok_or_else(|| {
                                        PyValueError::new_err(format!(
                                            "Task '{}' not found in registry. Make sure it was decorated with @task.",
                                            func_name
                                        ))
                                    })?;

                            self.inner =
                                self.inner.clone().add_task(task_instance).map_err(|e| {
                                    PyValueError::new_err(format!("Failed to add task: {}", e))
                                })?;

                            Ok(())
                        }
                        Err(e) => Err(PyValueError::new_err(format!(
                            "Function has __name__ but it's not a string: {}",
                            e
                        ))),
                    },
                    Err(e) => Err(PyValueError::new_err(format!(
                        "Failed to get __name__ from function: {}",
                        e
                    ))),
                },
                Ok(false) => Err(PyValueError::new_err(
                    "Task must be either a string task ID or a function object with __name__ attribute",
                )),
                Err(e) => Err(PyValueError::new_err(format!(
                    "Failed to check if object has __name__ attribute: {}",
                    e
                ))),
            },
        }
    }

build

fn build(&self) -> PyResult<PyWorkflow>

Python API: cloaca.WorkflowBuilder.build

Build the workflow

Source

    pub fn build(&self) -> PyResult<PyWorkflow> {
        let workflow = self
            .inner
            .clone()
            .build()
            .map_err(|e| PyValueError::new_err(format!("Failed to build workflow: {}", e)))?;
        Ok(PyWorkflow { inner: workflow })
    }

_​_​enter_​_​

fn __enter__(slf: PyRef<Self>) -> PyRef<Self>

Python API: cloaca.WorkflowBuilder.__enter__

Context manager entry - establish workflow context for task decorators

Source

    pub fn __enter__(slf: PyRef<Self>) -> PyRef<Self> {
        push_workflow_context(slf.context.clone());
        slf
    }

_​_​exit_​_​

fn __exit__(
    &mut self,
    _py: Python,
    _exc_type: Option<&Bound<PyAny>>,
    _exc_value: Option<&Bound<PyAny>>,
    _traceback: Option<&Bound<PyAny>>,
) -> PyResult<bool>

Python API: cloaca.WorkflowBuilder.__exit__

Context manager exit - clean up context and build workflow

Source

    pub fn __exit__(
        &mut self,
        _py: Python,
        _exc_type: Option<&Bound<PyAny>>,
        _exc_value: Option<&Bound<PyAny>>,
        _traceback: Option<&Bound<PyAny>>,
    ) -> PyResult<bool> {
        pop_workflow_context();

        let (tenant_id, package_name, workflow_id) = self.context.as_components();

        let mut workflow = cloacina::Workflow::new(workflow_id);
        workflow.set_tenant(tenant_id);
        workflow.set_package(package_name);

        // Preserve description and tags set on the builder during the `with` block
        if let Some(desc) = self.inner.get_description() {
            workflow.set_description(desc);
        }
        for (key, value) in self.inner.get_tags() {
            workflow.add_tag(key, value);
        }

        let rt = crate::runtime_scope::current_runtime().ok_or_else(|| {
            PyValueError::new_err("WorkflowBuilder.__exit__ called outside a Runtime scope")
        })?;
        for namespace in rt.task_namespaces() {
            if namespace.tenant_id == tenant_id
                && namespace.package_name == package_name
                && namespace.workflow_id == workflow_id
            {
                if let Some(task_instance) = rt.get_task(&namespace) {
                    workflow
                        .add_task(task_instance)
                        .map_err(|e| PyValueError::new_err(format!("Failed to add task: {}", e)))?;
                }
            }
        }

        workflow
            .validate()
            .map_err(|e| PyValueError::new_err(format!("Workflow validation failed: {}", e)))?;
        let final_workflow = workflow.finalize();

        let workflow_name = final_workflow.name().to_string();
        rt.register_workflow(workflow_name, move || final_workflow.clone());

        Ok(false)
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.WorkflowBuilder.__repr__

String representation

Source

    pub fn __repr__(&self) -> String {
        format!("WorkflowBuilder(name='{}')", self.inner.name())
    }

class Workflow

Binding

Python API: cloaca.Workflow

Python wrapper for Workflow

Fields

Name	Type	Description
inner	cloacina::Workflow

Methods

name

fn name(&self) -> &str

Python API: cloaca.Workflow.name

Get workflow name

Source

    pub fn name(&self) -> &str {
        self.inner.name()
    }

description

fn description(&self) -> String

Python API: cloaca.Workflow.description

Get workflow description

Source

    pub fn description(&self) -> String {
        self.inner
            .metadata()
            .description
            .clone()
            .unwrap_or_default()
    }

version

fn version(&self) -> &str

Python API: cloaca.Workflow.version

Get workflow version

Source

    pub fn version(&self) -> &str {
        &self.inner.metadata().version
    }

topological_​sort

fn topological_sort(&self) -> PyResult<Vec<String>>

Python API: cloaca.Workflow.topological_sort

Get topological sort of tasks

Source

    pub fn topological_sort(&self) -> PyResult<Vec<String>> {
        self.inner
            .topological_sort()
            .map(|namespaces| namespaces.into_iter().map(|ns| ns.to_string()).collect())
            .map_err(|e| PyValueError::new_err(format!("Failed to sort tasks: {}", e)))
    }

get_​execution_​levels

fn get_execution_levels(&self) -> PyResult<Vec<Vec<String>>>

Python API: cloaca.Workflow.get_execution_levels

Get execution levels (tasks that can run in parallel)

Source

    pub fn get_execution_levels(&self) -> PyResult<Vec<Vec<String>>> {
        self.inner
            .get_execution_levels()
            .map(|levels| {
                levels
                    .into_iter()
                    .map(|level| level.into_iter().map(|ns| ns.to_string()).collect())
                    .collect()
            })
            .map_err(|e| PyValueError::new_err(format!("Failed to get execution levels: {}", e)))
    }

get_​roots

fn get_roots(&self) -> Vec<String>

Python API: cloaca.Workflow.get_roots

Get root tasks (no dependencies)

Source

    pub fn get_roots(&self) -> Vec<String> {
        self.inner
            .get_roots()
            .into_iter()
            .map(|ns| ns.to_string())
            .collect()
    }

get_​leaves

fn get_leaves(&self) -> Vec<String>

Python API: cloaca.Workflow.get_leaves

Get leaf tasks (no dependents)

Source

    pub fn get_leaves(&self) -> Vec<String> {
        self.inner
            .get_leaves()
            .into_iter()
            .map(|ns| ns.to_string())
            .collect()
    }

validate

fn validate(&self) -> PyResult<()>

Python API: cloaca.Workflow.validate

Validate the workflow

Source

    pub fn validate(&self) -> PyResult<()> {
        self.inner
            .validate()
            .map_err(|e| PyValueError::new_err(format!("Workflow validation failed: {}", e)))
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.Workflow.__repr__

String representation

Source

    pub fn __repr__(&self) -> String {
        format!(
            "Workflow(name='{}', tasks={})",
            self.inner.name(),
            self.inner.get_task_ids().len()
        )
    }

class WorkflowParamsDecorator

Binding

Python API: cloaca.WorkflowParamsDecorator

Pass-through decorator returned by workflow_params. CLOACI-T-0760: the declaration is parsed from source by the compiler at build time, so at runtime this just returns the decorated object unchanged. It exists only so that importing a package which uses @cloaca.workflow_params(...) doesn’t fail (the reconciler imports the module via PyO3).

Methods

_​_​call_​_​

fn __call__(&self, obj: Py<PyAny>) -> Py<PyAny>

Python API: cloaca.WorkflowParamsDecorator.__call__

Source

    fn __call__(&self, obj: Py<PyAny>) -> Py<PyAny> {
        obj
    }

class WorkflowSecretsDecorator

Binding

Python API: cloaca.WorkflowSecretsDecorator

Pass-through decorator returned by workflow_secrets. CLOACI-T-0859: like PyWorkflowParamsDecorator, the declaration is parsed from source by the compiler at build time into encrypted InputSlots; at runtime it is a no-op pass-through so importing a package that uses @cloaca.workflow_secrets(...) doesn’t fail.

Methods

_​_​call_​_​

fn __call__(&self, obj: Py<PyAny>) -> Py<PyAny>

Python API: cloaca.WorkflowSecretsDecorator.__call__

Source

    fn __call__(&self, obj: Py<PyAny>) -> Py<PyAny> {
        obj
    }

class BoundarySchemaDecorator

Binding

Pass-through decorator returned by boundary_schema. CLOACI-T-0770: like PyWorkflowParamsDecorator, the declaration is parsed from source by the compiler at build time; at runtime this just returns the decorated object.

Methods

_​_​call_​_​

fn __call__(&self, obj: Py<PyAny>) -> Py<PyAny>

Source

    fn __call__(&self, obj: Py<PyAny>) -> Py<PyAny> {
        obj
    }

Functions

fn py_​register_​workflow

Binding

Python API: cloaca.register_workflow_constructor

fn py_register_workflow(
    name: String) -> PyResult<(,
    constructor: Py<PyAny>) -> PyResult<(,
) -> PyResult<()>

Register a workflow constructor function on the thread-local Runtime scope. Exported to Python as register_workflow_constructor for backwards compatibility; the Rust symbol name is py_register_workflow so it does not collide with the (now deleted) cloacina::register_workflow_constructor.

Source

pub fn py_register_workflow(name: String, constructor: Py<PyAny>) -> PyResult<()> {
    Python::attach(|py| {
        let workflow_obj = constructor.call0(py).map_err(|e| {
            PyValueError::new_err(format!("Failed to call workflow constructor: {}", e))
        })?;

        let py_workflow: PyWorkflow = workflow_obj.extract(py).map_err(|e| {
            PyValueError::new_err(format!(
                "Failed to extract workflow from constructor: {}",
                e
            ))
        })?;

        let workflow = py_workflow.inner.clone();
        let rt = crate::runtime_scope::current_runtime().ok_or_else(|| {
            PyValueError::new_err("register_workflow_constructor called outside a Runtime scope")
        })?;
        rt.register_workflow(name, move || workflow.clone());

        Ok(())
    })
}

fn workflow_​params

Binding

Python API: cloaca.workflow_params

fn workflow_params(
    _kwargs: Option<Bound<'_, pyo3::types::PyDict>>,
) -> PyWorkflowParamsDecorator

@cloaca.workflow_params(name=type, name=(type, default), ...) — declares a packaged Python workflow’s typed input params (CLOACI-I-0128 / T-0760).

Parity with Rust’s #[workflow(params(...))]. The authoritative extraction happens at build time (the compiler parses this decorator from source into JSON-Schema-typed InputSlots); at runtime it is a no-op pass-through so the decorated function/workflow is unaffected. Supported scalar types: str, int, float, bool, list, dict.

Source

pub fn workflow_params(
    _kwargs: Option<Bound<'_, pyo3::types::PyDict>>,
) -> PyWorkflowParamsDecorator {
    PyWorkflowParamsDecorator
}

fn workflow_​secrets

Binding

Python API: cloaca.workflow_secrets

fn workflow_secrets(
    _args: Bound<'_, pyo3::types::PyTuple>,
) -> PyWorkflowSecretsDecorator

@cloaca.workflow_secrets("db_prod", "api_token", ...) — declares the SECRETS a packaged Python workflow requires (CLOACI-I-0133 / T-0859), the parity of Rust’s #[workflow(secrets(...))]. Each name becomes a required encrypted InputSlot (encrypted: true) carried in the package manifest alongside the declared params; a secret is bound per-instance via a {"$secret": "name"} reference and resolved encrypted at fire time (its value never enters the durable Context). The authoritative extraction happens at build time (the compiler parses this decorator from source); at runtime it is a no-op pass-through.

Source

pub fn workflow_secrets(_args: Bound<'_, pyo3::types::PyTuple>) -> PyWorkflowSecretsDecorator {
    PyWorkflowSecretsDecorator
}

fn boundary_​schema

Binding

Python API: cloaca.boundary_schema

fn boundary_schema(
    _kwargs: Option<Bound<'_, pyo3::types::PyDict>>,
) -> PyBoundarySchemaDecorator

@cloaca.boundary_schema(field=type, ...) — declares the typed shape of a Python accumulator’s boundary (CLOACI-T-0770), the parity of deriving schemars::JsonSchema on a Rust boundary type. The compiler parses this from source at build time into the accumulator’s declared input-interface slot (so inject/fire-with render typed forms); at runtime it’s a no-op pass-through. Applied to the accumulator function:

@cloaca.boundary_schema(bid=float, ask=float)
@cloaca.passthrough_accumulator
def py_alpha(event): ...

Supported scalar types: str, int, float, bool, list, dict.

Source

pub fn boundary_schema(
    _kwargs: Option<Bound<'_, pyo3::types::PyDict>>,
) -> PyBoundarySchemaDecorator {
    PyBoundarySchemaDecorator
}