Trigger

cloaca.python.trigger Binding

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

Classes

`cloaca.python.trigger.TriggerResult`

Rust Implementation: cloacina::python::trigger::PyTriggerResult

Python-side trigger result returned from poll functions.

Usage from Python:

from cloaca import TriggerResult
@cloaca.trigger(name="my_trigger", poll_interval="10s")
def my_trigger():
if some_condition():
return TriggerResult(should_fire=True, context={"key": "value"})
return TriggerResult(should_fire=False)

Methods

`new`

new(should_fire: bool, context: Optional[Any]) -> Self

Rust Implementation: cloacina::python::trigger::PyTriggerResult::new

Source

    fn new(should_fire: bool, context: Option<PyObject>) -> Self {
        Self {
            should_fire,
            context,
        }
    }

`repr`

__repr__() -> str

Rust Implementation: cloacina::python::trigger::PyTriggerResult::repr

Source

    fn __repr__(&self) -> String {
        if self.should_fire {
            "TriggerResult(should_fire=True)".to_string()
        } else {
            "TriggerResult(should_fire=False)".to_string()
        }
    }

`cloaca.python.trigger.TriggerDecorator`

Rust Implementation: cloacina::python::trigger::TriggerDecorator

Decorator for defining Python triggers.

@cloaca.trigger(name="check_inbox", poll_interval="30s")
def check_inbox():
# Return TriggerResult(should_fire=True, context={...}) to fire
return TriggerResult(should_fire=False)

Methods

`call`

__call__(func: Any) -> Any

Rust Implementation: cloacina::python::trigger::TriggerDecorator::call

Source

    pub fn __call__(&self, py: Python, func: PyObject) -> PyResult<PyObject> {
        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),
        };

        PYTHON_TRIGGER_REGISTRY.lock().push(def);

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

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

Functions

`cloaca.python.trigger.trigger`

trigger(name: Optional[str], poll_interval: str, allow_concurrent: bool) -> TriggerDecorator

Rust Implementation: cloacina::python::trigger::trigger

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

Parameters:

Name	Type	Description
`name`	`Optional[str]`
`poll_interval`	`str`
`allow_concurrent`	`bool`

Source

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

    Ok(TriggerDecorator {
        name,
        poll_interval: interval,
        allow_concurrent,
    })
}