cloacina-python::bindings::runner

Rust

Structs

`class WorkflowResult`

Binding

Python API: cloaca.WorkflowResult

Python wrapper for WorkflowExecutionResult

Fields

Name	Type	Description
`inner`	`cloacina::executor::WorkflowExecutionResult`

Methods

`status`

fn status(&self) -> String

Python API: cloaca.WorkflowResult.status

Source

    pub fn status(&self) -> String {
        format!("{:?}", self.inner.status)
    }

`start_time`

fn start_time(&self) -> String

Python API: cloaca.WorkflowResult.start_time

Source

    pub fn start_time(&self) -> String {
        self.inner.start_time.to_rfc3339()
    }

`end_time`

fn end_time(&self) -> Option<String>

Python API: cloaca.WorkflowResult.end_time

Source

    pub fn end_time(&self) -> Option<String> {
        self.inner.end_time.map(|t| t.to_rfc3339())
    }

`final_context`

fn final_context(&self) -> PyContext

Python API: cloaca.WorkflowResult.final_context

Source

    pub fn final_context(&self) -> PyContext {
        let new_context = self.inner.final_context.clone_data();
        PyContext::from_rust_context(new_context)
    }

`error_message`

fn error_message(&self) -> Option<&str>

Python API: cloaca.WorkflowResult.error_message

Source

    pub fn error_message(&self) -> Option<&str> {
        self.inner.error_message.as_deref()
    }

`repr`

fn __repr__(&self) -> String

Python API: cloaca.WorkflowResult.__repr__

Source

    pub fn __repr__(&self) -> String {
        format!(
            "WorkflowResult(status={}, error={})",
            self.status(),
            self.error_message().unwrap_or("None")
        )
    }

`from_result`

pub

fn from_result(
    result: cloacina::executor::WorkflowExecutionResult,
) -> Self

Source

    pub fn from_result(result: cloacina::executor::WorkflowExecutionResult) -> Self {
        PyWorkflowResult { inner: result }
    }

`class DefaultRunner`

Binding

Python API: cloaca.DefaultRunner

Python wrapper for DefaultRunner

Fields

Name	Type	Description
`runtime_handle`	`Arc<Mutex<AsyncRuntimeHandle>>`

Methods

`new`

