prefect.server.events.triggers

The triggers consumer watches events streaming in from the event bus and decides whether to act on them based on the automations that users have set up.

act(firing) async

Given a Automation that has been triggered, the triggering labels and event (if there was one), publish an action for the actions service to process.

Source code in src/prefect/server/events/triggers.py

async def act(firing: Firing):
    """Given a Automation that has been triggered, the triggering labels and event
    (if there was one), publish an action for the `actions` service to process."""
    automation = firing.trigger.automation

    state_change_events: Dict[TriggerState, ReceivedEvent] = {
        trigger_state: firing.trigger.create_automation_state_change_event(
            firing=firing,
            trigger_state=trigger_state,
        )
        for trigger_state in sorted(firing.trigger_states, key=list(TriggerState).index)
    }
    await messaging.publish(state_change_events.values())

    # By default, all `automation.actions` are fired
    source_actions: List[Tuple[Optional[ReceivedEvent], ServerActionTypes]] = [
        (firing.triggering_event, action) for action in automation.actions
    ]

    # Conditionally add in actions that fire on specific trigger states
    if TriggerState.Triggered in firing.trigger_states:
        source_actions += [
            (state_change_events[TriggerState.Triggered], action)
            for action in automation.actions_on_trigger
        ]

    if TriggerState.Resolved in firing.trigger_states:
        source_actions += [
            (state_change_events[TriggerState.Resolved], action)
            for action in automation.actions_on_resolve
        ]

    actions = [
        TriggeredAction(
            automation=automation,
            firing=firing,
            triggered=firing.triggered,
            triggering_labels=firing.triggering_labels,
            triggering_event=action_triggering_event,
            action=action,
            action_index=index,
        )
        for index, (action_triggering_event, action) in enumerate(source_actions)
    ]

    async with messaging.create_actions_publisher() as publisher:
        for action in actions:
            await publisher.publish_data(action.model_dump_json().encode(), {})

consumer(periodic_granularity=timedelta(seconds=5)) async

The triggers.consumer processes all Events arriving on the event bus to determine if they meet the automation criteria, queuing up a corresponding TriggeredAction for the actions service if the automation criteria is met.

Source code in src/prefect/server/events/triggers.py

@asynccontextmanager
async def consumer(
    periodic_granularity: timedelta = timedelta(seconds=5),
) -> AsyncGenerator[MessageHandler, None]:
    """The `triggers.consumer` processes all Events arriving on the event bus to
    determine if they meet the automation criteria, queuing up a corresponding
    `TriggeredAction` for the `actions` service if the automation criteria is met."""
    async with automations_session() as session:
        await load_automations(session)

    proactive_task = asyncio.create_task(evaluate_periodically(periodic_granularity))

    ordering = causal_ordering()

    async def message_handler(message: Message):
        if not message.data:
            logger.warning("Message had no data")

            return

        if not message.attributes:
            logger.warning("Message had no attributes")

            return

        if message.attributes.get("event") == "prefect.log.write":
            return

        try:
            event_id = UUID(message.attributes["id"])
        except (KeyError, ValueError, TypeError):
            logger.warning(
                "Unable to get event ID from message attributes: %s",
                repr(message.attributes),
            )
            return

        if await ordering.event_has_been_seen(event_id):
            return

        event = ReceivedEvent.model_validate_json(message.data)

        try:
            await reactive_evaluation(event)
        except EventArrivedEarly:
            pass  # it's fine to ACK this message, since it is safe in the DB

    try:
        logger.debug("Starting reactive evaluation task")
        yield message_handler
    finally:
        proactive_task.cancel()

ensure_bucket(db, session, trigger, bucketing_key, start, end, last_event, initial_count=0) async

Ensures that a bucket has been started for the given automation and key, returning the current bucket. Will not modify the existing bucket.

Source code in src/prefect/server/events/triggers.py

