Triggers¶

Triggers monitor external conditions (S3 uploads, database changes, asset updates) and automatically execute callbacks to update your app's page when conditions are met.

Overview¶

The trigger system integrates with APScheduler to periodically check conditions. When a trigger fires, it runs a page_update_callback to refresh the app's UI — no manual user action required.

Triggers are used with SingletonApp for live dashboards and monitoring scenarios.

Trigger Base Class¶

Trigger ¶

Trigger(name: str, app_id: Optional[str] = None, flow_id: Optional[str] = None, user_id: Optional[str] = None, callback: Optional[PageUpdateCallback] = None, organization_id: Optional[str] = None, trigger_metadata: Optional[Dict[str, Any]] = None, trigger_id: Optional[str] = None, max_retries: int = 3, step_name: Optional[str] = None, dry_run: bool = False, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: bool = True)

Base class for all triggers

Triggers define conditions that, when met, execute a callback to update an app's page. Examples include S3 object changes, database record updates, or custom conditions.

Triggers persist to the database and run as background arq jobs in the worker.

Initialize a trigger instance.

Runtime parameters (app_id, flow_id, user_id, callback, step_name) can be provided at initialization or later via the configure() method.

All triggers are checked by APScheduler at a configured interval (default: 10s).

Parameters:

name (str) –

Human-readable trigger name
app_id (Optional[str], default: None ) –

ID of the app this trigger belongs to (optional, set at activation)
flow_id (Optional[str], default: None ) –

ID of the flow instance (optional, set at activation)
user_id (Optional[str], default: None ) –

ID of the user who owns this trigger (optional, set at activation)
callback (Optional[PageUpdateCallback], default: None ) –

PageUpdateCallback to execute when trigger conditions are met (optional, set at activation)
organization_id (Optional[str], default: None ) –

Organization ID for multi-tenancy (optional, set at activation)
trigger_metadata (Optional[Dict[str, Any]], default: None ) –

Additional trigger-specific configuration
trigger_id (Optional[str], default: None ) –

Unique trigger ID (auto-generated if not provided)
max_retries (int, default: 3 ) –

Maximum number of retry attempts on failure
step_name (Optional[str], default: None ) –

Step name to associate with trigger (optional, defaults to first step at activation)
dry_run (bool, default: False ) –

If True, log what would execute without actually executing callbacks
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to the callback (will be pickled and stored)
pass_store_interface (bool, default: True ) –

If True (default), pass store_interface to callback. If False, only pass callback_kwargs

id `property` ¶

id: str

Get the unique trigger ID

name `property` ¶

name: str

Get the trigger name

is_running `property` ¶

is_running: bool

Check if the trigger is currently active

should_retry `property` ¶

should_retry: bool

Check if the trigger should retry after a failure

configure ¶

configure(app_id: str, flow_id: str, user_id: str, callback: PageUpdateCallback, organization_id: Optional[str] = None, trigger_id: Optional[str] = None, step_name: Optional[str] = None, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: Optional[bool] = None) -> None

Configure runtime parameters for the trigger.

This method should be called when activating an app to set the app_id, flow_id, user_id, callback, and step_name that are only known at runtime.

Parameters:

app_id (str) –

ID of the app this trigger belongs to
flow_id (str) –

ID of the flow instance
user_id (str) –

ID of the user who owns this trigger
callback (PageUpdateCallback) –

PageUpdateCallback to execute when trigger conditions are met
organization_id (Optional[str], default: None ) –

Organization ID for multi-tenancy
trigger_id (Optional[str], default: None ) –

Optional trigger ID to override the auto-generated one
step_name (Optional[str], default: None ) –

Step name to associate with trigger (optional)
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to the callback
pass_store_interface (Optional[bool], default: None ) –

If True, pass store_interface to callback (overrides initialization value)

is_configured ¶

is_configured() -> bool

Check if the trigger has been configured with runtime parameters.

Returns:

bool –

True if all required runtime parameters are set

check `abstractmethod` `async` ¶

check() -> bool

Check if the trigger condition has been met.

This method should be implemented by subclasses to define the specific condition logic (e.g., check if S3 object exists, check DB value, etc.)

Returns:

bool –

True if the trigger condition is met, False otherwise

execute `async` ¶

execute() -> None

Execute the trigger callback with proper error handling.