fn new(py: Python<'_>, database_url: &str) -> PyResult<Self>

Python API: cloaca.DefaultRunner.new

Create a new DefaultRunner with database connection

Source

    pub fn new(py: Python<'_>, database_url: &str) -> PyResult<Self> {
        let database_url = database_url.to_string();
        // Share the caller's ScopedRuntime (installed by the cloaca pymodule
        // init, and where `@task`/`WorkflowBuilder` register) so the runner's
        // scheduler can find workflows registered before the runner existed.
        let rt_arc = crate::runtime_scope::current_runtime();
        spawn_runtime(py, move |rt| {
            info!(
                "Creating DefaultRunner with database_url: {}",
                cloacina::logging::mask_db_url(&database_url)
            );
            let mut builder = cloacina::DefaultRunner::builder().database_url(&database_url);
            if let Some(arc) = rt_arc {
                builder = builder.runtime_arc(arc);
            }
            rt.block_on(builder.build())
        })
    }

`with_config`

fn with_config(
    py: Python<'_>,
    database_url: &str,
    config: &super::context::PyDefaultRunnerConfig,
) -> PyResult<PyDefaultRunner>

Python API: cloaca.DefaultRunner.with_config

Create a new DefaultRunner with custom configuration

Source

    pub fn with_config(
        py: Python<'_>,
        database_url: &str,
        config: &super::context::PyDefaultRunnerConfig,
    ) -> PyResult<PyDefaultRunner> {
        let database_url = database_url.to_string();
        let rust_config = config.to_rust_config();
        let rt_arc = crate::runtime_scope::current_runtime();
        spawn_runtime(py, move |rt| {
            let mut builder = cloacina::DefaultRunner::builder()
                .database_url(&database_url)
                .with_config(rust_config);
            if let Some(arc) = rt_arc {
                builder = builder.runtime_arc(arc);
            }
            rt.block_on(builder.build())
        })
    }

`with_schema`

fn with_schema(
    py: Python<'_>,
    database_url: &str,
    schema: &str,
) -> PyResult<PyDefaultRunner>

Python API: cloaca.DefaultRunner.with_schema

Create a new DefaultRunner with PostgreSQL schema-based multi-tenancy

Each tenant gets their own schema with independent tables, migrations, and data. Requires a PostgreSQL database.

Parameters:

Name	Type	Description
`database_url`	`&str`	PostgreSQL connection string
`schema`	`&str`	Schema name for tenant isolation (alphanumeric + underscores only)

Source

    pub fn with_schema(
        py: Python<'_>,
        database_url: &str,
        schema: &str,
    ) -> PyResult<PyDefaultRunner> {
        if !database_url.starts_with("postgres://") && !database_url.starts_with("postgresql://") {
            return Err(PyValueError::new_err(
                "Schema-based multi-tenancy requires PostgreSQL. \
                 SQLite does not support database schemas. \
                 Use a PostgreSQL URL like 'postgres://user:pass@host/db'",
            ));
        }
        if schema.is_empty() {
            return Err(PyValueError::new_err("Schema name cannot be empty"));
        }
        if !schema.chars().all(|c| c.is_alphanumeric() || c == '_') {
            return Err(PyValueError::new_err(
                "Schema name must contain only alphanumeric characters and underscores",
            ));
        }

        let database_url = database_url.to_string();
        let schema = schema.to_string();
        let rt_arc = crate::runtime_scope::current_runtime();
        spawn_runtime(py, move |rt| {
            info!(
                "Creating DefaultRunner with schema: {} and database_url: {}",
                schema,
                cloacina::logging::mask_db_url(&database_url)
            );
            let mut builder = cloacina::DefaultRunner::builder()
                .database_url(&database_url)
                .schema(&schema);
            if let Some(arc) = rt_arc {
                builder = builder.runtime_arc(arc);
            }
            rt.block_on(builder.build())
        })
    }

`execute`

fn execute(
    &self,
    workflow_name: &str,
    context: &PyContext,
    py: Python,
) -> PyResult<PyWorkflowResult>

Python API: cloaca.DefaultRunner.execute

Execute a workflow by name with context

Source

    pub fn execute(
        &self,
        workflow_name: &str,
        context: &PyContext,
        py: Python,
    ) -> PyResult<PyWorkflowResult> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::Execute {
            workflow_name: workflow_name.to_string(),
            context: context.clone_inner(),
            response_tx,
        };
        let result = self.send_and_recv(message, response_rx, py, "Workflow execution failed")?;
        Ok(PyWorkflowResult::from_result(result))
    }

`shutdown`

fn shutdown(&self, py: Python) -> PyResult<()>

Python API: cloaca.DefaultRunner.shutdown

Shutdown the runner and cleanup resources

Source

    pub fn shutdown(&self, py: Python) -> PyResult<()> {
        info!("Starting async runtime shutdown from Python");
        let result = py.detach(|| self.runtime_handle.lock().unwrap().shutdown());
        match result {
            Ok(()) => {
                info!("Async runtime shutdown completed successfully");
                Ok(())
            }
            Err(e) => {
                error!("Async runtime shutdown failed: {}", e);
                Err(PyValueError::new_err(format!(
                    "Failed to shutdown async runtime: {}",
                    e
                )))
            }
        }
    }

`register_workflow_instance`

fn register_workflow_instance(
    &self,
    workflow_name: String,
    instance_name: String,
    cron_expression: String,
    timezone: String,
    params: &crate::context::PyContext,
    py: Python,
) -> PyResult<String>

Python API: cloaca.DefaultRunner.register_workflow_instance

Parameters:

Name	Type	Description
`workflow_name`	`String`	Name of the workflow to execute
`cron_expression`	`String`	Standard cron expression (e.g., “0 2 * * *”)
`timezone`	`String`	Timezone for cron interpretation (e.g., “UTC”)

Returns:

Schedule ID as a string Register a NAMED, PARAMETERIZED cron instance (CLOACI-I-0116): params (a dict) is the instance’s fully-resolved bound parameters, merged into the context at every fire under flat top-level keys.

Source

    pub fn register_workflow_instance(
        &self,
        workflow_name: String,
        instance_name: String,
        cron_expression: String,
        timezone: String,
        params: &crate::context::PyContext,
        py: Python,
    ) -> PyResult<String> {
        // A cloaca Context carries the params dict; its JSON form is the
        // fully-resolved instance param object persisted on the schedule.
        let params_json = params.to_json()?;
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::RegisterWorkflowInstance {
            workflow_name,
            instance_name,
            cron_expression,
            timezone,
            params_json,
            response_tx,
        };
        self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to register workflow instance",
        )
    }