@db_injector
async def ensure_bucket(
    db: PrefectDBInterface,
    session: AsyncSession,
    trigger: EventTrigger,
    bucketing_key: Tuple[str, ...],
    start: DateTime,
    end: DateTime,
    last_event: Optional[ReceivedEvent],
    initial_count: int = 0,
) -> "ORMAutomationBucket":
    """Ensures that a bucket has been started for the given automation and key,
    returning the current bucket.  Will not modify the existing bucket."""
    automation = trigger.automation
    additional_updates: dict = {"last_event": last_event} if last_event else {}
    await session.execute(
        db.insert(db.AutomationBucket)
        .values(
            automation_id=automation.id,
            trigger_id=trigger.id,
            bucketing_key=bucketing_key,
            last_event=last_event,
            start=start,
            end=end,
            count=initial_count,
            last_operation="ensure_bucket[insert]",
        )
        .on_conflict_do_update(
            index_elements=[
                db.AutomationBucket.automation_id,
                db.AutomationBucket.trigger_id,
                db.AutomationBucket.bucketing_key,
            ],
            set_=dict(
                # no-op, but this counts as an update so the query returns a row
                count=db.AutomationBucket.count,
                **additional_updates,
            ),
        )
    )

    return await read_bucket_by_trigger_id(
        session, automation.id, trigger.id, bucketing_key
    )

evaluate(session, trigger, bucket, now, triggering_event) async

Evaluates an Automation, either triggered by a specific event or proactively on a time interval. Evaluating a Automation updates the associated counters for each automation, and will fire the associated action if it has met the threshold.

Source code in src/prefect/server/events/triggers.py

async def evaluate(
    session: AsyncSession,
    trigger: EventTrigger,
    bucket: "ORMAutomationBucket",
    now: DateTime,
    triggering_event: Optional[ReceivedEvent],
) -> Optional["ORMAutomationBucket"]:
    """Evaluates an Automation, either triggered by a specific event or proactively
    on a time interval.  Evaluating a Automation updates the associated counters for
    each automation, and will fire the associated action if it has met the threshold."""
    automation = trigger.automation

    # Implementation notes:
    #
    # This triggering algorithm maintains an invariant that there is exactly one
    # time-based "bucket" open and collecting events for each automation at a time. When
    # an event comes in that matches the automation, one of three things can happen:
    #
    # 1. The event would have matched an older bucket that has either expired or has
    #    already filled up, and thus is no longer relevant;
    # 2. The event matches the current bucket, but the bucket does not meet it's
    #    threshold yet;
    # 3. The event matches the current bucket, causes it to meet the the threshold, so
    #    we act immediately and advance the bucket to the next time period.
    #
    # Automations are also evaluated proactively without an event to see if they have
    # met their proactive threshold (where not enough events have happened in the time
    # period)

    # If there was a triggering event, then we need to "spend" this count somewhere,
    # either in the currently open bucket or in the next time period's bucket
    count = 1 if triggering_event else 0

    if now < bucket.start:
        # This is an older out-of-order message or a redundant event for a reactive
        # trigger that has has already fired, so it should not should not affect the
        # current bucket.  We can safely ignore this event/timestamp entirely.
        return bucket

    if count and (trigger.immediate or bucket.start <= now < bucket.end):
        # we are still within the automation time period, so spend the count in the
        # current bucket
        bucket = await increment_bucket(session, bucket, count, triggering_event)
        count = 0

    # Reactive automations will fire "eagerly", meaning they will fire as _soon_ as the
    # threshold is reached, then wait for the rest of their time period before
    # firing again.  This is done by creating a new bucket in the future after the
    # trigger has fired.

    # Proactive automations must wait until the whole bucket has expired before they can
    # fire, because we can't know if we'll get one late event just in time to cause the
    # automation _not_ to fire.

    ready_to_fire = trigger.posture == Posture.Reactive or bucket.end <= now

    if ready_to_fire and trigger.meets_threshold(bucket.count):
        logger.info(
            (
                "Automation %s (%r) trigger %s triggered for keys (%r) %s, "
                "having occurred %s times between %s and %s"
            ),
            automation.id,
            automation.name,
            trigger.id,
            bucket.bucketing_key,
            "reactively" if triggering_event else "proactively",
            bucket.count,
            bucket.start,
            bucket.end,
            extra={
                "automation": automation.id,
                "trigger": trigger.id,
                "bucketing_key": bucket.bucketing_key,
                "triggering_event": (
                    {
                        "id": triggering_event.id,
                        "event": triggering_event.event,
                    }
                    if triggering_event
                    else None
                ),
            },
        )

        firing = Firing(
            trigger=trigger,
            trigger_states={TriggerState.Triggered},
            triggered=pendulum.now("UTC"),
            triggering_labels={
                label: value
                for label, value in zip(sorted(trigger.for_each), bucket.bucketing_key)
            },
            triggering_event=triggering_event or bucket.last_event,
        )

        await fire(session, firing)

        # when acting, remove the current bucket from the database immediately to avoid
        # potentially double-acting in the case of a crash between now and the next
        # time we backup buckets to the database
        await remove_bucket(session, bucket)

    elif now < bucket.end:
        # We didn't fire this time, and also the bucket still has more time, so leave
        # before setting up a new future bucket
        return bucket

    # We are now outside of the automation's time period or we triggered this
    # time.  That means it's time to start a new bucket for the next possible time
    # window (if this automation does not require an event to start `after`):
    if trigger.after:
        # remove the bucket because it should only get re-created if we see another
        # appropriate starting event
        return await remove_bucket(session, bucket)
    else:
        if trigger.within == timedelta(seconds=0):
            return None

        start = pendulum.instance(max(bucket.end, now))
        return await start_new_bucket(
            session,
            trigger,
            bucketing_key=bucket.bucketing_key,
            start=start,
            end=start + trigger.within,
            count=count,
        )