In dry-run mode, logs what would execute without calling the callback. Calls the page update callback with the appropriate context and handles any errors that occur during execution.

start `async` ¶

start() -> None

Start the trigger, marking it as active.

This is called when the trigger is first created or reactivated. Subclasses can override to perform additional setup.

stop `async` ¶

stop() -> None

Stop the trigger gracefully.

This is called when the trigger should be deactivated. Subclasses can override to perform cleanup.

get_state ¶

get_state() -> Optional[Dict[str, Any]]

Get the current trigger state.

Returns:

Optional[Dict[str, Any]] –

Dictionary containing the last known state, or None if no state

set_state ¶

set_state(state: Dict[str, Any]) -> None

Set the trigger state.

This is used to track changes between checks (e.g., last S3 ETag, last DB timestamp).

Parameters:

state (Dict[str, Any]) –

Dictionary containing state to track

get_metrics ¶

get_metrics() -> Dict[str, Any]

Get trigger execution metrics.

Returns:

Dict[str, Any] –

Dictionary containing execution statistics

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize trigger to a dictionary for database persistence.

Returns:

Dict[str, Any] –

Dictionary representation of the trigger

S3 Trigger¶

Monitor an S3 bucket for new or modified objects.

S3Trigger ¶

S3Trigger(name: str, bucket: Optional[str] = None, key: Optional[str] = None, prefix: Optional[str] = None, mode: S3TriggerMode = S3TriggerMode.EXISTS, app_id: Optional[str] = None, flow_id: Optional[str] = None, user_id: Optional[str] = None, callback: Optional[PageUpdateCallback] = None, region_name: Optional[str] = None, endpoint_url: Optional[str] = None, organization_id: Optional[str] = None, trigger_metadata: Optional[Dict[str, Any]] = None, trigger_id: Optional[str] = None, max_retries: int = 3, step_name: Optional[str] = None, dry_run: bool = False, min_size: Optional[int] = None, max_size: Optional[int] = None, file_pattern: Optional[str] = None, content_type: Optional[str] = None, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: bool = True)

Trigger that monitors S3 objects for existence or content changes.

NOTE: All triggers are checked by APScheduler every N seconds (configurable via TRIGGER_CHECK_INTERVAL env var, default 10 seconds). When a trigger's condition is met, an ARQ job is enqueued to execute the callback.

Uses shared connection pooling across all S3Trigger instances for better performance.

Examples: # Trigger when file exists trigger = S3Trigger( name="data_ready", bucket="my-bucket", key="data/output.csv", mode=S3TriggerMode.EXISTS, callback=on_data_ready, )

# Trigger when file content changes
trigger = S3Trigger(
    name="config_updated",
    bucket="my-bucket",
    key="config/settings.json",
    mode=S3TriggerMode.CHANGED,
    callback=on_config_updated,
)

# Trigger when new files appear in prefix
trigger = S3Trigger(
    name="new_uploads",
    bucket="my-bucket",
    prefix="uploads/",
    mode=S3TriggerMode.NEW,
    callback=on_new_upload,
)

Initialize S3 trigger.

NOTE: All triggers are checked by APScheduler every N seconds (default 10s, configurable via TRIGGER_CHECK_INTERVAL env var).

Runtime parameters (app_id, flow_id, user_id, callback, step_name) can be provided at initialization or later via the configure() method.

Parameters:

name (str) –

Trigger name
bucket (Optional[str], default: None ) –

S3 bucket name
key (Optional[str], default: None ) –

S3 object key (for EXISTS/CHANGED modes)
prefix (Optional[str], default: None ) –

S3 prefix (for NEW mode)
mode (S3TriggerMode, default: EXISTS ) –

Trigger mode (EXISTS, CHANGED, or NEW)
app_id (Optional[str], default: None ) –

App ID (optional, set at activation)
flow_id (Optional[str], default: None ) –

Flow ID (optional, set at activation)
user_id (Optional[str], default: None ) –

User ID (optional, set at activation)
callback (Optional[PageUpdateCallback], default: None ) –

Callback to execute (optional, set at activation)
region_name (Optional[str], default: None ) –

AWS region (defaults to config)
endpoint_url (Optional[str], default: None ) –

S3 endpoint URL (defaults to config)
organization_id (Optional[str], default: None ) –

Organization ID (optional, set at activation)
trigger_metadata (Optional[Dict[str, Any]], default: None ) –

