cloacina-python::trigger_rules

Rust

Python trigger-rule authoring (CLOACI-T-0763): Python parity for Rust’s #[task(trigger_rules = ...)]. These builder pyfunctions return plain JSON-shaped dicts matching the core TriggerRule / TriggerCondition serde representation (execution_planner::trigger_rules), so @cloaca.task( trigger_rules=...) can gate a task and have it land in the real Skipped state — the same evaluator the Rust DSL feeds.

Conditions: context_value, task_success, task_failed, task_skipped. Rules: all_of, any_of, none_of, always. A bare condition passed to trigger_rules= is treated as all_of(that_condition).

Functions

fn parse_​trigger_​rules

pub

fn parse_trigger_rules(
    py: Python,
    obj: &Py<PyAny>,
) -> PyResult<serde_json::Value>

Parse + validate a trigger_rules= argument into the canonical TriggerRule JSON. A bare condition (e.g. a context_value(...)) is wrapped as All. Errors clearly if the value isn’t a valid trigger rule.

Source

pub fn parse_trigger_rules(py: Python, obj: &Py<PyAny>) -> PyResult<serde_json::Value> {
    let mut value: serde_json::Value = depythonize(obj.bind(py)).map_err(|e| {
        PyValueError::new_err(format!(
            "@cloaca.task(trigger_rules=...): not JSON-serializable: {e}"
        ))
    })?;

    // Allow a bare condition for the common single-condition case → wrap as All.
    if let Some(t) = value.get("type").and_then(|v| v.as_str()) {
        if CONDITION_TYPES.contains(&t) {
            value = serde_json::json!({ "type": "All", "conditions": [value] });
        }
    }

    // Validate against the real core type so bad shapes fail at decoration time.
    serde_json::from_value::<cloacina::execution_planner::TriggerRule>(value.clone()).map_err(
        |e| {
            PyValueError::new_err(format!(
                "@cloaca.task(trigger_rules=...): invalid trigger rule: {e}. Build rules with \
                 cloaca.context_value / task_success / task_failed / task_skipped, optionally \
                 inside cloaca.all_of / any_of / none_of.",
            ))
        },
    )?;
    Ok(value)
}

fn context_​value

Binding

Python API: cloaca.context_value

fn context_value(
    py: Python,
    key: String,
    operator: String,
    value: Py<PyAny>,
) -> PyResult<Py<PyAny>>

A context-value condition: context_value("key", "Equals", value).

Source

pub fn context_value(
    py: Python,
    key: String,
    operator: String,
    value: Py<PyAny>,
) -> PyResult<Py<PyAny>> {
    let v: serde_json::Value = depythonize(value.bind(py)).map_err(|e| {
        PyValueError::new_err(format!("context_value: value not JSON-serializable: {e}"))
    })?;
    let op = normalize_operator(&operator)?;
    to_py(
        py,
        serde_json::json!({ "type": "ContextValue", "key": key, "operator": op, "value": v }),
    )
}

fn task_​success

Binding

Python API: cloaca.task_success

fn task_success(py: Python, task_name: String) -> PyResult<Py<PyAny>>

Source

pub fn task_success(py: Python, task_name: String) -> PyResult<Py<PyAny>> {
    to_py(
        py,
        serde_json::json!({ "type": "TaskSuccess", "task_name": task_name }),
    )
}

fn task_​failed

Binding

Python API: cloaca.task_failed

fn task_failed(py: Python, task_name: String) -> PyResult<Py<PyAny>>

Source

pub fn task_failed(py: Python, task_name: String) -> PyResult<Py<PyAny>> {
    to_py(
        py,
        serde_json::json!({ "type": "TaskFailed", "task_name": task_name }),
    )
}

fn task_​skipped

Binding

Python API: cloaca.task_skipped

fn task_skipped(py: Python, task_name: String) -> PyResult<Py<PyAny>>

Source

pub fn task_skipped(py: Python, task_name: String) -> PyResult<Py<PyAny>> {
    to_py(
        py,
        serde_json::json!({ "type": "TaskSkipped", "task_name": task_name }),
    )
}

fn all_​of

Binding

Python API: cloaca.all_of

fn all_of(py: Python, conditions: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>>

Run only if ALL conditions are met.

Source

pub fn all_of(py: Python, conditions: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    combine(py, "All", conditions)
}

fn any_​of

Binding

Python API: cloaca.any_of

fn any_of(py: Python, conditions: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>>

Run if ANY condition is met.

Source

pub fn any_of(py: Python, conditions: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    combine(py, "Any", conditions)
}

fn none_​of

Binding

Python API: cloaca.none_of

fn none_of(
    py: Python,
    conditions: Vec<Py<PyAny>>,
) -> PyResult<Py<PyAny>>

Run only if NONE of the conditions are met.

Source

pub fn none_of(py: Python, conditions: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    combine(py, "None", conditions)
}

fn always

Binding

Python API: cloaca.always

fn always(py: Python) -> PyResult<Py<PyAny>>

The default rule — always run.

Source

pub fn always(py: Python) -> PyResult<Py<PyAny>> {
    to_py(py, serde_json::json!({ "type": "Always" }))
}