evaluate_periodically(periodic_granularity) async

Runs periodic evaluation on the given interval

Source code in src/prefect/server/events/triggers.py

async def evaluate_periodically(periodic_granularity: timedelta):
    """Runs periodic evaluation on the given interval"""
    logger.debug(
        "Starting periodic evaluation task every %s seconds",
        periodic_granularity.total_seconds(),
    )
    while True:
        try:
            await periodic_evaluation(pendulum.now("UTC"))
        except Exception:
            logger.exception("Error running periodic evaluation")
        finally:
            await asyncio.sleep(periodic_granularity.total_seconds())

forget_automation(automation_id)

Unloads the given automation from memory

Source code in src/prefect/server/events/triggers.py

def forget_automation(automation_id: UUID):
    """Unloads the given automation from memory"""
    if automation := automations_by_id.pop(automation_id, None):
        for trigger in automation.triggers():
            triggers.pop(trigger.id, None)
            next_proactive_runs.pop(trigger.id, None)

get_events_clock_offset() async

Calculate the current clock offset. This takes into account both the occurred of the last event, as well as the time we saw the last event. This helps to ensure that in low volume environments, we don't end up getting huge offsets.

Source code in src/prefect/server/events/triggers.py

async def get_events_clock_offset() -> float:
    """Calculate the current clock offset.  This takes into account both the `occurred`
    of the last event, as well as the time we _saw_ the last event.  This helps to
    ensure that in low volume environments, we don't end up getting huge offsets."""
    global _events_clock, _events_clock_updated

    async with _events_clock_lock():
        if _events_clock is None or _events_clock_updated is None:
            return 0.0

        now: float = pendulum.now("UTC").float_timestamp
        offset = (_events_clock - now) + (now - _events_clock_updated)

    return offset

get_lost_followers() async

Get followers that have been sitting around longer than our lookback

Source code in src/prefect/server/events/triggers.py

@retry_async_fn(max_attempts=3, retry_on_exceptions=(sa.exc.OperationalError,))
async def get_lost_followers():
    """Get followers that have been sitting around longer than our lookback"""
    return await causal_ordering().get_lost_followers()

increment_bucket(db, session, bucket, count, last_event) async

Adds the given count to the bucket, returning the new bucket

Source code in src/prefect/server/events/triggers.py

@db_injector
async def increment_bucket(
    db: PrefectDBInterface,
    session: AsyncSession,
    bucket: "ORMAutomationBucket",
    count: int,
    last_event: Optional[ReceivedEvent],
) -> "ORMAutomationBucket":
    """Adds the given count to the bucket, returning the new bucket"""
    additional_updates: dict = {"last_event": last_event} if last_event else {}
    await session.execute(
        db.insert(db.AutomationBucket)
        .values(
            automation_id=bucket.automation_id,
            trigger_id=bucket.trigger_id,
            bucketing_key=bucket.bucketing_key,
            start=bucket.start,
            end=bucket.end,
            count=count,
            last_operation="increment_bucket[insert]",
        )
        .on_conflict_do_update(
            index_elements=[
                db.AutomationBucket.automation_id,
                db.AutomationBucket.trigger_id,
                db.AutomationBucket.bucketing_key,
            ],
            set_=dict(
                count=db.AutomationBucket.count + count,
                last_operation="increment_bucket[update]",
                updated=pendulum.now("UTC"),
                **additional_updates,
            ),
        )
    )

    return await read_bucket_by_trigger_id(
        session,
        bucket.automation_id,
        bucket.trigger_id,
        bucket.bucketing_key,
    )