`register_cron_workflow`

fn register_cron_workflow(
    &self,
    workflow_name: String,
    cron_expression: String,
    timezone: String,
    py: Python,
) -> PyResult<String>

Python API: cloaca.DefaultRunner.register_cron_workflow

Source

    pub fn register_cron_workflow(
        &self,
        workflow_name: String,
        cron_expression: String,
        timezone: String,
        py: Python,
    ) -> PyResult<String> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::RegisterCronWorkflow {
            workflow_name,
            cron_expression,
            timezone,
            response_tx,
        };
        self.send_and_recv(message, response_rx, py, "Failed to register cron workflow")
    }

`list_cron_schedules`

fn list_cron_schedules(
    &self,
    enabled_only: Option<bool>,
    limit: Option<i64>,
    offset: Option<i64>,
    py: Python,
) -> PyResult<Vec<Py<PyAny>>>

Python API: cloaca.DefaultRunner.list_cron_schedules

List all cron schedules

Parameters:

Name	Type	Description
`enabled_only`	`Option<bool>`	If True, only return enabled schedules
`limit`	`Option<i64>`	Maximum number of schedules to return (default: 100)
`offset`	`Option<i64>`	Number of schedules to skip (default: 0)

Source

    pub fn list_cron_schedules(
        &self,
        enabled_only: Option<bool>,
        limit: Option<i64>,
        offset: Option<i64>,
        py: Python,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::ListCronSchedules {
            enabled_only: enabled_only.unwrap_or(false),
            limit: limit.unwrap_or(100),
            offset: offset.unwrap_or(0),
            response_tx,
        };
        let schedules =
            self.send_and_recv(message, response_rx, py, "Failed to list cron schedules")?;
        schedules
            .into_iter()
            .map(|s| schedule_to_cron_dict(s, py))
            .collect()
    }

`set_cron_schedule_enabled`

fn set_cron_schedule_enabled(
    &self,
    schedule_id: String,
    enabled: bool,
    py: Python,
) -> PyResult<()>

Python API: cloaca.DefaultRunner.set_cron_schedule_enabled

Enable or disable a cron schedule

Source

    pub fn set_cron_schedule_enabled(
        &self,
        schedule_id: String,
        enabled: bool,
        py: Python,
    ) -> PyResult<()> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::SetCronScheduleEnabled {
            schedule_id,
            enabled,
            response_tx,
        };
        self.send_and_recv(message, response_rx, py, "Failed to update cron schedule")
    }

`delete_cron_schedule`

fn delete_cron_schedule(
    &self) -> PyResult<(,
    schedule_id: String) -> PyResult<(,
    py: Python) -> PyResult<(,
) -> PyResult<()>

Python API: cloaca.DefaultRunner.delete_cron_schedule

Delete a cron schedule

Source

    pub fn delete_cron_schedule(&self, schedule_id: String, py: Python) -> PyResult<()> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::DeleteCronSchedule {
            schedule_id,
            response_tx,
        };
        self.send_and_recv(message, response_rx, py, "Failed to delete cron schedule")
    }

`get_cron_schedule`

fn get_cron_schedule(
    &self,
    schedule_id: String,
    py: Python,
) -> PyResult<Py<PyAny>>

Python API: cloaca.DefaultRunner.get_cron_schedule

Get details of a specific cron schedule

Source

    pub fn get_cron_schedule(&self, schedule_id: String, py: Python) -> PyResult<Py<PyAny>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::GetCronSchedule {
            schedule_id,
            response_tx,
        };
        let schedule =
            self.send_and_recv(message, response_rx, py, "Failed to get cron schedule")?;
        schedule_to_cron_dict(schedule, py)
    }

`update_cron_schedule`

fn update_cron_schedule(
    &self,
    schedule_id: String,
    cron_expression: String,
    timezone: String,
    py: Python,
) -> PyResult<()>

Python API: cloaca.DefaultRunner.update_cron_schedule

Update a cron schedule’s expression and timezone

