cloacina-python::trigger

Rust

Python trigger bindings via PyO3.

Provides:

• @cloaca.trigger decorator for defining custom Python triggers
• TriggerResult Python class for returning poll results
• PythonTriggerWrapper implementing the Rust Trigger trait

Structs

struct PythonTriggerDef

pub

A collected Python trigger definition.

Fields

Name	Type	Description
name	String
poll_interval	Duration
allow_concurrent	bool
python_function	Py<PyAny>
workflow_name	String	Workflow this trigger fires, from @cloaca.trigger(on=…). Mirrors the Rust #[trigger(on = "…")] binding (CLOACI-T-0688). Empty for legacy poll triggers that bind via #[workflow(triggers=[…])] subscription.
cron_expression	Option<String>	Cron expression from @cloaca.trigger(cron=…). When set, the reconciler routes this to the cron scheduler instead of the poll registry.
timezone	Option<String>	Timezone for the cron schedule (@cloaca.trigger(timezone=…)).

class TriggerDecorator

Binding

Python API: cloaca.TriggerDecorator

Decorator for defining Python triggers.

@cloaca.trigger(name="check_inbox", poll_interval="30s")
def check_inbox():
    # Return TriggerResult.fire(context) to fire, .skip() otherwise
    return TriggerResult.skip()

Fields

Name	Type	Description
name	Option<String>
poll_interval	Duration
allow_concurrent	bool
workflow_name	String
cron_expression	Option<String>
timezone	Option<String>

Methods

_​_​call_​_​

fn __call__(&self, py: Python, func: Py<PyAny>) -> PyResult<Py<PyAny>>

Python API: cloaca.TriggerDecorator.__call__

Source

    pub fn __call__(&self, py: Python, func: Py<PyAny>) -> PyResult<Py<PyAny>> {
        let trigger_name = if let Some(name) = &self.name {
            name.clone()
        } else {
            func.getattr(py, "__name__")?.extract::<String>(py)?
        };

        let def = PythonTriggerDef {
            name: trigger_name.clone(),
            poll_interval: self.poll_interval,
            allow_concurrent: self.allow_concurrent,
            python_function: func.clone_ref(py),
            workflow_name: self.workflow_name.clone(),
            cron_expression: self.cron_expression.clone(),
            timezone: self.timezone.clone(),
        };

        PYTHON_TRIGGER_REGISTRY.lock().push(def);

        tracing::debug!("Registered Python trigger: {}", trigger_name);

        // Return the original function (decorator is transparent)
        Ok(func)
    }

struct PythonTriggerWrapper

pub

Rust wrapper that implements the Trigger trait by calling a Python function.

Methods

new

pub

fn new(def: &PythonTriggerDef) -> Self

Source

    pub fn new(def: &PythonTriggerDef) -> Self {
        let function = Python::attach(|py| def.python_function.clone_ref(py));
        Self {
            name: def.name.clone(),
            poll_interval: def.poll_interval,
            allow_concurrent: def.allow_concurrent,
            python_function: function,
            workflow_name: def.workflow_name.clone(),
            cron_expression: def.cron_expression.clone(),
        }
    }

Functions

fn drain_​python_​triggers

pub

fn drain_python_triggers() -> Vec<PythonTriggerDef>

Collect all registered Python triggers and clear the registry.

Source

pub fn drain_python_triggers() -> Vec<PythonTriggerDef> {
    let mut registry = PYTHON_TRIGGER_REGISTRY.lock();
    std::mem::take(&mut *registry)
}

fn trigger

Binding

Python API: cloaca.trigger

fn trigger(
    name: Option<String>,
    on: Option<String>,
    poll_interval: Option<String>,
    cron: Option<String>,
    timezone: Option<String>,
    allow_concurrent: bool,
) -> PyResult<TriggerDecorator>

@cloaca.trigger(...) decorator factory.

Mirrors the Rust #[trigger] macro (CLOACI-T-0688). Two modes:

• Custom poll — poll_interval + a function body that returns a TriggerResult.
• Cron — cron (+ optional timezone); the framework schedules the on workflow on that cron expression (the function body is unused). The reconciler routes cron_expression-bearing triggers to the cron scheduler.

on names the workflow the trigger fires; it is required for cron triggers. cron and poll_interval are mutually exclusive.

Source

pub fn trigger(
    name: Option<String>,
    on: Option<String>,
    poll_interval: Option<String>,
    cron: Option<String>,
    timezone: Option<String>,
    allow_concurrent: bool,
) -> PyResult<TriggerDecorator> {
    // Mirror the Rust macro's validation.
    if cron.is_some() && poll_interval.is_some() {
        return Err(PyValueError::new_err(
            "@cloaca.trigger cannot set both 'cron' and 'poll_interval' — pick one",
        ));
    }
    if cron.is_some() && on.is_none() {
        return Err(PyValueError::new_err(
            "@cloaca.trigger(cron=…) requires 'on' (the workflow the cron schedule fires)",
        ));
    }

    // Poll interval still defaults to 30s for the poll path; ignored for cron.
    let interval_str = poll_interval.unwrap_or_else(|| "30s".to_string());
    let interval = parse_duration_str(&interval_str).map_err(|e| {
        PyValueError::new_err(format!("Invalid poll_interval '{}': {}", interval_str, e))
    })?;

    Ok(TriggerDecorator {
        name,
        poll_interval: interval,
        allow_concurrent,
        workflow_name: on.unwrap_or_default(),
        cron_expression: cron,
        timezone,
    })
}