Temporal Integration

from datetime import timedelta
import scale_gp_beta.lib.tracing as tracing
from scale_gp_beta import SGPClient
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker

@activity.defn
async def fetch_data_activity(url: str, trace_id: str, parent_span_id: str) -> dict:
    with tracing.create_span(
        name="fetch_data",
        trace_id=trace_id,
        parent_id=parent_span_id,
        input={"url": url},
    ) as span:
        # ... actual work ...
        result = {"status": "ok", "url": url}
        span.output = result
        return result

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, urls: list[str], trace_id: str, parent_span_id: str) -> list[dict]:
        results = []
        for url in urls:
            result = await workflow.execute_activity(
                fetch_data_activity,
                args=(url, trace_id, parent_span_id),
                start_to_close_timeout=timedelta(seconds=30),
            )
            results.append(result)
        return results

async def run_worker():
    tracing.init(SGPClient(api_key="YOUR_API_KEY", account_id="YOUR_ACCOUNT_ID"))
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="my-queue",
        workflows=[MyWorkflow],
        activities=[fetch_data_activity],
    )
    await worker.run()

import scale_gp_beta.lib.tracing as tracing
from scale_gp_beta import SGPClient
from temporalio.client import Client

# Initialize the SDK once at the caller process entry point. The caller and
# the worker run in different processes, each with its own queue manager, so
# tracing.init() is required in both.
tracing.init(SGPClient(api_key="YOUR_API_KEY", account_id="YOUR_ACCOUNT_ID"))

async def kick_off():
    client = await Client.connect("localhost:7233")
    with tracing.create_trace("my_pipeline") as trace:
        with tracing.create_span("orchestrate_pipeline") as parent:
            await client.execute_workflow(
                MyWorkflow.run,
                args=[["https://a.example", "https://b.example"], trace.trace_id, parent.span_id],
                id="my-workflow-1",
                task_queue="my-queue",
            )

from datetime import timedelta
from scale_gp_beta import AsyncSGPClient
from scale_gp_beta.types.span_upsert_batch_params import Item as SpanRequest
from temporalio import activity, workflow
from temporalio.common import RetryPolicy

# Module-level client so we reuse the connection pool across activity invocations.
_client = AsyncSGPClient(api_key="YOUR_API_KEY", account_id="YOUR_ACCOUNT_ID")

# 4xx-style errors that should not retry: bad input, auth failure, validation.
# 5xx, rate limits, and transport errors are retried by Temporal per the policy below.
_NON_RETRYABLE = [
    "BadRequestError",
    "AuthenticationError",
    "PermissionDeniedError",
    "NotFoundError",
    "UnprocessableEntityError",
]

@activity.defn
async def flush_spans_activity(spans: list[SpanRequest]) -> None:
    # No tracing.init() needed here: this activity bypasses the SDK queue and
    # writes directly via upsert_batch. tracing.init() is only required if you
    # call tracing.create_span(...) inside an activity (Pattern 1).
    await _client.spans.upsert_batch(items=spans)

@workflow.defn
class BufferedTracingWorkflow:
    def __init__(self) -> None:
        self._buffer: list[SpanRequest] = []

    def _enqueue(self, span: SpanRequest) -> None:
        self._buffer.append(span)

    async def _maybe_flush(self, force: bool = False) -> None:
        if not self._buffer:
            return
        if not force and len(self._buffer) < 200:
            return
        spans_to_send = self._buffer
        self._buffer = []
        await workflow.execute_activity(
            flush_spans_activity,
            args=(spans_to_send,),
            start_to_close_timeout=timedelta(seconds=30),
            schedule_to_close_timeout=timedelta(minutes=10),
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(seconds=30),
                maximum_attempts=5,
                non_retryable_error_types=_NON_RETRYABLE,
            ),
        )

    @workflow.run
    async def run(self, items: list[str]) -> None:
        trace_id = workflow.uuid4().hex
        root_span_id = workflow.uuid4().hex
        root: SpanRequest = {
            "id": root_span_id,
            "trace_id": trace_id,
            "name": "workflow_root",
            "start_timestamp": workflow.now().isoformat(),
        }
        self._enqueue(root)

        for item in items:
            child: SpanRequest = {
                "id": workflow.uuid4().hex,
                "trace_id": trace_id,
                "parent_id": root_span_id,
                "name": f"process_{item}",
                "start_timestamp": workflow.now().isoformat(),
            }
            # Real work goes through an activity:
            # await workflow.execute_activity(do_work, item, start_to_close_timeout=...)
            child["end_timestamp"] = workflow.now().isoformat()
            self._enqueue(child)
            await self._maybe_flush()

        # Set root's end_timestamp and re-enqueue so the upsert lands. The
        # upsert_batch endpoint is keyed on `id`, so a second send with
        # end_timestamp set just updates the existing row. This handles the
        # case where root was already flushed mid-loop (buffer hit 200) and
        # the in-memory mutation alone would be lost.
        root["end_timestamp"] = workflow.now().isoformat()
        self._enqueue(root)
        await self._maybe_flush(force=True)