Additional metadata
trigger_id (Optional[str], default: None ) –

Unique trigger ID
max_retries (int, default: 3 ) –

Max retry attempts
step_name (Optional[str], default: None ) –

Step name to associate with trigger (optional)
dry_run (bool, default: False ) –

If True, log what would execute without actually executing callbacks
min_size (Optional[int], default: None ) –

Minimum file size in bytes (filter for EXISTS/CHANGED modes)
max_size (Optional[int], default: None ) –

Maximum file size in bytes (filter for EXISTS/CHANGED modes)
file_pattern (Optional[str], default: None ) –

Regex pattern for filename filtering
content_type (Optional[str], default: None ) –

MIME type filter (e.g., "application/json", "text/csv")
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to callback (will be pickled)
pass_store_interface (bool, default: True ) –

If True (default), pass store_interface to callback

check `async` ¶

check() -> bool

Check if trigger condition is met.

Returns:

bool –

True if condition met, False otherwise

stop `async` ¶

stop() -> None

Stop the trigger and cleanup resources.

Closes the aioboto3 session to prevent resource leaks.

get_shared_session `async` `classmethod` ¶

get_shared_session() -> aioboto3.Session

Get or create shared S3 session with connection pooling.

This session is shared across all S3Trigger instances to reduce connection overhead and improve performance.

Returns:

Session –

Shared aioboto3 session

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize trigger to dictionary.

Overrides base to include S3-specific fields.

Returns:

Dict[str, Any] –

Dictionary representation

S3TriggerMode ¶

Modes for S3 trigger behavior.

EXISTS: Trigger when object exists at path CHANGED: Trigger when object content changes (uses ETag) NEW: Trigger when new objects appear in prefix

from virtualitics_sdk.triggers.s3_trigger import S3Trigger, S3TriggerMode
from virtualitics_sdk.types.callbacks import page_update_callback

@page_update_callback
async def on_new_file(store_interface):
    page = await store_interface.get_page()
    status = page.get_element_by_id("status_text")
    status.content = "New file detected — refreshing data..."
    # Load and display the new data
    table = page.get_element_by_id("data_table")
    table.data = load_latest_data()

trigger = S3Trigger(
    name="Watch uploads",
    bucket="my-bucket",
    prefix="data/uploads/",
    mode=S3TriggerMode.NEW_OBJECTS,
    callback=on_new_file
)

PostgreSQL Trigger¶

Monitor a PostgreSQL table for new or modified rows.

PostgresTrigger ¶

PostgresTrigger(name: str, query: Select, mode: PostgresTriggerMode = PostgresTriggerMode.EXISTS, connection_id: Optional[str] = None, use_store_uri: bool = False, count_threshold: Optional[int] = None, app_id: Optional[str] = None, flow_id: Optional[str] = None, user_id: Optional[str] = None, callback: Optional[PageUpdateCallback] = None, organization_id: Optional[str] = None, trigger_metadata: Optional[Dict[str, Any]] = None, trigger_id: Optional[str] = None, max_retries: int = 3, step_name: Optional[str] = None, dry_run: bool = False, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: bool = True, query_timeout: int = 30)

Trigger that monitors PostgreSQL database records and executes callbacks when conditions are met.

NOTE: All triggers are checked by APScheduler every N seconds (configurable via TRIGGER_CHECK_INTERVAL env var, default 10 seconds). When a trigger's condition is met, an ARQ job is enqueued to execute the callback.

Supports both internal store database (predict_configs.store_engine_uri) and external PostgreSQL databases (via connection_store connection_id).

Security features: - Only accepts SQLAlchemy Select objects (prevents SQL injection) - Uses connection pooling to prevent resource exhaustion - Leverages connection_store encryption for credentials - Implements query timeouts to prevent long-running queries

Examples: # Trigger when active users exist from sqlalchemy import select, table, column

users_table = table('users', column('id'), column('status'))
query = select(users_table).where(users_table.c.status == 'active')

trigger = PostgresTrigger(
    name="active_users_monitor",
    query=query,
    mode=PostgresTriggerMode.EXISTS,
    connection_id="my_postgres_connection",
    callback=my_callback,
)

# Trigger when data changes
trigger = PostgresTrigger(
    name="data_change_monitor",
    query=select(my_table),
    mode=PostgresTriggerMode.CHANGED,
    use_store_uri=True,
    callback=on_data_changed,
)