load_automation(automation)

Loads the given automation into memory so that it is available for evaluations

Source code in src/prefect/server/events/triggers.py

def load_automation(automation: Optional[Automation]):
    """Loads the given automation into memory so that it is available for evaluations"""
    if not automation:
        return

    event_triggers = automation.triggers_of_type(EventTrigger)

    if not automation.enabled or not event_triggers:
        forget_automation(automation.id)
        return

    automations_by_id[automation.id] = automation

    for trigger in event_triggers:
        triggers[trigger.id] = trigger
        next_proactive_runs.pop(trigger.id, None)

load_automations(db, session) async

Loads all automations for the given set of accounts

Source code in src/prefect/server/events/triggers.py

@db_injector
async def load_automations(db: PrefectDBInterface, session: AsyncSession):
    """Loads all automations for the given set of accounts"""
    query = sa.select(db.Automation)

    logger.debug("Loading automations")

    result = await session.execute(query)
    for automation in result.scalars().all():
        load_automation(Automation.model_validate(automation, from_attributes=True))

    logger.debug(
        "Loaded %s automations with %s triggers", len(automations_by_id), len(triggers)
    )

periodic_evaluation(now) async

Periodic tasks that should be run regularly, but not as often as every event

Source code in src/prefect/server/events/triggers.py

async def periodic_evaluation(now: DateTime):
    """Periodic tasks that should be run regularly, but not as often as every event"""
    offset = await get_events_clock_offset()
    as_of = now + timedelta(seconds=offset)

    logger.debug("Running periodic evaluation as of %s (offset %ss)", as_of, offset)

    # Any followers that have been sitting around longer than our lookback are never
    # going to see their leader event (maybe it was lost or took too long to arrive).
    # These events can just be evaluated now in the order they occurred.
    for event in await get_lost_followers():
        await reactive_evaluation(event)

    async with automations_session() as session:
        await sweep_closed_buckets(
            session,
            as_of - PREFECT_EVENTS_EXPIRED_BUCKET_BUFFER.value(),
        )
        await session.commit()

proactive_evaluation(trigger, as_of) async

The core proactive evaluation operation for a single Automation

Source code in src/prefect/server/events/triggers.py

async def proactive_evaluation(trigger: EventTrigger, as_of: DateTime) -> DateTime:
    """The core proactive evaluation operation for a single Automation"""
    assert isinstance(trigger, EventTrigger), repr(trigger)
    automation = trigger.automation

    offset = await get_events_clock_offset()
    as_of += timedelta(seconds=offset)

    logger.debug(
        "Evaluating automation %s trigger %s proactively as of %s (offset %ss)",
        automation.id,
        trigger.id,
        as_of,
        offset,
    )

    # By default, the next run will come after the full trigger window, but it
    # may be sooner based on the state of the buckets
    run_again_at = as_of + trigger.within

    async with automations_session() as session:
        try:
            if not trigger.for_each:
                await ensure_bucket(
                    session,
                    trigger,
                    bucketing_key=tuple(),
                    start=as_of,
                    end=as_of + trigger.within,
                    last_event=None,
                )

            # preemptively delete buckets where possible without
            # evaluating them in memory
            await remove_buckets_exceeding_threshold(session, trigger)

            async for bucket in read_buckets_for_automation(session, trigger):
                next_bucket = await evaluate(
                    session, trigger, bucket, as_of, triggering_event=None
                )
                if next_bucket and as_of < next_bucket.end < run_again_at:
                    run_again_at = pendulum.instance(next_bucket.end)

            return run_again_at
        finally:
            await session.commit()

reactive_evaluation(event, depth=0) async

Evaluate all automations that may apply to this event.

event (ReceivedEvent): The event to evaluate. This object contains all the necessary information about the event, including its type, associated resources, and metadata. depth (int, optional): The current recursion depth. This is used to prevent infinite recursion due to cyclic event dependencies. Defaults to 0 and is incremented with each recursive call.

Source code in src/prefect/server/events/triggers.py