Source

    pub fn update_cron_schedule(
        &self,
        schedule_id: String,
        cron_expression: String,
        timezone: String,
        py: Python,
    ) -> PyResult<()> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::UpdateCronSchedule {
            schedule_id,
            cron_expression,
            timezone,
            response_tx,
        };
        self.send_and_recv(message, response_rx, py, "Failed to update cron schedule")
    }

`get_cron_execution_history`

fn get_cron_execution_history(
    &self,
    schedule_id: String,
    limit: Option<i64>,
    offset: Option<i64>,
    py: Python,
) -> PyResult<Vec<Py<PyAny>>>

Python API: cloaca.DefaultRunner.get_cron_execution_history

Get execution history for a specific cron schedule

Source

    pub fn get_cron_execution_history(
        &self,
        schedule_id: String,
        limit: Option<i64>,
        offset: Option<i64>,
        py: Python,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::GetCronExecutionHistory {
            schedule_id,
            limit: limit.unwrap_or(100),
            offset: offset.unwrap_or(0),
            response_tx,
        };
        let executions = self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to get cron execution history",
        )?;
        executions
            .into_iter()
            .map(|e| cron_execution_to_dict(e, py))
            .collect()
    }

`get_cron_execution_stats`

fn get_cron_execution_stats(
    &self,
    since: String,
    py: Python,
) -> PyResult<Py<PyAny>>

Python API: cloaca.DefaultRunner.get_cron_execution_stats

Get execution statistics for cron schedules

Parameters:

Name	Type	Description
`since`	`String`	Start time for statistics collection (ISO 8601 string)

Source

    pub fn get_cron_execution_stats(&self, since: String, py: Python) -> PyResult<Py<PyAny>> {
        let since_dt = chrono::DateTime::parse_from_rfc3339(&since)
            .map_err(|e| PyValueError::new_err(format!("Invalid datetime format: {}", e)))?
            .with_timezone(&chrono::Utc);

        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::GetCronExecutionStats {
            since: since_dt,
            response_tx,
        };
        let stats = self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to get cron execution stats",
        )?;

        let dict = pyo3::types::PyDict::new(py);
        dict.set_item("total_executions", stats.total_executions)?;
        dict.set_item("successful_executions", stats.successful_executions)?;
        dict.set_item("lost_executions", stats.lost_executions)?;
        dict.set_item("success_rate", stats.success_rate)?;
        Ok(dict.into())
    }

`list_trigger_schedules`

fn list_trigger_schedules(
    &self,
    enabled_only: Option<bool>,
    limit: Option<i64>,
    offset: Option<i64>,
    py: Python,
) -> PyResult<Vec<Py<PyAny>>>

Python API: cloaca.DefaultRunner.list_trigger_schedules

List all trigger schedules

Source

    pub fn list_trigger_schedules(
        &self,
        enabled_only: Option<bool>,
        limit: Option<i64>,
        offset: Option<i64>,
        py: Python,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::ListTriggerSchedules {
            enabled_only: enabled_only.unwrap_or(false),
            limit: limit.unwrap_or(100),
            offset: offset.unwrap_or(0),
            response_tx,
        };
        let schedules =
            self.send_and_recv(message, response_rx, py, "Failed to list trigger schedules")?;
        schedules
            .into_iter()
            .map(|s| schedule_to_trigger_dict(s, py))
            .collect()
    }

`get_trigger_schedule`

fn get_trigger_schedule(
    &self,
    trigger_name: String,
    py: Python,
) -> PyResult<Option<Py<PyAny>>>

Python API: cloaca.DefaultRunner.get_trigger_schedule

Get details of a specific trigger schedule

Source

    pub fn get_trigger_schedule(
        &self,
        trigger_name: String,
        py: Python,
    ) -> PyResult<Option<Py<PyAny>>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::GetTriggerSchedule {
            trigger_name,
            response_tx,
        };
        let schedule_opt =
            self.send_and_recv(message, response_rx, py, "Failed to get trigger schedule")?;
        match schedule_opt {
            Some(schedule) => Ok(Some(schedule_to_trigger_dict(schedule, py)?)),
            None => Ok(None),
        }
    }

`set_trigger_enabled`

fn set_trigger_enabled(
    &self,
    trigger_name: String,
    enabled: bool,
    py: Python,
) -> PyResult<()>

Python API: cloaca.DefaultRunner.set_trigger_enabled

Enable or disable a trigger

