Events

from cellario_cloud_events import AuthenticatedClient

client = AuthenticatedClient(base_url="https://your-instance", token="TOKEN")

import json
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Annotated

import httpx
from hrb_scripting_api.scripting_api import IAsyncScriptingApi
from hrb_scripting_api.scripting_decorators import Main, OutParam
from cellario_cloud_events.models import (
    EventsApiEventsCreateRequest,
    EventsApiEventsEventSourceCategory,
    EventsApiEventsEventSourceComponent,
    EventsApiEventsEventSeverity,
)


@dataclass
class Result:
    event_posted: Annotated[
        bool,
        OutParam(display_name="Event Posted", description="True when the event was created successfully"),
    ]


@Main
async def publish_step_event(api: IAsyncScriptingApi) -> Result:
    body = EventsApiEventsCreateRequest(
        type_="my.step.completed",
        source_name="my-script",
        source_id=api.workflow_order_id,
        source_created_at=datetime.now(timezone.utc),
        source_category=EventsApiEventsEventSourceCategory.CELLARIO_OS,
        source_component=EventsApiEventsEventSourceComponent.OTHER,
        source_component_extended_description="my-script-name",
        stream_id=api.workflow_order_id,
        payload_type="my.step.completed.v1",
        payload=json.dumps({"order_id": api.workflow_order_id, "step_id": api.workflow_order_step_id}),
        severity=EventsApiEventsEventSeverity.INFO,
    )

    try:
        await api.events.events.create(body=body)
    except Exception as exc:
        cause = getattr(exc, "__cause__", None) or exc
        if isinstance(cause, httpx.HTTPStatusError):
            raise RuntimeError(
                f"Events API {cause.response.status_code}: {cause.response.text}"
            ) from cause
        raise

    return Result(event_posted=True)

from cellario_cloud_events.models import (
    EventsApiEventsFileReferencesFileReferenceInput,
    EventsApiEventsFileReferencesFileReferenceType,
)

file_refs = [
    EventsApiEventsFileReferencesFileReferenceInput(
        file_name="report.json",
        description="Step output report",
        type_=EventsApiEventsFileReferencesFileReferenceType.CELLARIO_OS_BLOB,
        link="https://your-instance.cellario.cloud/api/storage/api/v1/buckets/{bucketName}/files/{fileId}",
    ),
]

body = EventsApiEventsCreateRequest(
    ...,
    file_references=file_refs,
)