async def reactive_evaluation(event: ReceivedEvent, depth: int = 0):
    """
    Evaluate all automations that may apply to this event.

    Args:
    event (ReceivedEvent): The event to evaluate. This object contains all the necessary information
        about the event, including its type, associated resources, and metadata.
    depth (int, optional): The current recursion depth. This is used to prevent infinite recursion
        due to cyclic event dependencies. Defaults to 0 and is incremented with
        each recursive call.

    """
    async with AsyncExitStack() as stack:
        await update_events_clock(event)
        await stack.enter_async_context(
            causal_ordering().preceding_event_confirmed(
                reactive_evaluation, event, depth
            )
        )

        interested_triggers = find_interested_triggers(event)
        if not interested_triggers:
            return

        for trigger in interested_triggers:
            logger.info(
                "Automation %s, trigger %s covers event %r (%s) for %r at %r",
                trigger.automation.id,
                trigger.id,
                event.event,
                event.id,
                event.resource.id,
                event.occurred.isoformat(),
            )

            bucketing_key = trigger.bucketing_key(event)

            async with automations_session(begin_transaction=True) as session:
                try:
                    bucket: Optional["ORMAutomationBucket"] = None

                    if trigger.after and trigger.starts_after(event.event):
                        # When an event matches both the after and expect, each event
                        # can both start a new bucket and increment the bucket that was
                        # started by the previous event.  Here we offset the bucket to
                        # start at -1 so that the first event will leave the bucket at 0
                        # after evaluation.  See the tests:
                        #
                        #   test_same_event_in_expect_and_after_never_reacts_immediately
                        #   test_same_event_in_expect_and_after_reacts_after_threshold_is_met
                        #   test_same_event_in_expect_and_after_proactively_does_not_fire
                        #   test_same_event_in_expect_and_after_proactively_fires
                        #
                        # in test_triggers_regressions.py for examples of how we expect
                        # this to behave.
                        #
                        # https://github.com/PrefectHQ/nebula/issues/4201
                        initial_count = -1 if trigger.expects(event.event) else 0
                        bucket = await ensure_bucket(
                            session,
                            trigger,
                            bucketing_key,
                            start=event.occurred,
                            end=event.occurred + trigger.within,
                            last_event=event,
                            initial_count=initial_count,
                        )

                    if (
                        not bucket
                        and not trigger.after
                        and trigger.expects(event.event)
                    ):
                        # When ensuring a bucket and _creating it for the first time_,
                        # use an old time so that we can catch any other events flowing
                        # through the system at the same time even if they are out of
                        # order.  After the trigger fires and creates its next bucket,
                        # time will start from that point forward.  We'll use our
                        # preceding event lookback variable as the horizon that we'll
                        # accept these older events.
                        #
                        # https://github.com/PrefectHQ/nebula/issues/7230
                        start = event.occurred - PRECEDING_EVENT_LOOKBACK

                        bucket = await ensure_bucket(
                            session,
                            trigger,
                            bucketing_key=bucketing_key,
                            start=start,
                            end=event.occurred + trigger.within,
                            last_event=event,
                        )

                    if not trigger.expects(event.event):
                        continue

                    if not bucket:
                        bucket = await read_bucket(session, trigger, bucketing_key)
                        if not bucket:
                            continue

                    await evaluate(
                        session,
                        trigger,
                        bucket,
                        event.occurred,
                        triggering_event=event,
                    )
                finally:
                    await session.commit()

read_bucket(db, session, trigger, bucketing_key) async

Gets the bucket this event would fall into for the given Automation, if there is one currently

Source code in src/prefect/server/events/triggers.py

@db_injector
async def read_bucket(
    db: PrefectDBInterface,
    session: AsyncSession,
    trigger: Trigger,
    bucketing_key: Tuple[str, ...],
) -> Optional["ORMAutomationBucket"]:
    """Gets the bucket this event would fall into for the given Automation, if there is
    one currently"""
    return await read_bucket_by_trigger_id(
        session,
        trigger.automation.id,
        trigger.id,
        bucketing_key,
    )

read_bucket_by_trigger_id(db, session, automation_id, trigger_id, bucketing_key) async

Gets the bucket this event would fall into for the given Automation, if there is one currently

Source code in src/prefect/server/events/triggers.py

@db_injector
async def read_bucket_by_trigger_id(
    db: PrefectDBInterface,
    session: AsyncSession,
    automation_id: UUID,
    trigger_id: UUID,
    bucketing_key: Tuple[str, ...],
) -> Optional["ORMAutomationBucket"]:
    """Gets the bucket this event would fall into for the given Automation, if there is
    one currently"""
    query = sa.select(db.AutomationBucket).where(
        db.AutomationBucket.automation_id == automation_id,
        db.AutomationBucket.trigger_id == trigger_id,
        db.AutomationBucket.bucketing_key == bucketing_key,
    )
    result = await session.execute(query)
    bucket = result.scalars().first()
    if bucket:
        await session.refresh(bucket)

    return bucket