Source

    pub fn set_trigger_enabled(
        &self,
        trigger_name: String,
        enabled: bool,
        py: Python,
    ) -> PyResult<()> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::SetTriggerEnabled {
            trigger_name,
            enabled,
            response_tx,
        };
        self.send_and_recv(message, response_rx, py, "Failed to update trigger")
    }

`get_trigger_execution_history`

fn get_trigger_execution_history(
    &self,
    trigger_name: String,
    limit: Option<i64>,
    offset: Option<i64>,
    py: Python,
) -> PyResult<Vec<Py<PyAny>>>

Python API: cloaca.DefaultRunner.get_trigger_execution_history

Get execution history for a specific trigger

Source

    pub fn get_trigger_execution_history(
        &self,
        trigger_name: String,
        limit: Option<i64>,
        offset: Option<i64>,
        py: Python,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::GetTriggerExecutionHistory {
            trigger_name,
            limit: limit.unwrap_or(100),
            offset: offset.unwrap_or(0),
            response_tx,
        };
        let executions = self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to get trigger execution history",
        )?;
        executions
            .into_iter()
            .map(|e| trigger_execution_to_dict(e, py))
            .collect()
    }

`subscribe_workflow_to_reactor`

fn subscribe_workflow_to_reactor(
    &self,
    reactor: String,
    workflow: String,
    tenant: Option<String>,
    when: Option<String>,
    py: Python,
) -> PyResult<String>

Python API: cloaca.DefaultRunner.subscribe_workflow_to_reactor

Subscribe a workflow to a reactor’s firings.

Each future fire of reactor (within tenant) will dispatch workflow via the unified scheduler’s reactor poll tick. The dispatched workflow’s input context carries the boundary cache from the firing.

Idempotent — calling twice with the same (reactor, workflow, tenant) upserts; a later call’s when predicate replaces the earlier one’s.

Parameters:

Name	Type	Description
`reactor`	`String`	The reactor name.
`workflow`	`String`	The workflow to dispatch on each firing.
`tenant`	`Option<String>`	Tenant scope. Defaults to `"public"` if omitted.
`when`	`Option<String>`	Optional CEL filter expression (CLOACI-T-0602). When provided, the scheduler evaluates it against the firing payload and only dispatches when it evaluates to true. The watermark still advances when it evaluates to false. Variables available: `payload` (dict, keys are boundary source names), `reactor` (str), `tenant` (str). Invalid CEL is rejected immediately with a ValueError — no row written.

Examples:

# Unfiltered — fires on every firing.
runner.subscribe_workflow_to_reactor("pricing", "alert")

# Filtered — fires only when the boundary's `quote` source
# carries a price above $100 in us-east.
runner.subscribe_workflow_to_reactor(
    "pricing", "alert",
    when="payload.quote.price > 100 && payload.quote.region == 'us-east'",
)

Source

    pub fn subscribe_workflow_to_reactor(
        &self,
        reactor: String,
        workflow: String,
        tenant: Option<String>,
        when: Option<String>,
        py: Python,
    ) -> PyResult<String> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::SubscribeWorkflowToReactor {
            reactor,
            workflow,
            tenant,
            predicate: when,
            response_tx,
        };
        self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to subscribe workflow to reactor",
        )
    }

`unsubscribe_workflow_from_reactor`

fn unsubscribe_workflow_from_reactor(
    &self,
    reactor: String,
    workflow: String,
    tenant: Option<String>,
    py: Python,
) -> PyResult<bool>

Python API: cloaca.DefaultRunner.unsubscribe_workflow_from_reactor

Remove a workflow-to-reactor subscription.

Returns True if a row was deleted, False if no subscription matched (reactor, workflow, tenant).

Source

    pub fn unsubscribe_workflow_from_reactor(
        &self,
        reactor: String,
        workflow: String,
        tenant: Option<String>,
        py: Python,
    ) -> PyResult<bool> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::UnsubscribeWorkflowFromReactor {
            reactor,
            workflow,
            tenant,
            response_tx,
        };
        self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to unsubscribe workflow from reactor",
        )
    }

`list_reactor_subscriptions`

fn list_reactor_subscriptions(
    &self,
    tenant: Option<String>,
    py: Python,
) -> PyResult<Vec<Py<PyAny>>>

Python API: cloaca.DefaultRunner.list_reactor_subscriptions

List enabled reactor subscriptions for a tenant.