# Trigger when new records appear
query = select(events_table.c.id).order_by(events_table.c.created_at)
trigger = PostgresTrigger(
    name="new_events_monitor",
    query=query,
    mode=PostgresTriggerMode.NEW,
    connection_id="events_db",
    callback=process_new_events,
)

# Trigger when count threshold met
query = select(alerts_table)
trigger = PostgresTrigger(
    name="alert_threshold_monitor",
    query=query,
    mode=PostgresTriggerMode.COUNT,
    count_threshold=10,  # Trigger when >= 10 alerts
    use_store_uri=True,
    callback=handle_alert_threshold,
)

Initialize PostgreSQL trigger.

NOTE: All triggers are checked by APScheduler every N seconds (default 10s, configurable via TRIGGER_CHECK_INTERVAL env var).

Runtime parameters (app_id, flow_id, user_id, callback, step_name) can be provided at initialization or later via the configure() method.

Parameters:

name (str) –

Trigger name
query (Select) –

SQLAlchemy Select statement to execute
mode (PostgresTriggerMode, default: EXISTS ) –

Trigger mode (EXISTS, CHANGED, NEW, or COUNT)
connection_id (Optional[str], default: None ) –

Connection ID from connection_store (optional)
use_store_uri (bool, default: False ) –

If True, use predict_configs.store_engine_uri (default False)
count_threshold (Optional[int], default: None ) –

Threshold for COUNT mode (required if mode=COUNT)
app_id (Optional[str], default: None ) –

App ID (optional, set at activation)
flow_id (Optional[str], default: None ) –

Flow ID (optional, set at activation)
user_id (Optional[str], default: None ) –

User ID (optional, set at activation)
callback (Optional[PageUpdateCallback], default: None ) –

Callback to execute (optional, set at activation)
organization_id (Optional[str], default: None ) –

Organization ID (optional, set at activation)
trigger_metadata (Optional[Dict[str, Any]], default: None ) –

Additional metadata
trigger_id (Optional[str], default: None ) –

Unique trigger ID
max_retries (int, default: 3 ) –

Max retry attempts
step_name (Optional[str], default: None ) –

Step name to associate with trigger (optional)
dry_run (bool, default: False ) –

If True, log what would execute without actually executing callbacks
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to callback (will be pickled)
pass_store_interface (bool, default: True ) –

If True (default), pass store_interface to callback
query_timeout (int, default: 30 ) –

Query execution timeout in seconds (default 30)

check `async` ¶

check() -> bool

Check if trigger condition is met.

Returns:

bool –

True if condition met, False otherwise

execute `async` ¶

execute() -> None

Execute the callback with matched records.

Overrides parent execute() to pass matched records as a kwarg 'records'.

stop `async` ¶

stop() -> None

Stop the trigger and cleanup resources.

Note: We don't dispose of the shared engine pool here since other trigger instances may still be using it. Engines will be cleaned up when the process exits or during worker shutdown.

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize trigger to dictionary.

Overrides base to include PostgreSQL-specific fields.

Returns:

Dict[str, Any] –

Dictionary representation

PostgresTriggerMode ¶

Modes for PostgreSQL trigger behavior.

EXISTS: Trigger when query returns at least one row CHANGED: Trigger when query results change (tracks hash of results) NEW: Trigger when new records appear (tracks set of IDs) COUNT: Trigger when count meets or exceeds threshold

from virtualitics_sdk.triggers.postgres_trigger import PostgresTrigger, PostgresTriggerMode

trigger = PostgresTrigger(
    name="Watch orders",
    table_name="orders",
    mode=PostgresTriggerMode.NEW_ROWS,
    callback=on_new_order
)

Asset Trigger¶

Monitor asset uploads or changes on the platform.

AssetTrigger ¶

AssetTrigger(name: str, mode: AssetTriggerMode = AssetTriggerMode.CREATED, asset_id: Optional[str] = None, asset_name: Optional[str] = None, asset_type: Optional[str] = None, label: Optional[str] = None, user_id: Optional[str] = None, app_id: Optional[str] = None, flow_id: Optional[str] = None, callback: Optional[PageUpdateCallback] = None, trigger_metadata: Optional[Dict[str, Any]] = None, trigger_id: Optional[str] = None, max_retries: int = 3, step_name: Optional[str] = None, dry_run: bool = False, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: bool = True, query_timeout: int = 30)

