Working with Triggers¶

Triggers alert you when query results cross a threshold. They continuously monitor your data and notify you when conditions are met.

Basic Trigger Operations¶

List Triggers¶

async def list_triggers(client: HoneycombClient, dataset: str) -> list:
    """List all triggers in a dataset.

    Args:
        client: Authenticated HoneycombClient
        dataset: Dataset slug to list triggers from

    Returns:
        List of triggers
    """
    triggers = await client.triggers.list_async(dataset)
    for trigger in triggers:
        status = "enabled" if not trigger.disabled else "disabled"
        print(f"{trigger.name} ({status}): threshold {trigger.threshold.value}")
    return triggers

Get a Specific Trigger¶

async def get_trigger(client: HoneycombClient, dataset: str, trigger_id: str):
    """Get a trigger by ID.

    Args:
        client: Authenticated HoneycombClient
        dataset: Dataset slug containing the trigger
        trigger_id: ID of the trigger to retrieve

    Returns:
        The trigger object
    """
    trigger = await client.triggers.get_async(dataset, trigger_id)
    print(f"Name: {trigger.name}")
    print(f"Threshold: {trigger.threshold.op.value} {trigger.threshold.value}")
    print(f"Frequency: every {trigger.frequency}s")
    print(f"Status: {'enabled' if not trigger.disabled else 'disabled'}")
    return trigger

Update a Trigger¶

async def update_trigger(
    client: HoneycombClient, dataset: str, trigger_id: str
):
    """Update a trigger's threshold and enable it.

    Args:
        client: Authenticated HoneycombClient
        dataset: Dataset slug containing the trigger
        trigger_id: ID of the trigger to update

    Returns:
        The updated trigger
    """
    # Get existing trigger first to preserve values
    existing = await client.triggers.get_async(dataset, trigger_id)

    # Update with new values
    updated = TriggerWithInlineQuery(
        name=existing.name,
        description=existing.description,
        threshold=TriggerThreshold(
            op=existing.threshold.op,
            value=200.0,  # Change threshold value
        ),
        frequency=existing.frequency,
        disabled=False,  # Enable the trigger
        query=existing.query,
    )

    result = await client.triggers.update_async(dataset, trigger_id, updated)
    return result

Delete a Trigger¶

async def delete_trigger(client: HoneycombClient, dataset: str, trigger_id: str) -> None:
    """Delete a trigger.

    Args:
        client: Authenticated HoneycombClient
        dataset: Dataset slug containing the trigger
        trigger_id: ID of the trigger to delete
    """
    await client.triggers.delete_async(dataset, trigger_id)

Creating Triggers¶

Using TriggerBuilder (Recommended)¶

async def create_simple_trigger(client: HoneycombClient, dataset: str) -> str:
    """Create a simple count-based trigger using TriggerBuilder.

    Args:
        client: Authenticated HoneycombClient
        dataset: Dataset slug to create trigger in

    Returns:
        The created trigger ID
    """
    bundle = (
        TriggerBuilder("High Request Count")
        .dataset(dataset)
        .description("Alert when request count exceeds threshold")
        .last_30_minutes()
        .count()
        .threshold_gt(1000)
        .every_15_minutes()
        .disabled()  # Start disabled for safety
        .build()
    )
    created = await client.triggers.create_from_bundle_async(bundle)
    return created.id

Manual Construction¶

async def create_trigger_manual(client: HoneycombClient, dataset: str) -> str:
    """Create a trigger using manual construction.

    Args:
        client: Authenticated HoneycombClient
        dataset: Dataset slug to create trigger in

    Returns:
        The created trigger ID
    """
    trigger = TriggerWithInlineQuery(
        name="Manual Test Trigger",
        description="Created without builder",
        threshold=TriggerThreshold(
            op=TriggerThresholdOp.GREATER_THAN,
            value=100.0,
        ),
        frequency=900,
        disabled=True,
        query=QueryBuilder().last_30_minutes().count().build_for_trigger(),
    )
    created = await client.triggers.create_async(dataset, trigger)
    return created.id

TriggerBuilder Features¶

TriggerBuilder provides a fluent interface that combines: - Query specification (inherited from QueryBuilder) - Threshold configuration (.threshold_gt(), .threshold_lt(), etc.) - Frequency settings (.every_5_minutes(), .every_15_minutes(), etc.) - Recipients (.email(), .slack(), .pagerduty(), .webhook()) - Tags (.tag(key, value) - max 10 tags) - Baseline comparison (.baseline_1_hour_ago(), .baseline_1_day_ago(), etc.)

See the QueryBuilder documentation for full query methods.

Important Constraints¶

Triggers have API-enforced limits that TriggerBuilder validates at build time:

Time range: Maximum 3600 seconds (1 hour)
Frequency vs duration: time_range ≤ frequency × 4
Single calculation: Only one calculation per trigger (e.g., COUNT, P99, AVG)
No absolute time: Must use relative time ranges

Sync Usage¶

All trigger operations have sync equivalents:

with HoneycombClient(api_key="...", sync=True) as client:
    # List triggers
    triggers = client.triggers.list("my-dataset")

    # Create trigger
    trigger = TriggerBuilder("High Errors").dataset("my-dataset").count().threshold_gt(100).build()
    created = client.triggers.create("my-dataset", trigger)

    # Get trigger
    trigger = client.triggers.get("my-dataset", trigger_id)

    # Update trigger
    updated = client.triggers.update("my-dataset", trigger_id, TriggerCreate(...))

    # Delete trigger
    client.triggers.delete("my-dataset", trigger_id)