Source

    pub fn list_reactor_subscriptions(
        &self,
        tenant: Option<String>,
        py: Python,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let (response_tx, response_rx) = oneshot::channel();
        let message = RuntimeMessage::ListReactorSubscriptions {
            tenant,
            response_tx,
        };
        let subs = self.send_and_recv(
            message,
            response_rx,
            py,
            "Failed to list reactor subscriptions",
        )?;
        subs.into_iter()
            .map(|s| reactor_subscription_to_dict(s, py))
            .collect()
    }

`repr`

fn __repr__(&self) -> String

Python API: cloaca.DefaultRunner.__repr__

Source

    pub fn __repr__(&self) -> String {
        "DefaultRunner(thread_separated_async_runtime)".to_string()
    }

`enter`

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

Python API: cloaca.DefaultRunner.__enter__

Source

    pub fn __enter__(slf: PyRef<Self>) -> PyRef<Self> {
        slf
    }

`exit`

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

Python API: cloaca.DefaultRunner.__exit__

Source

    pub fn __exit__(
        &self,
        py: Python,
        _exc_type: Option<&Bound<PyAny>>,
        _exc_value: Option<&Bound<PyAny>>,
        _traceback: Option<&Bound<PyAny>>,
    ) -> PyResult<bool> {
        self.shutdown(py)?;
        Ok(false) // Don't suppress exceptions
    }

Enums

`enum ShutdownError`

pub

Errors that can occur during async runtime shutdown

Variants

ChannelClosed - Failed to send shutdown signal to runtime thread
ThreadPanic - Runtime thread panicked during shutdown
Timeout - Shutdown timed out waiting for thread to complete

Functions

`fn _shutdown_all_runners`

Binding

fn _shutdown_all_runners(py: Python) -> PyResult<()>

Shut down every still-live runner. Registered with Python’s atexit by the wheel’s #[pymodule] init; also callable directly (idempotent — shutdown of an already-joined handle is a no-op).

Source

pub fn _shutdown_all_runners(py: Python) -> PyResult<()> {
    let handles: Vec<_> = {
        let mut reg = LIVE_RUNNERS.lock().unwrap();
        reg.drain(..).filter_map(|w| w.upgrade()).collect()
    };
    py.detach(|| {
        for h in &handles {
            if let Err(e) = h.lock().unwrap().shutdown() {
                warn!("atexit runner shutdown reported: {}", e);
            }
        }
    });
    ATEXIT_FIRED.store(true, std::sync::atomic::Ordering::SeqCst);
    Ok(())
}

cloacina-python::bindings::runner

Structs

class WorkflowResult

Fields

Methods

status

start_​time

end_​time

final_​context

error_​message

_​_​repr_​_​

from_​result

class DefaultRunner

Fields

Methods

new

with_​config

with_​schema

execute

shutdown

register_​workflow_​instance

register_​cron_​workflow

list_​cron_​schedules

set_​cron_​schedule_​enabled

delete_​cron_​schedule

get_​cron_​schedule

update_​cron_​schedule

get_​cron_​execution_​history

get_​cron_​execution_​stats

list_​trigger_​schedules

get_​trigger_​schedule

set_​trigger_​enabled

get_​trigger_​execution_​history

subscribe_​workflow_​to_​reactor

unsubscribe_​workflow_​from_​reactor

list_​reactor_​subscriptions

_​_​repr_​_​

_​_​enter_​_​

_​_​exit_​_​

Enums

enum ShutdownError

Variants

Functions

fn _​shutdown_​all_​runners

`class WorkflowResult`

`status`

`start_time`

`end_time`

`final_context`

`error_message`

`repr`

`from_result`

`class DefaultRunner`

`new`

`with_config`

`with_schema`

`execute`

`shutdown`

`register_workflow_instance`

`register_cron_workflow`

`list_cron_schedules`

`set_cron_schedule_enabled`

`delete_cron_schedule`

`get_cron_schedule`

`update_cron_schedule`

`get_cron_execution_history`

`get_cron_execution_stats`

`list_trigger_schedules`

`get_trigger_schedule`

`set_trigger_enabled`

`get_trigger_execution_history`

`subscribe_workflow_to_reactor`

`unsubscribe_workflow_from_reactor`

`list_reactor_subscriptions`

`repr`

`enter`

`exit`

`enum ShutdownError`

`fn _shutdown_all_runners`