Specialized trigger for monitoring assets in the asset store.

NOTE: All triggers are checked by APScheduler every N seconds (configurable via TRIGGER_CHECK_INTERVAL env var, default 10 seconds). When a trigger's condition is met, an ARQ job is enqueued to execute the callback.

Inherits from PostgresTrigger and provides asset-specific monitoring capabilities. Automatically connects to the internal asset store database.

Key Features: - Monitors asset creation (CREATED mode) - Monitors asset updates (CHANGED mode) - Filters by asset_id, asset_name, asset_type, label, user_id - Automatically enforces organization_id from user for multi-tenancy security - Tracks state to detect new/changed assets

Security: - AssetTrigger ALWAYS enforces the organization_id of the user who created it - Users can ONLY monitor assets within their own organization - organization_id is not a configurable parameter and is set automatically

Examples: # Trigger when a specific asset is created trigger = AssetTrigger( name="dataset_created_monitor", mode=AssetTriggerMode.CREATED, asset_name="customer_data", asset_type="dataset", callback=on_asset_created, )

# Trigger when any asset for a user changes
trigger = AssetTrigger(
    name="user_asset_monitor",
    mode=AssetTriggerMode.CHANGED,
    user_id="user123",
    callback=on_asset_changed,
)

# Trigger when assets with specific label are created
# (organization_id automatically set from user)
trigger = AssetTrigger(
    name="labeled_assets_monitor",
    mode=AssetTriggerMode.CREATED,
    label="production",
    asset_type="model",
    callback=on_labeled_asset_created,
)

Initialize Asset trigger.

NOTE: All triggers are checked by APScheduler every N seconds (default 10s, configurable via TRIGGER_CHECK_INTERVAL env var).

NOTE: organization_id is automatically enforced during trigger activation (via configure()) to match the user's actual organization for security.

Parameters:

name (str) –

Trigger name
mode (AssetTriggerMode, default: CREATED ) –

AssetTriggerMode (CREATED or CHANGED)
asset_id (Optional[str], default: None ) –

Specific asset ID to monitor (optional)
asset_name (Optional[str], default: None ) –

Asset name pattern to match (optional)
asset_type (Optional[str], default: None ) –

Asset type to filter by (e.g., "dataset", "model") (optional)
label (Optional[str], default: None ) –

Asset label to filter by (optional)
user_id (Optional[str], default: None ) –

User ID to filter assets by (optional, set at activation if not provided)
app_id (Optional[str], default: None ) –

App ID (optional, set at activation)
flow_id (Optional[str], default: None ) –

Flow ID (optional, set at activation)
callback (Optional[PageUpdateCallback], default: None ) –

Callback to execute (optional, set at activation)
trigger_metadata (Optional[Dict[str, Any]], default: None ) –

Additional metadata
trigger_id (Optional[str], default: None ) –

Unique trigger ID
max_retries (int, default: 3 ) –

Max retry attempts
step_name (Optional[str], default: None ) –

Step name to associate with trigger (optional)
dry_run (bool, default: False ) –

If True, log what would execute without actually executing callbacks
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to callback
pass_store_interface (bool, default: True ) –

If True (default), pass store_interface to callback
query_timeout (int, default: 30 ) –

Query execution timeout in seconds (default 30)

configure ¶

configure(app_id: str, flow_id: str, user_id: str, callback: PageUpdateCallback, organization_id: Optional[str] = None, trigger_id: Optional[str] = None, step_name: Optional[str] = None, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: Optional[bool] = None) -> None

Configure runtime parameters for the AssetTrigger.

Automatically fetches and enforces organization_id from user_id to ensure assets are only monitored within the user's organization

Parameters:

app_id (str) –

ID of the app this trigger belongs to
flow_id (str) –

ID of the flow instance
user_id (str) –

ID of the user who owns this trigger
callback (PageUpdateCallback) –

PageUpdateCallback to execute when trigger conditions are met
organization_id (Optional[str], default: None ) –

