cloacina-python::bindings::trigger

Rust

Python trigger support for event-driven workflow execution.

This module provides Python bindings for defining triggers that poll user-defined conditions and fire workflows when those conditions are met.

Structs

class TriggerResult

Binding

Python API: cloaca.TriggerResult

Python TriggerResult class - represents the result of a trigger poll.

Use TriggerResult.skip() when the condition is not met. Use TriggerResult.fire(context=None) when the condition is met.

Fields

Name	Type	Description
is_fire	bool
data	Option<std::collections::HashMap<String, Value>>

Methods

skip

fn skip() -> Self

Python API: cloaca.TriggerResult.skip

Create a Skip result - condition not met, continue polling.

Source

    fn skip() -> Self {
        PyTriggerResult {
            is_fire: false,
            data: None,
        }
    }

fire

fn fire(context: Option<&PyContext>) -> Self

Python API: cloaca.TriggerResult.fire

Create a Fire result - condition met, trigger the workflow.

Parameters:

Name	Type	Description
context	Option<&PyContext>	Optional context to pass to the workflow

Source

    fn fire(context: Option<&PyContext>) -> Self {
        let data = context.map(|c| c.get_data_clone());
        PyTriggerResult {
            is_fire: true,
            data,
        }
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.TriggerResult.__repr__

Source

    fn __repr__(&self) -> String {
        if !self.is_fire {
            "TriggerResult.Skip".to_string()
        } else if self.data.is_none() {
            "TriggerResult.Fire(None)".to_string()
        } else {
            "TriggerResult.Fire(<context>)".to_string()
        }
    }

is_​fire_​result

fn is_fire_result(&self) -> bool

Python API: cloaca.TriggerResult.is_fire_result

Check if this is a Fire result

Source

    fn is_fire_result(&self) -> bool {
        self.is_fire
    }

is_​skip_​result

fn is_skip_result(&self) -> bool

Python API: cloaca.TriggerResult.is_skip_result

Check if this is a Skip result

Source

    fn is_skip_result(&self) -> bool {
        !self.is_fire
    }

into_​rust

pub

fn into_rust(self) -> TriggerResult

Convert to Rust TriggerResult by consuming self.

Source

    pub fn into_rust(self) -> TriggerResult {
        if !self.is_fire {
            TriggerResult::Skip
        } else {
            let ctx = self.data.map(|d| {
                let mut context = Context::new();
                for (key, value) in d {
                    context.insert(key, value).ok();
                }
                context
            });
            TriggerResult::Fire(ctx)
        }
    }

clone_​into_​rust

pub

fn clone_into_rust(&self) -> TriggerResult

Convert to Rust TriggerResult by cloning self (for &self callers such as the cross-API loader-side PythonTriggerWrapper.poll).

Source

    pub fn clone_into_rust(&self) -> TriggerResult {
        if !self.is_fire {
            TriggerResult::Skip
        } else {
            let ctx = self.data.clone().map(|d| {
                let mut context = Context::new();
                for (key, value) in d {
                    context.insert(key, value).ok();
                }
                context
            });
            TriggerResult::Fire(ctx)
        }
    }

struct PythonTriggerWrapper

pub

Python trigger wrapper implementing Rust Trigger trait.

This struct allows Python functions to be registered and executed as triggers within the Cloacina trigger scheduler.

Methods

workflow_​name

pub

fn workflow_name(&self) -> &str

Get the workflow name this trigger is associated with

Source

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

class TriggerDecorator

Binding

Decorator class that holds trigger configuration

Fields

Name	Type	Description
name	Option<String>
workflow	String
poll_interval	Duration
allow_concurrent	bool

Methods

_​_​call_​_​

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

Source

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

        // Store values for the closure
        let workflow_name = self.workflow.clone();
        let poll_interval = self.poll_interval;
        let allow_concurrent = self.allow_concurrent;
        let name_for_constructor = trigger_name.clone();

        // Create Arc'd function for sharing with constructor
        let shared_function = Arc::new(func.clone_ref(py));

        // Register trigger constructor in the scoped Runtime
        let rt = crate::runtime_scope::current_runtime().ok_or_else(|| {
            PyValueError::new_err("@trigger decorator called outside a Runtime scope")
        })?;
        rt.register_trigger(trigger_name.clone(), move || {
            let function_clone = Python::attach(|py| (*shared_function).clone_ref(py));
            Arc::new(PythonTriggerWrapper {
                name: name_for_constructor.clone(),
                workflow_name: workflow_name.clone(),
                poll_interval,
                allow_concurrent,
                python_function: function_clone,
            }) as Arc<dyn Trigger>
        });

        tracing::info!(
            trigger_name = %trigger_name,
            workflow = %self.workflow,
            poll_interval_ms = %self.poll_interval.as_millis(),
            "Registered Python trigger"
        );

        // Return the original function
        Ok(func)
    }

Functions

fn trigger

Binding

fn trigger(
    workflow: String,
    name: Option<String>,
    poll_interval: &str,
    allow_concurrent: bool,
) -> PyResult<TriggerDecorator>

Python @trigger decorator function

This function is exposed to Python as a decorator that registers Python functions as triggers in the Cloacina trigger scheduler.

Examples:

import cloaca
import random

@cloaca.trigger(
    workflow="my_workflow",
    poll_interval="5s",
    allow_concurrent=False
)
def my_trigger():
    # Check some condition
    if random.randint(1, 100) == 42:
        ctx = cloaca.Context({"triggered_at": "now"})
        return cloaca.TriggerResult.fire(ctx)
    return cloaca.TriggerResult.skip()

Source

pub fn trigger(
    workflow: String,
    name: Option<String>,
    poll_interval: &str,
    allow_concurrent: bool,
) -> PyResult<TriggerDecorator> {
    let duration = parse_duration(poll_interval)
        .map_err(|e| PyValueError::new_err(format!("Invalid poll_interval: {}", e)))?;

    Ok(TriggerDecorator {
        name,
        workflow,
        poll_interval: duration,
        allow_concurrent,
    })
}