read_buckets_for_automation(db, session, trigger, batch_size=AUTOMATION_BUCKET_BATCH_SIZE) async

Yields buckets for the given automation and trigger in batches.

Source code in src/prefect/server/events/triggers.py

@db_injector
async def read_buckets_for_automation(
    db: PrefectDBInterface,
    session: AsyncSession,
    trigger: Trigger,
    batch_size: int = AUTOMATION_BUCKET_BATCH_SIZE,
) -> AsyncGenerator["ORMAutomationBucket", None]:
    """Yields buckets for the given automation and trigger in batches."""
    offset = 0

    while True:
        query = (
            sa.select(db.AutomationBucket)
            .where(
                db.AutomationBucket.automation_id == trigger.automation.id,
                db.AutomationBucket.trigger_id == trigger.id,
            )
            .order_by(db.AutomationBucket.start)
            .limit(batch_size)
            .offset(offset)
        )

        result = await session.execute(query)
        buckets = result.scalars().all()

        if not buckets:
            break

        for bucket in buckets:
            yield bucket

        offset += batch_size

remove_bucket(db, session, bucket) async

Removes the given bucket from the database

Source code in src/prefect/server/events/triggers.py

@db_injector
async def remove_bucket(
    db: PrefectDBInterface, session: AsyncSession, bucket: "ORMAutomationBucket"
):
    """Removes the given bucket from the database"""
    await session.execute(
        sa.delete(db.AutomationBucket).where(
            db.AutomationBucket.automation_id == bucket.automation_id,
            db.AutomationBucket.trigger_id == bucket.trigger_id,
            db.AutomationBucket.bucketing_key == bucket.bucketing_key,
        )
    )

remove_buckets_exceeding_threshold(db, session, trigger) async

Deletes bucket where the count has already exceeded the threshold

Source code in src/prefect/server/events/triggers.py

@db_injector
async def remove_buckets_exceeding_threshold(
    db: PrefectDBInterface, session: AsyncSession, trigger: EventTrigger
):
    """Deletes bucket where the count has already exceeded the threshold"""
    assert isinstance(trigger, EventTrigger), repr(trigger)
    await session.execute(
        sa.delete(db.AutomationBucket).where(
            db.AutomationBucket.automation_id == trigger.automation.id,
            db.AutomationBucket.trigger_id == trigger.id,
            db.AutomationBucket.count >= trigger.threshold,
        )
    )

reset() async

Resets the in-memory state of the service

Source code in src/prefect/server/events/triggers.py

async def reset():
    """Resets the in-memory state of the service"""
    reset_events_clock()
    automations_by_id.clear()
    triggers.clear()
    next_proactive_runs.clear()

start_new_bucket(db, session, trigger, bucketing_key, start, end, count, triggered_at=None) async

Ensures that a bucket with the given start and end exists with the given count, returning the new bucket

Source code in src/prefect/server/events/triggers.py

@db_injector
async def start_new_bucket(
    db: PrefectDBInterface,
    session: AsyncSession,
    trigger: EventTrigger,
    bucketing_key: Tuple[str, ...],
    start: DateTime,
    end: DateTime,
    count: int,
    triggered_at: Optional[DateTime] = None,
) -> "ORMAutomationBucket":
    """Ensures that a bucket with the given start and end exists with the given count,
    returning the new bucket"""
    automation = trigger.automation

    await session.execute(
        db.insert(db.AutomationBucket)
        .values(
            automation_id=automation.id,
            trigger_id=trigger.id,
            bucketing_key=bucketing_key,
            start=start,
            end=end,
            count=count,
            last_operation="start_new_bucket[insert]",
            triggered_at=triggered_at,
        )
        .on_conflict_do_update(
            index_elements=[
                db.AutomationBucket.automation_id,
                db.AutomationBucket.trigger_id,
                db.AutomationBucket.bucketing_key,
            ],
            set_=dict(
                start=start,
                end=end,
                count=count,
                last_operation="start_new_bucket[update]",
                updated=pendulum.now("UTC"),
                triggered_at=triggered_at,
            ),
        )
    )

    return await read_bucket_by_trigger_id(
        session,
        automation.id,
        trigger.id,
        bucketing_key,
    )