Skip to content
Cloacina

Packaged Triggers

In this tutorial, you’ll learn how to define event-driven triggers alongside your Python workflows so they can be packaged and deployed to the Cloacina daemon. While Tutorial 07 introduced triggers running directly in your Python process, this tutorial focuses on the packaging story — how triggers and workflows are declared, bundled, and auto-registered when loaded by the reconciler (the daemon component that discovers, validates, and registers packages).

Learning Objectives

Section titled “Learning Objectives”
  • Define triggers alongside tasks in a packaged workflow
  • Understand that @cloaca.trigger decorators are the declaration — there is no manifest trigger section
  • Package a trigger-bearing workflow as a .cloacina archive
  • See how the reconciler wires triggers to workflows on load

Prerequisites

Section titled “Prerequisites”

Time Estimate

Section titled “Time Estimate”

20-25 minutes

Step 1: Define the Workflow

Section titled “Step 1: Define the Workflow”

Start with a workflow that the trigger will fire. In your entry module (workflow/data_ingest/tasks.py), declare tasks with bare @cloaca.task decorators at module level — in a packaged workflow the loader builds the workflow context from workflow_name in package.toml, so you do not wrap tasks in a WorkflowBuilder (doing so would shadow the loader’s context and the package would load with no tasks):

import cloaca
from datetime import datetime


@cloaca.task(dependencies=[])
def validate(context):
    """Validate the incoming data file."""
    filename = context.get("filename", "unknown")
    print(f"  Validating: {filename}")
    context.set("valid", True)
    return context


@cloaca.task(dependencies=["validate"])
def load(context):
    """Load validated data into the warehouse."""
    filename = context.get("filename", "unknown")
    print(f"  Loading: {filename}")
    context.set("loaded_at", datetime.now().isoformat())
    return context

This is a standard workflow — nothing special about triggers yet.

Step 2: Define the Trigger

Section titled “Step 2: Define the Trigger”

Add a trigger at module level alongside the tasks. Triggers are registered independently from the workflow’s task graph:

@cloaca.trigger(
    name="inbox_watcher",
    on="data_ingest",          # the workflow this trigger fires (its workflow_name)
    poll_interval="5s",
    allow_concurrent=False
)
def inbox_watcher():
    """
    Poll for new files in the inbox directory.


    Returns TriggerResult.fire() with context when a new file
    is detected, or TriggerResult.skip() when nothing is found.
    """
    # In a real trigger, you'd check a filesystem, API, queue, etc.
    import os
    inbox = os.environ.get("INBOX_PATH", "/data/inbox/")


    new_files = [f for f in os.listdir(inbox) if f.endswith(".parquet")]
    if new_files:
        filename = new_files[0]
        ctx = cloaca.Context({
            "filename": filename,
            "trigger_name": "inbox_watcher",
            "triggered_at": datetime.now().isoformat(),
        })
        return cloaca.TriggerResult.fire(ctx)


    return cloaca.TriggerResult.skip()

Notice the parameters: name identifies the trigger, on names the workflow the trigger fires (here data_ingest, matching workflow_name in package.toml), poll_interval controls how often the function is called, and allow_concurrent=False prevents overlapping executions. The on binding is the one to get right — if you omit it, the reconciler falls back to the trigger’s own name (inbox_watcher) as the target workflow, which won’t match data_ingest and nothing will fire. The decorator is the only place these are declared — there is no separate manifest entry to keep in sync.

Step 3: Triggers Are Declared in Code, Not the Manifest

Section titled “Step 3: Triggers Are Declared in Code, Not the Manifest”

There is no trigger section in the package manifest. The @cloaca.trigger decorator is the declaration: when the reconciler imports your entry_module, the decorator registers the trigger (name, poll interval, config), and the reconciler reads the trigger’s on value to persist a schedule row binding it to that workflow (data_ingest). The manifest (package.toml) carries only package identity + [metadata]; adding triggers to it (or a package_type key) is rejected at upload.

So the only thing to get right for triggers is that the @cloaca.trigger decorator runs at import time (module level) — same rule as @cloaca.task.

Step 4: Set Up the Package

Section titled “Step 4: Set Up the Package”

A .cloacina package is a top-level package.toml plus your module tree under workflow/:

data-ingest/
├── package.toml
└── workflow/
    └── data_ingest/
        ├── __init__.py
        └── tasks.py        # @cloaca.task + @cloaca.trigger here

package.toml:

[package]
name = "data-ingest"
version = "1.0.0"
interface = "cloacina-workflow-plugin"
interface_version = 1
extension = "cloacina"


[metadata]
language = "python"
workflow_name = "data_ingest"
entry_module = "data_ingest.tasks"
description = "File ingestion workflow with inbox watcher trigger"
requires_python = ">=3.10"

entry_module is the dotted path relative to workflow/ that the loader imports for task + trigger discovery. See Packaging Python Workflows for the full format and the build/deploy steps.

Step 5: Test Locally

Section titled “Step 5: Test Locally”

Before packaging, test the trigger and workflow in library mode. Because the module uses bare decorators, wrap the import in a WorkflowBuilder named to match workflow_name — this stands in for the context the packaged loader supplies (it is for in-process testing only, not part of the package):

import cloaca


with cloaca.WorkflowBuilder("data_ingest"):
    import data_ingest.tasks   # registers validate/load + inbox_watcher


def test_trigger_and_workflow():
    """Simulate what the daemon does on trigger fire."""
    runner = cloaca.DefaultRunner(":memory:")


    try:
        # Simulate a trigger poll that fires
        result = data_ingest.tasks.inbox_watcher()


        if result.is_fire_result():
            # Execute the workflow with the trigger's context
            context = cloaca.Context({
                "filename": "orders_20260328.parquet",
                "trigger_name": "inbox_watcher",
            })
            wf_result = runner.execute("data_ingest", context)
            print(f"Workflow status: {wf_result.status}")
    finally:
        runner.shutdown()

Step 6: Deploy to the Daemon

Section titled “Step 6: Deploy to the Daemon”

Copy your .cloacina package into the daemon’s watch directory. The reconciler will import your entry_module, which runs your @cloaca.trigger decorator at import time — registering the trigger and starting the poll loop automatically.

Terminal window
cp data-ingest-1.0.0.cloacina ~/.cloacina/packages/

What You Learned

Section titled “What You Learned”
  • @cloaca.trigger is the declaration — it registers the trigger when the module is imported, and its on= names the workflow the trigger fires
  • The manifest (package.toml) carries package identity + [metadata] only; no trigger section
  • Triggers are packaged alongside tasks in the same .cloacina archive
  • The reconciler imports your entry_module on load, which wires everything together

Where to go next

Section titled “Where to go next”

This is an optional, packaging-focused follow-on to 07 — Event Triggers — it takes the in-process trigger you learned there and ships it inside a .cloacina package. It’s the end of the tutorial track.