Organization ID (will be overridden with user's organization)
trigger_id (Optional[str], default: None ) –

Optional trigger ID to override the auto-generated one
step_name (Optional[str], default: None ) –

Step name to associate with trigger
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to callback
pass_store_interface (Optional[bool], default: None ) –

If True, pass store_interface to callback

check `async` ¶

check() -> bool

Check if trigger condition is met.

Rebuilds the query on each check to ensure current user permissions are used. This allows the trigger to pick up permission changes without reconfiguration.

Returns:

bool –

True if condition met, False otherwise

execute `async` ¶

execute() -> None

Execute the callback with matched asset_ids.

Overrides parent execute() to pass asset_ids as a kwarg instead of full records.

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize trigger to dictionary.

Overrides parent to include asset-specific fields.

Returns:

Dict[str, Any] –

Dictionary representation

AssetTriggerMode ¶

Modes for Asset trigger behavior.

CREATED: Trigger when a new asset is created matching criteria CHANGED: Trigger when an existing asset is updated/modified

from virtualitics_sdk.triggers.asset_trigger import AssetTrigger, AssetTriggerMode

trigger = AssetTrigger(
    name="Watch datasets",
    mode=AssetTriggerMode.NEW_ASSET,
    callback=on_new_asset
)

Composite Trigger¶

Combine multiple triggers with AND/OR logic.

CompositeTrigger ¶

CompositeTrigger(name: str, triggers: List[Trigger], logic: Literal['AND', 'OR'] = 'AND', app_id: Optional[str] = None, flow_id: Optional[str] = None, user_id: Optional[str] = None, callback: Optional[PageUpdateCallback] = None, organization_id: Optional[str] = None, trigger_metadata: Optional[Dict[str, Any]] = None, trigger_id: Optional[str] = None, max_retries: int = 3, step_name: Optional[str] = None, dry_run: bool = False, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: bool = True)

Trigger that combines multiple triggers with AND/OR logic.

Example: # Trigger when BOTH conditions are met trigger = CompositeTrigger( name="data_pipeline_ready", triggers=[ S3Trigger(..., key="input/data.csv", mode=S3TriggerMode.EXISTS), S3Trigger(..., key="config/settings.json", mode=S3TriggerMode.CHANGED), ], logic='AND', callback=run_pipeline )

# Trigger when ANY condition is met
trigger = CompositeTrigger(
    name="alert_on_any_change",
    triggers=[
        S3Trigger(..., key="alert1.txt"),
        S3Trigger(..., key="alert2.txt"),
    ],
    logic='OR',
    callback=send_alert
)

Initialize composite trigger.

Parameters:

name (str) –

Trigger name
triggers (List[Trigger]) –

List of sub-triggers to combine
logic (Literal['AND', 'OR'], default: 'AND' ) –

Combination logic - 'AND' (all must be True) or 'OR' (any must be True)
app_id (Optional[str], default: None ) –

App ID (optional, set at activation)
flow_id (Optional[str], default: None ) –

Flow ID (optional, set at activation)
user_id (Optional[str], default: None ) –

User ID (optional, set at activation)
callback (Optional[PageUpdateCallback], default: None ) –

Callback to execute (optional, set at activation)
organization_id (Optional[str], default: None ) –

Organization ID (optional, set at activation)
trigger_metadata (Optional[Dict[str, Any]], default: None ) –

Additional metadata
trigger_id (Optional[str], default: None ) –

Unique trigger ID
max_retries (int, default: 3 ) –

Max retry attempts
step_name (Optional[str], default: None ) –

Step name to associate with trigger (optional)
dry_run (bool, default: False ) –

If True, log what would execute without actually executing callbacks
callback_kwargs (Optional[Dict[str, Any]], default: None ) –

Additional keyword arguments to pass to callback (will be pickled)
pass_store_interface (bool, default: True ) –

If True (default), pass store_interface to callback

deserialize_sub_triggers `staticmethod` ¶

deserialize_sub_triggers(sub_triggers_data: List[Dict[str, str]]) -> List[Trigger]

Deserialize sub-triggers from trigger_metadata.

Parameters:

sub_triggers_data (List[Dict[str, str]]) –

List of serialized trigger dicts

Returns:

List[Trigger] –

List of trigger instances

configure ¶

configure(app_id: str, flow_id: str, user_id: str, callback: PageUpdateCallback, organization_id: Optional[str] = None, trigger_id: Optional[str] = None, step_name: Optional[str] = None, callback_kwargs: Optional[Dict[str, Any]] = None, pass_store_interface: Optional[bool] = None) -> None

Configure runtime parameters and propagate to sub-triggers.

check `async` ¶

check() -> bool

Check all sub-triggers and apply logic.

Returns:

bool –

True if composite condition is met based on logic

stop `async` ¶

stop() -> None

Stop the composite trigger and all sub-triggers.

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize trigger to dictionary.

Returns:

Dict[str, Any] –

Dictionary representation

from virtualitics_sdk.triggers.composite_trigger import CompositeTrigger

# Fire when BOTH conditions are met
composite = CompositeTrigger(
    name="Watch all sources",
    triggers=[s3_trigger, db_trigger]
)

Using Triggers with SingletonApp¶

from virtualitics_sdk import SingletonApp
from virtualitics_sdk.triggers.s3_trigger import S3Trigger, S3TriggerMode

@page_update_callback
async def refresh_dashboard(store_interface):
    page = await store_interface.get_page()
    # Update dashboard elements...

s3_watcher = S3Trigger(
    name="Data uploads",
    bucket="analytics-data",
    prefix="daily/",
    mode=S3TriggerMode.NEW_OBJECTS,
    callback=refresh_dashboard
)

app = SingletonApp(
    name="Live Analytics",
    description="Auto-updating analytics dashboard",
    triggers=[s3_watcher]
)
app.chain([dashboard_step])

Trigger Flow Execution¶

For triggering entire flow executions (not just page updates), use the trigger_flow_execution utility:

trigger_flow_execution `async` ¶

trigger_flow_execution(flow_name: str, store_interface: StoreInterface, input_parameters: Optional[dict] = {})

A utility function to trigger the execution of another app (to be run headless). For example trigger the execution of another app upon the completion of a different app.

Parameters:

flow_name (str) –

The name of the app to trigger. App names are usually CamelCaseNames.
store_interface (StoreInterface) –

The StoreInterface to pass metadata about the app execution environment.
input_parameters (Optional[dict], default: {} ) –

Additional input parameters to pass to the triggered app

from virtualitics_sdk.trigger.trigger import trigger_flow_execution

async def run_downstream(flow_metadata):
    await trigger_flow_execution(
        flow_metadata=flow_metadata,
        app_name="Downstream Analysis",
        inputs={"data": processed_df}
    )

Best Practices¶

Idempotent callbacks: Trigger callbacks may fire multiple times — ensure they're safe to repeat
Lightweight checks: Trigger conditions are checked frequently (every ~10s) — keep check logic fast
Error handling: Use max_retries to handle transient failures
Dry run: Test triggers with dry_run=True before deploying
Naming: Use descriptive trigger names for debugging

Triggers¶

Overview¶

Trigger Base Class¶

Trigger ¶

id property ¶

name property ¶

is_running property ¶

should_retry property ¶

configure ¶

is_configured ¶

check abstractmethod async ¶

execute async ¶

start async ¶

stop async ¶

get_state ¶

set_state ¶

get_metrics ¶

to_dict ¶

S3 Trigger¶

S3Trigger ¶

check async ¶

stop async ¶

get_shared_session async classmethod ¶

to_dict ¶

S3TriggerMode ¶

PostgreSQL Trigger¶

PostgresTrigger ¶

check async ¶

execute async ¶

stop async ¶

to_dict ¶

PostgresTriggerMode ¶

Asset Trigger¶

AssetTrigger ¶

configure ¶

check async ¶

execute async ¶

to_dict ¶

AssetTriggerMode ¶

Composite Trigger¶

CompositeTrigger ¶

deserialize_sub_triggers staticmethod ¶

configure ¶

check async ¶

stop async ¶

to_dict ¶

Using Triggers with SingletonApp¶

Trigger Flow Execution¶

trigger_flow_execution async ¶

Best Practices¶

See Also¶

id `property` ¶

name `property` ¶

is_running `property` ¶

should_retry `property` ¶

check `abstractmethod` `async` ¶

execute `async` ¶

start `async` ¶

stop `async` ¶

check `async` ¶

stop `async` ¶

get_shared_session `async` `classmethod` ¶

check `async` ¶

execute `async` ¶

stop `async` ¶

check `async` ¶

execute `async` ¶

deserialize_sub_triggers `staticmethod` ¶

check `async` ¶

stop `async` ¶

trigger_flow_execution `async` ¶