Events and schedules

Install the relevant client packages from the checkout; see Getting started. Examples use existing resources and require permission for the requested operation.

These are three independent clients:

Client	Purpose
`EventBridgeClient()`	Event buses, events, rules, targets, archives, and replays
`EventBridgeSchedulerClient()`	One-time and recurring schedules, time zones, and flexible time windows
`EventBridgePipesClient()`	Connect event sources to targets with filtering and optional enrichment

Traditional scheduled rules belong to EventBridgeClient(). EventBridge Scheduler has its own EventBridgeSchedulerClient() client.

Publish a typed event to an existing bus:

import json

from rboto import EventBridgeClient
from rboto_eventbridge.types import PutEventsRequestEntry


async def publish_order(bus_name: str, order_id: str) -> list[str]:
    entry: PutEventsRequestEntry = {
        "event_bus_name": bus_name,
        "source": "myapp.orders",
        "detail_type": "OrderCreated",
        "detail": json.dumps({"order_id": order_id}),
    }
    response = await EventBridgeClient(region="us-east-1").put_events(entries=[entry])
    entries = response.entries or []
    if response.failed_entry_count:
        failures = [
            (item.error_code, item.error_message) for item in entries if item.error_code is not None
        ]
        raise RuntimeError(f"Event publication failed: {failures}")
    return [item.event_id for item in entries if item.event_id is not None]

PutEvents reports per-entry failures in a successful HTTP response. Check failed_entry_count and each entry's error_code.

Create a schedule using an existing target and execution role:

from rboto import EventBridgeSchedulerClient
from rboto_scheduler.types import Target


async def schedule_message(queue_arn: str, role_arn: str) -> str:
    target: Target = {
        "arn": queue_arn,
        "role_arn": role_arn,
        "input": '{"message":"scheduled by rboto"}',
    }
    response = await EventBridgeSchedulerClient(region="us-east-1").create_schedule(
        name="daily-message",
        schedule_expression="cron(0 9 * * ? *)",
        schedule_expression_timezone="America/Sao_Paulo",
        flexible_time_window={"mode": "OFF"},
        target=target,
    )
    return response.schedule_arn

Create a pipe between existing SQS queues:

from rboto import EventBridgePipesClient


async def connect_queues(source_arn: str, target_arn: str, role_arn: str) -> str | None:
    response = await EventBridgePipesClient(region="us-east-1").create_pipe(
        name="queue-forwarder",
        source=source_arn,
        target=target_arn,
        role_arn=role_arn,
        desired_state="RUNNING",
        source_parameters={"sqs_queue_parameters": {"batch_size": 1}},
    )
    return response.arn

The execution roles must trust their respective services and have access to their sources and targets. Pipe creation is asynchronous; use describe_pipe() to check current_state. Remove resources with delete_schedule() and delete_pipe() when done.