Skip to main content

Documentation Index

Fetch the complete documentation index at: https://launchpad.datalumina.com/llms.txt

Use this file to discover all available pages before exploring further.

You now set up an endpoint to receive events. The endpoint validates input, persists the event, and enqueues processing.

​
Endpoint Implementation

The implementation below creates a simple POST endpoint without authentication (for demonstration purposes): 
import json
from http import HTTPStatus

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from starlette.responses import Response

from launchpad.database.event import Event
from launchpad.database.repository import GenericRepository
from launchpad.database.session import db_session
from launchpad.worker.config import celery_app
from launchpad.workflows.workflow_registry import WorkflowRegistry

router = APIRouter()


@router.post("/", dependencies=[])
def handle_event(
    data: dict,
    session: Session = Depends(db_session),
) -> Response:
    repository = GenericRepository(session=session, model=Event)
    raw_event = data.model_dump(mode="json")
    event = Event(data=raw_event, workflow_type=get_workflow_type())
    repository.create(obj=event)

    task_id = celery_app.send_task(
        "process_incoming_event",
        args=[str(event.id)],
    )

    return Response(
        content=json.dumps({"message": f"process_incoming_event started `{task_id}` "}),
        status_code=HTTPStatus.ACCEPTED,
    )


def get_workflow_type() -> str:
    return WorkflowRegistry.QUICKSTART.name
This router is mounted at /events by app/launchpad/api/router.py, so the full path is POST /events/. The body is accepted as a raw dict and validated later by the workflow against CustomerCareEventSchema (the event_schema on CustomerCareWorkflow). This keeps a single endpoint usable for multiple registered workflows.

​
Key Actions Performed

1

Data validation

FastAPI parses the request body as JSON, then stores it. The workflow itself validates against CustomerCareEventSchema when it runs, raising a validation error if the payload is malformed.
2

Database storage

The event is stored with raw JSON, the workflow type from the registry, a generated ID, and timestamps.
3

Task queuing

A Celery task named process_incoming_event is queued with the event ID and returns a task ID for tracking.
4

Response

The endpoint returns HTTP 202 (Accepted) to indicate asynchronous processing.

​
Understanding the Components

  • GenericRepository: Provides database operations (create, read, update, delete) for any model
  • db_session: FastAPI dependency that provides a database session for the request
  • celery_app: The Celery application instance for queuing background tasks
  • WorkflowRegistry: Enum containing all registered workflows in the system

​
Response Status Code

The endpoint returns HTTP 202 (Accepted) rather than 200 (OK) because the workflow processing happens asynchronously, the client is informed that the request is queued, and the actual processing will happen in the background.

​
Security Considerations

Security Note: This example endpoint has no authentication for simplicity. In production, you should add authentication middleware, implement rate limiting, validate API keys or JWT tokens, and use HTTPS for secure communication.