Cloud Processors

Processors are serverless, event-driven cloud functions that run on AWS Lambda. They react to platform events -- new channel messages, aggregate updates, scheduled timers, external HTTP requests, deployments, and manual invocations -- and execute short-lived logic in response.

Unlike Docker applications that run continuously on edge devices, processors are stateless and ephemeral. Each invocation handles a single event, performs its work, and exits. State is persisted between invocations through tags and channels.

Processors are the right choice when your logic:

• Reacts to data changes rather than polling for them
• Does not need direct hardware access (no GPIO, Modbus, or platform I/O)
• Benefits from cloud-scale execution (cross-device orchestration, external API calls, report generation)
• Should run on a schedule (cron or rate-based)
• Needs to receive inbound HTTP requests from third-party systems

Common use cases include alerting, data transformation, dashboard aggregation, integration with external APIs, and cross-device orchestration.

Every processor follows the same structure: subclass Application, override the event handlers you need, and wire up a Lambda handler function that calls run_app.

from pydoover.processor import Application, run_app
from pydoover.models import MessageCreateEvent


class MyProcessor(Application):
    async def setup(self):
        # One-time initialisation before the event handler runs.
        # Tags, config, UI, and the API client are all available here.
        pass

    async def on_message_create(self, event: MessageCreateEvent):
        # React to a new message on a subscribed channel.
        temp = event.message.data.get("temperature")
        if temp and temp > 40.0:
            await self.send_notification(
                f"High temperature alert: {temp}C",
                title="Temperature Warning",
            )


# Lambda entry point
app = MyProcessor()

def handler(event, context):
    return run_app(app, event, context)

The run_app function is the Lambda entry point. It extracts the event payload from the SNS or EventBridge wrapper, sets up structured JSON logging, and routes the event through the processor's lifecycle: pre-hook filter, setup, post-setup filter, event dispatch, tag commit, and cleanup.

run_app(app, event, context, setup_logging=True) is the function you call from your Lambda handler.

It handles:

• Event extraction: Unwraps SNS records or passes EventBridge payloads through directly
• Logging setup: Redirects print() and sys.stderr through a structured JSON logger so all output carries the Lambda request ID and app metadata
• Event routing: Dispatches to the appropriate handler based on the event's op field

The setup_logging parameter defaults to True. Set it to False only if you are managing logging configuration yourself (e.g., in tests).

Each processor invocation follows these phases:

1. Pre-hook filter -- fast rejection before any API calls. Examines the raw event payload to decide whether processing should continue.
2. Setup -- token upgrade, tag and UI initialisation, configuration injection, and the user's setup() method.
3. Post-setup filter -- second rejection gate with access to tags, config, and the API client.
4. Event dispatch -- routes to the appropriate handler method (on_message_create, on_schedule, etc.).
5. Tag commit -- all tag changes made during the handler are committed to the cloud automatically.
6. Invocation summary -- timing, status, and error information published to configured channels.
7. Cleanup -- the user's close() method runs and the API client session is closed.

• Application Class -- detailed coverage of the Application class, all handler methods, filtering, and utilities
• Event Types -- all event types with their fields and handler patterns
• Processor Data Client -- the ProcessorDataClient and its anti-recursion protections
• Configuration -- defining typed configuration schemas for your processor