From 1783aadbc388c2919eed51eac9c608fbb3e7aa37 Mon Sep 17 00:00:00 2001 From: RapidPoseidon Date: Thu, 25 Jun 2026 12:51:38 +0000 Subject: [PATCH 1/6] feat(signal): add SignalManager to Python SDK Add SDK support for the Signal Service (recurring audience-job schedules): - RapidataSignalManager exposed as client.signals (create, get, list, get_run). - RapidataSignal with lifecycle methods (pause, resume, delete, trigger, update), get_runs / get_run, and a blocking wait_for_next_run() helper. - SignalRun dataclass with is_terminal / succeeded convenience properties. Built on the autogenerated SignalApi (the signal service is already part of the generated OpenAPI client on main), wired through the same RapidataApiClient as every other manager, so 4xx/5xx surface as RapidataError. Behaviour follows the real backend contract: create() fetches the full signal after the id-only create response; trigger() returns None since the run is spawned asynchronously (observe it via wait_for_next_run()); SignalRun omits owner/created_at fields the run payload does not carry. Co-Authored-By: Claude Opus 4.8 Co-Authored-By: lino --- src/rapidata/__init__.py | 3 + src/rapidata/rapidata_client/__init__.py | 1 + .../rapidata_client/rapidata_client.py | 10 + .../rapidata_client/signal/__init__.py | 3 + .../rapidata_client/signal/rapidata_signal.py | 294 ++++++++++++++++++ .../signal/rapidata_signal_manager.py | 137 ++++++++ .../rapidata_client/signal/signal_run.py | 71 +++++ src/rapidata/service/openapi_service.py | 9 + src/rapidata/service/services/__init__.py | 2 + .../service/services/signal_service.py | 20 ++ 10 files changed, 550 insertions(+) create mode 100644 src/rapidata/rapidata_client/signal/__init__.py create mode 100644 src/rapidata/rapidata_client/signal/rapidata_signal.py create mode 100644 src/rapidata/rapidata_client/signal/rapidata_signal_manager.py create mode 100644 src/rapidata/rapidata_client/signal/signal_run.py create mode 100644 src/rapidata/service/services/signal_service.py diff --git a/src/rapidata/__init__.py b/src/rapidata/__init__.py index c2fe4d086..3d68ff428 100644 --- a/src/rapidata/__init__.py +++ b/src/rapidata/__init__.py @@ -11,6 +11,9 @@ RapidataJob, RapidataJobDefinition, RapidataJobManager, + RapidataSignal, + RapidataSignalManager, + SignalRun, ValidationSetManager, RapidataValidationSet, Box, diff --git a/src/rapidata/rapidata_client/__init__.py b/src/rapidata/rapidata_client/__init__.py index 07b312c02..4d15e44d2 100644 --- a/src/rapidata/rapidata_client/__init__.py +++ b/src/rapidata/rapidata_client/__init__.py @@ -7,6 +7,7 @@ ) from .order import RapidataOrderManager, RapidataOrder from .job import RapidataJob, RapidataJobDefinition, RapidataJobManager +from .signal import RapidataSignal, RapidataSignalManager, SignalRun from .validation import ValidationSetManager, RapidataValidationSet, Box from .results import RapidataResults from .selection import ( diff --git a/src/rapidata/rapidata_client/rapidata_client.py b/src/rapidata/rapidata_client/rapidata_client.py index e41937b18..2546f50ad 100644 --- a/src/rapidata/rapidata_client/rapidata_client.py +++ b/src/rapidata/rapidata_client/rapidata_client.py @@ -37,6 +37,9 @@ from rapidata.rapidata_client.datapoints._asset_uploader import AssetUploader from rapidata.rapidata_client.job.rapidata_job_manager import RapidataJobManager from rapidata.rapidata_client.flow.rapidata_flow_manager import RapidataFlowManager +from rapidata.rapidata_client.signal.rapidata_signal_manager import ( + RapidataSignalManager, +) from rapidata.rapidata_client.api.rapidata_api_client import ( optional_api_call, mark_sdk_outdated, @@ -108,6 +111,8 @@ def __init__( audience (RapidataAudienceManager): The RapidataAudienceManager instance. job (JobManager): The JobManager instance. mri (RapidataBenchmarkManager): The RapidataBenchmarkManager instance. + signals (RapidataSignalManager): The RapidataSignalManager instance for managing + recurring audience-job schedules (signals) and observing their runs. context (ContextManager): The ContextManager instance for shortening datapoint contexts against a question. """ @@ -165,6 +170,11 @@ def __init__( logger.debug("Initializing RapidataBenchmarkManager") self.mri = RapidataBenchmarkManager(openapi_service=self._openapi_service) + logger.debug("Initializing RapidataSignalManager") + self.signals = RapidataSignalManager( + openapi_service=self._openapi_service + ) + logger.debug("Initializing RapidataAudienceManager") self.audience = RapidataAudienceManager( openapi_service=self._openapi_service diff --git a/src/rapidata/rapidata_client/signal/__init__.py b/src/rapidata/rapidata_client/signal/__init__.py new file mode 100644 index 000000000..4680b31e9 --- /dev/null +++ b/src/rapidata/rapidata_client/signal/__init__.py @@ -0,0 +1,3 @@ +from .rapidata_signal import RapidataSignal +from .rapidata_signal_manager import RapidataSignalManager +from .signal_run import SignalRun diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal.py b/src/rapidata/rapidata_client/signal/rapidata_signal.py new file mode 100644 index 000000000..d5728d278 --- /dev/null +++ b/src/rapidata/rapidata_client/signal/rapidata_signal.py @@ -0,0 +1,294 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from time import monotonic, sleep +from typing import TYPE_CHECKING, Union + +from rapidata.rapidata_client.config import logger, managed_print, tracer +from rapidata.rapidata_client.signal.signal_run import SignalRun + +if TYPE_CHECKING: + from rapidata.api_client.models.get_signal_by_id_endpoint_output import ( + GetSignalByIdEndpointOutput, + ) + from rapidata.api_client.models.query_signals_endpoint_output import ( + QuerySignalsEndpointOutput, + ) + from rapidata.service.openapi_service import OpenAPIService + + SignalOutput = Union[GetSignalByIdEndpointOutput, QuerySignalsEndpointOutput] + + +class RapidataSignal: + """An instance of a Rapidata signal. + + A signal runs an audience job creation on a recurring schedule. Each run produces a + :class:`SignalRun` that wraps the audience job that was created (or an explanation + of why the run was skipped or failed). Use the manager (:py:attr:`RapidataClient.signals`) + to create or look up signals; use the instance methods on this class to manage an + existing signal's lifecycle and observe its runs. + """ + + def __init__( + self, + openapi_service: OpenAPIService, + data: SignalOutput, + ): + self._openapi_service = openapi_service + self._apply(data) + logger.debug("RapidataSignal initialized: %s", self.id) + + def _apply(self, data: SignalOutput) -> None: + """Replace this instance's cached state with the latest API payload.""" + self._id: str = data.id + self._name: str = data.name + self._description: str | None = data.description + self._audience_id: str = data.audience_id + self._job_definition_id: str = data.job_definition_id + self._revision_number: int | None = data.revision_number + self._interval_seconds: int = data.interval_seconds + self._next_run_at: datetime = data.next_run_at + self._last_run_at: datetime | None = data.last_run_at + self._is_paused: bool = data.is_paused + self._is_public: bool = data.is_public + self._created_at: datetime = data.created_at + + @property + def id(self) -> str: + return self._id + + @property + def name(self) -> str: + return self._name + + @property + def description(self) -> str | None: + return self._description + + @property + def audience_id(self) -> str: + return self._audience_id + + @property + def job_definition_id(self) -> str: + return self._job_definition_id + + @property + def revision_number(self) -> int | None: + return self._revision_number + + @property + def interval_seconds(self) -> int: + return self._interval_seconds + + @property + def next_run_at(self) -> datetime: + return self._next_run_at + + @property + def last_run_at(self) -> datetime | None: + return self._last_run_at + + @property + def is_paused(self) -> bool: + return self._is_paused + + @property + def is_public(self) -> bool: + return self._is_public + + @property + def created_at(self) -> datetime: + return self._created_at + + def refresh(self) -> RapidataSignal: + """Re-fetch this signal from the server and update local state. + + Returns: + RapidataSignal: ``self`` for chaining. + """ + with tracer.start_as_current_span("RapidataSignal.refresh"): + data = self._openapi_service.signal.signal_api.signal_signal_id_get(self.id) + self._apply(data) + return self + + def pause(self) -> RapidataSignal: + """Pause the signal. Scheduled runs stop until :py:meth:`resume` is called. + + Returns: + RapidataSignal: ``self`` for chaining. + """ + with tracer.start_as_current_span("RapidataSignal.pause"): + logger.info("Pausing signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_pause_post(self.id) + self.refresh() + managed_print(f"Signal '{self}' has been paused.") + return self + + def resume(self) -> RapidataSignal: + """Resume a paused signal. Scheduled runs start firing again at the configured interval. + + Returns: + RapidataSignal: ``self`` for chaining. + """ + with tracer.start_as_current_span("RapidataSignal.resume"): + logger.info("Resuming signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_resume_post( + self.id + ) + self.refresh() + managed_print(f"Signal '{self}' has been resumed.") + return self + + def delete(self) -> None: + """Delete the signal. Existing runs and their audience jobs are unaffected.""" + with tracer.start_as_current_span("RapidataSignal.delete"): + logger.info("Deleting signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_delete(self.id) + managed_print(f"Signal '{self}' has been deleted.") + + def trigger(self) -> None: + """Trigger the signal manually, creating one extra run outside the schedule. + + The run is created asynchronously; use :py:meth:`wait_for_next_run` to block + until it appears and reaches a terminal status. + """ + with tracer.start_as_current_span("RapidataSignal.trigger"): + logger.info("Triggering signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_trigger_post( + self.id + ) + + def update( + self, + *, + name: str | None = None, + description: str | None = None, + interval_seconds: int | None = None, + ) -> RapidataSignal: + """Update mutable fields on the signal. + + Args: + name: New display name. Omit to leave unchanged. + description: New description. Omit to leave unchanged. + interval_seconds: New scheduling interval. Omit to leave unchanged. + + Returns: + RapidataSignal: ``self``, refreshed with the server's response. + """ + if name is None and description is None and interval_seconds is None: + # Nothing requested — avoid a useless round trip. + return self + + from rapidata.api_client.models.update_signal_endpoint_input import ( + UpdateSignalEndpointInput, + ) + + with tracer.start_as_current_span("RapidataSignal.update"): + self._openapi_service.signal.signal_api.signal_signal_id_patch( + self.id, + UpdateSignalEndpointInput( + name=name, + description=description, + intervalSeconds=interval_seconds, + ), + ) + self.refresh() + return self + + def get_runs( + self, + page: int = 1, + page_size: int = 20, + sort_descending: bool = True, + ) -> list[SignalRun]: + """List historical runs of this signal. + + Args: + page: 1-indexed page number. + page_size: Number of runs per page. + sort_descending: When ``True`` (default), newest runs come first. + + Returns: + list[SignalRun]: The runs on the requested page. + """ + with tracer.start_as_current_span("RapidataSignal.get_runs"): + result = self._openapi_service.signal.signal_api.signal_signal_id_run_get( + self.id, + page=page, + page_size=page_size, + sort=["-started_at" if sort_descending else "started_at"], + ) + return [SignalRun._from_api(item) for item in result.items] + + def get_run(self, run_id: str) -> SignalRun: + """Get a specific run by ID. + + Args: + run_id: The ID of the run to fetch. + + Returns: + SignalRun: The requested run. + """ + with tracer.start_as_current_span("RapidataSignal.get_run"): + data = self._openapi_service.signal.signal_api.signal_run_run_id_get(run_id) + return SignalRun._from_api(data) + + def wait_for_next_run( + self, + timeout: float = 300, + poll_interval: float = 5.0, + ) -> SignalRun: + """Block until the next run of this signal reaches a terminal status. + + Polls the signal's runs and waits for any run that started *after* this call was + made to reach a terminal status (``Completed``, ``Failed``, or ``Skipped``). + Useful after :py:meth:`trigger` or when watching a scheduled run come through. + + Args: + timeout: Maximum seconds to wait. Raises :py:class:`TimeoutError` on expiry. + poll_interval: Seconds between polls. + + Returns: + SignalRun: The first new run to reach a terminal status. + + Raises: + TimeoutError: If no new terminal run is observed within ``timeout`` seconds. + """ + if timeout <= 0: + raise ValueError("timeout must be positive") + if poll_interval <= 0: + raise ValueError("poll_interval must be positive") + + with tracer.start_as_current_span("RapidataSignal.wait_for_next_run"): + started_waiting = datetime.now(timezone.utc) + deadline = monotonic() + timeout + + logger.info( + "Waiting up to %.0fs for the next terminal run of signal '%s'", + timeout, + self.id, + ) + while True: + # Pull the most recent runs and look for any whose started_at is after we + # began waiting AND that has reached a terminal status. We re-look every + # poll because a Pending run may transition while we wait. + runs = self.get_runs(page=1, page_size=20, sort_descending=True) + candidates = [r for r in runs if r.started_at >= started_waiting] + terminal = [r for r in candidates if r.is_terminal] + if terminal: + # `runs` came back descending, so the last terminal in the list is + # the earliest new terminal — return that one (the next to finish). + return terminal[-1] + + if monotonic() >= deadline: + raise TimeoutError( + f"No new terminal run of signal '{self.id}' within {timeout} seconds." + ) + sleep(poll_interval) + + def __str__(self) -> str: + return f"RapidataSignal(name='{self.name}', id='{self.id}')" + + def __repr__(self) -> str: + return self.__str__() diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py new file mode 100644 index 000000000..09763cd0c --- /dev/null +++ b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from rapidata.rapidata_client.config import logger, tracer +from rapidata.rapidata_client.signal.rapidata_signal import RapidataSignal +from rapidata.rapidata_client.signal.signal_run import SignalRun + +if TYPE_CHECKING: + from rapidata.service.openapi_service import OpenAPIService + + +class RapidataSignalManager: + """Manage signals: schedules that periodically create audience jobs. + + A signal binds an audience to a job definition and an interval. The signal service + creates one audience job per interval tick (a :class:`SignalRun`). Use this manager + to create, look up, and list signals; use methods on the returned + :class:`RapidataSignal` to control a specific signal and observe its runs. + + Access this manager via :py:attr:`RapidataClient.signals`. + """ + + def __init__(self, openapi_service: OpenAPIService): + self._openapi_service = openapi_service + logger.debug("RapidataSignalManager initialized") + + def create( + self, + name: str, + audience_id: str, + job_definition_id: str, + interval_seconds: int, + description: str | None = None, + revision_number: int | None = None, + is_public: bool = False, + ) -> RapidataSignal: + """Create a new signal. + + Args: + name: Display name for the signal. + audience_id: ID of the audience the signal targets each run. + job_definition_id: ID of the job definition that backs each run. + interval_seconds: How often the signal fires, in seconds. Minimum 60. + description: Optional human-readable description. + revision_number: Optional explicit revision of ``job_definition_id`` to pin + this signal to. If omitted, the signal follows the latest revision. + is_public: Whether other users can discover and read this signal. + + Returns: + RapidataSignal: The created signal. + """ + if interval_seconds <= 0: + raise ValueError("interval_seconds must be positive") + + from rapidata.api_client.models.create_signal_endpoint_input import ( + CreateSignalEndpointInput, + ) + + with tracer.start_as_current_span("RapidataSignalManager.create"): + logger.debug( + "Creating signal '%s' (audience=%s, job_definition=%s, interval=%ds)", + name, + audience_id, + job_definition_id, + interval_seconds, + ) + created = self._openapi_service.signal.signal_api.signal_post( + CreateSignalEndpointInput( + name=name, + description=description, + audienceId=audience_id, + jobDefinitionId=job_definition_id, + revisionNumber=revision_number, + intervalSeconds=interval_seconds, + isPublic=is_public, + ) + ) + # The create endpoint only echoes the id; fetch the full signal to populate. + return self.get(created.id) + + def get(self, signal_id: str) -> RapidataSignal: + """Fetch a signal by ID. + + Args: + signal_id: The signal to fetch. + + Returns: + RapidataSignal: The signal. + """ + with tracer.start_as_current_span("RapidataSignalManager.get"): + data = self._openapi_service.signal.signal_api.signal_signal_id_get( + signal_id + ) + return RapidataSignal(self._openapi_service, data) + + def list( + self, + page: int = 1, + page_size: int = 50, + ) -> list[RapidataSignal]: + """List signals visible to the caller (the caller's own signals plus public ones). + + Args: + page: 1-indexed page number. + page_size: Number of signals per page. + + Returns: + list[RapidataSignal]: The signals on the requested page. + """ + with tracer.start_as_current_span("RapidataSignalManager.list"): + result = self._openapi_service.signal.signal_api.signal_get( + page=page, + page_size=page_size, + ) + return [ + RapidataSignal(self._openapi_service, item) for item in result.items + ] + + def get_run(self, run_id: str) -> SignalRun: + """Fetch a single signal run by ID, when the signal it belongs to is unknown. + + Args: + run_id: The run ID. + + Returns: + SignalRun: The run. + """ + with tracer.start_as_current_span("RapidataSignalManager.get_run"): + data = self._openapi_service.signal.signal_api.signal_run_run_id_get(run_id) + return SignalRun._from_api(data) + + def __str__(self) -> str: + return "RapidataSignalManager" + + def __repr__(self) -> str: + return self.__str__() diff --git a/src/rapidata/rapidata_client/signal/signal_run.py b/src/rapidata/rapidata_client/signal/signal_run.py new file mode 100644 index 000000000..68cc5ab98 --- /dev/null +++ b/src/rapidata/rapidata_client/signal/signal_run.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import TYPE_CHECKING, Literal, Union + +if TYPE_CHECKING: + from rapidata.api_client.models.get_signal_run_by_id_endpoint_output import ( + GetSignalRunByIdEndpointOutput, + ) + from rapidata.api_client.models.query_signal_runs_endpoint_output import ( + QuerySignalRunsEndpointOutput, + ) + + SignalRunOutput = Union[ + GetSignalRunByIdEndpointOutput, QuerySignalRunsEndpointOutput + ] + +SignalRunStatus = Literal["Pending", "Running", "Completed", "Failed", "Skipped"] +SignalRunTriggerSource = Literal["Scheduled", "Manual"] +_TERMINAL_STATUSES: frozenset[str] = frozenset({"Completed", "Failed", "Skipped"}) + + +@dataclass(frozen=True) +class SignalRun: + """A single execution of a Signal — the result of one audience job creation. + + A run is created either by the signal's scheduled interval or by an explicit + :py:meth:`RapidataSignal.trigger` call. It transitions through statuses until + it reaches a terminal one (``Completed``, ``Failed`` or ``Skipped``). + """ + + id: str + signal_id: str + trigger_source: SignalRunTriggerSource + started_at: datetime + status: SignalRunStatus + audience_job_id: str | None + completed_at: datetime | None + result_file_name: str | None + failure_message: str | None + skipped_reason: str | None + + @property + def is_terminal(self) -> bool: + """True when the run has finished (``Completed``, ``Failed`` or ``Skipped``).""" + return self.status in _TERMINAL_STATUSES + + @property + def succeeded(self) -> bool: + """True when the run finished successfully (``Completed``).""" + return self.status == "Completed" + + @classmethod + def _from_api(cls, model: SignalRunOutput) -> SignalRun: + """Build a SignalRun from a signal-service run model.""" + return cls( + id=model.id, + signal_id=model.signal_id, + trigger_source=model.trigger_source.value, + started_at=model.started_at, + status=model.status.value, + audience_job_id=model.audience_job_id, + completed_at=model.completed_at, + result_file_name=model.result_file_name, + failure_message=model.failure_message, + skipped_reason=model.skipped_reason, + ) + + def __str__(self) -> str: + return f"SignalRun(id='{self.id}', status='{self.status}')" diff --git a/src/rapidata/service/openapi_service.py b/src/rapidata/service/openapi_service.py index d62c04b0e..615408eff 100644 --- a/src/rapidata/service/openapi_service.py +++ b/src/rapidata/service/openapi_service.py @@ -22,6 +22,7 @@ from rapidata.service.services.workflow_service import WorkflowService from rapidata.service.services.leaderboard_service import LeaderboardService from rapidata.service.services.rapid_service import RapidService + from rapidata.service.services.signal_service import SignalService from rapidata.service.services.translation_service import TranslationService @@ -76,6 +77,7 @@ def __init__( self._workflow: WorkflowService | None = None self._leaderboard: LeaderboardService | None = None self._rapid: RapidService | None = None + self._signal: SignalService | None = None self._translation: TranslationService | None = None if token: @@ -213,6 +215,13 @@ def rapid(self) -> RapidService: self._rapid = RapidService(self.api_client) return self._rapid + @property + def signal(self) -> SignalService: + if self._signal is None: + from rapidata.service.services.signal_service import SignalService + self._signal = SignalService(self.api_client) + return self._signal + @property def translation(self) -> TranslationService: if self._translation is None: diff --git a/src/rapidata/service/services/__init__.py b/src/rapidata/service/services/__init__.py index 12c007108..0cf244dd3 100644 --- a/src/rapidata/service/services/__init__.py +++ b/src/rapidata/service/services/__init__.py @@ -7,6 +7,7 @@ from rapidata.service.services.order_service import OrderService from rapidata.service.services.pipeline_service import PipelineService from rapidata.service.services.rapid_service import RapidService +from rapidata.service.services.signal_service import SignalService from rapidata.service.services.validation_service import ValidationService from rapidata.service.services.workflow_service import WorkflowService @@ -20,6 +21,7 @@ "OrderService", "PipelineService", "RapidService", + "SignalService", "ValidationService", "WorkflowService", ] diff --git a/src/rapidata/service/services/signal_service.py b/src/rapidata/service/services/signal_service.py new file mode 100644 index 000000000..00de11327 --- /dev/null +++ b/src/rapidata/service/services/signal_service.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from rapidata.api_client.api.signal_api import SignalApi + from rapidata.rapidata_client.api.rapidata_api_client import RapidataApiClient + + +class SignalService: + def __init__(self, api_client: RapidataApiClient) -> None: + self._api_client = api_client + self._signal_api: SignalApi | None = None + + @property + def signal_api(self) -> SignalApi: + if self._signal_api is None: + from rapidata.api_client.api.signal_api import SignalApi + self._signal_api = SignalApi(self._api_client) + return self._signal_api From f94ec4c2908d57832be281db5d726e1b3aad537e Mon Sep 17 00:00:00 2001 From: RapidPoseidon Date: Thu, 25 Jun 2026 13:03:56 +0000 Subject: [PATCH 2/6] docs(signal): add Signals guide Explain what a signal is (audience + job definition + interval, firing a run each tick), when to use one, and how to create, trigger, read results from, and manage signals. Wire the page into the nav and the llms.txt guide list. Co-Authored-By: Claude Opus 4.8 Co-Authored-By: lino --- docs/signals.md | 179 ++++++++++++++++++++++++++++++++++++++++++++++++ mkdocs.yml | 2 + 2 files changed, 181 insertions(+) create mode 100644 docs/signals.md diff --git a/docs/signals.md b/docs/signals.md new file mode 100644 index 000000000..999797f00 --- /dev/null +++ b/docs/signals.md @@ -0,0 +1,179 @@ +# Signals + +A **signal** runs a labeling job on a repeating schedule. Instead of assigning a +job once and waiting for results, you bind a [job definition](job_definition_parameters.md) +to an [audience](audiences.md) and an interval — and Rapidata creates a fresh +labeling job for you on every tick, automatically. + +## What a signal is + +A signal ties together three things: + +- an **audience** — who labels the data, +- a **job definition** — what they are asked to do (the task, datapoints, settings), +- an **interval** — how often it fires (in seconds, minimum 60). + +Each time the interval elapses, the signal fires and produces a **run** +(`SignalRun`). A run represents one execution: it spawns a single audience job +from the job definition and tracks it through to completion. + +```mermaid +graph LR + S[Signal
audience + job definition + interval] -->|every interval| R1[Run 1 → audience job] + S -->|every interval| R2[Run 2 → audience job] + S -->|every interval| R3[Run 3 → audience job] +``` + +## When to use it + +Reach for a signal whenever you want labeling to happen **continuously and +unattended** rather than as a single one-off job: + +- **Recurring data collection** — re-run the same task on a fresh batch every day/hour. +- **Ongoing model monitoring** — periodically gather human judgments on your model's latest outputs. +- **Scheduled evaluation** — keep a benchmark fed with new human votes over time. + +If you just need labels once, assign a job directly to an audience instead +(see [Custom Audiences](audiences.md)) — you don't need a signal. + +## Creating a signal + +You need an audience id and a job definition id. Both are created the same way +you would for a normal job: + +```py +from rapidata import RapidataClient + +client = RapidataClient() + +audience = client.audience.create_audience(name="Prompt Alignment Audience") + +job_definition = client.job.create_compare_job_definition( + name="Prompt Alignment Job", + instruction="Which image follows the prompt more accurately?", + datapoints=[ + ["https://assets.rapidata.ai/flux_book.jpg", + "https://assets.rapidata.ai/mj_book.jpg"] + ], + contexts=["A small blue book sitting on a large red book."], +) + +signal = client.signals.create( + name="Daily prompt alignment", + audience_id=audience.id, + job_definition_id=job_definition.id, + interval_seconds=86_400, # (1)! +) + +print(signal) +print("First run scheduled for:", signal.next_run_at) +``` + +1. Fires once per day. The minimum interval is 60 seconds. + +By default the signal follows the **latest** revision of the job definition at +fire time. Pin it to a fixed revision with `revision_number=...`, and make it +discoverable by other users in your organization with `is_public=True`. + +## Runs and their lifecycle + +Every firing — scheduled or manual — produces a `SignalRun`. A run moves +through these statuses: + +| Status | Meaning | +|---|---| +| `Pending` | The run has been created but the audience job hasn't started yet. | +| `Running` | The audience job is live and collecting labels. | +| `Completed` | The audience job finished successfully. | +| `Failed` | The run failed (see `failure_message`). | +| `Skipped` | The firing was skipped without creating a job (see `skipped_reason`). | + +`Completed`, `Failed` and `Skipped` are **terminal**. Two convenience +properties make this easy to check: + +```py +run.is_terminal # True once the run has finished (any terminal status) +run.succeeded # True only when the run Completed +``` + +List the runs a signal has produced (newest first by default): + +```py +for run in signal.get_runs(page_size=10): + print(run.started_at, run.status, run.audience_job_id) +``` + +## Triggering a run on demand + +You don't have to wait for the schedule — fire one extra run immediately. This +is the easiest way to test a signal end to end: + +```py +signal.trigger() # (1)! + +run = signal.wait_for_next_run(timeout=600) # (2)! +print("Run finished with status:", run.status) +``` + +1. `trigger()` returns `None` — the run is created asynchronously on the backend. +2. Blocks until the next run (the one you just triggered, or the next scheduled one) reaches a terminal status. Raises `TimeoutError` if none finishes in time. + +## Getting the labeling results of a run + +A run spawns a normal audience job. Once the run is terminal, use its +`audience_job_id` to fetch that job and read its [results](understanding_the_results.md): + +```py +run = signal.wait_for_next_run() + +if run.succeeded and run.audience_job_id: + job = client.job.get_job_by_id(run.audience_job_id) + results = job.get_results() + print(results) +``` + +## Managing a signal + +```py +signal.pause() # stop firing scheduled runs +signal.resume() # resume the schedule + +signal.update( # change mutable fields (omit any you don't want to change) + name="Hourly prompt alignment", + interval_seconds=3_600, +) + +signal.refresh() # re-fetch the latest server-side state +signal.delete() # stop the signal for good (existing runs/jobs are unaffected) +``` + +Look signals up later through the manager: + +```py +signal = client.signals.get("signal_id") # by id +signals = client.signals.list(page_size=20) # your signals + public ones +run = client.signals.get_run("run_id") # a run when you don't know its signal +``` + +## Property reference + +A `RapidataSignal` exposes: + +| Property | Description | +|---|---| +| `id` | The signal's unique id. | +| `name` / `description` | Display name and optional description. | +| `audience_id` | The audience each run targets. | +| `job_definition_id` | The job definition each run is created from. | +| `revision_number` | Pinned job-definition revision, or `None` for "latest at fire time". | +| `interval_seconds` | How often the signal fires. | +| `next_run_at` / `last_run_at` | Timestamps of the next and most recent runs. | +| `is_paused` | Whether the scheduler is currently skipping this signal. | +| `is_public` | Whether other users can discover and read it. | +| `created_at` | When the signal was created. | + +## Next Steps + +- Set up the [audience](audiences.md) that will label each run. +- Choose the task and tune the [job definition parameters](job_definition_parameters.md). +- Learn how to read a run's [results](understanding_the_results.md). diff --git a/mkdocs.yml b/mkdocs.yml index c6dbcf34d..9a3e2117d 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -59,6 +59,7 @@ plugins: - starting_page.md - quickstart.md - audiences.md + - signals.md - job_definition_parameters.md - understanding_the_results.md - error_handling.md @@ -120,6 +121,7 @@ nav: - Overview: starting_page.md - Quick Start: quickstart.md - Custom Audiences: audiences.md + - Signals: signals.md - Parameter Reference: job_definition_parameters.md - Understanding Results: understanding_the_results.md - Early Stopping: confidence_stopping.md From d113ab23329858b4a8afdc7b6f6944bd85fa72fd Mon Sep 17 00:00:00 2001 From: RapidPoseidon Date: Thu, 25 Jun 2026 13:35:09 +0000 Subject: [PATCH 3/6] refactor(signal): interval in hours, align method names with other managers - Express the firing interval in hours: create_signal(interval_hours=...), RapidataSignal.interval_hours, and update(interval_hours=...). Converted to the backend's intervalSeconds internally. - Rename manager methods to match the audience/job managers: create -> create_signal, get -> get_signal_by_id, list -> find_signals(name, amount, page), get_run -> get_run_by_id. - Rename RapidataSignal.get_run -> get_run_by_id. - Update the Signals guide to match. Co-Authored-By: Claude Opus 4.8 Co-Authored-By: lino --- docs/signals.md | 18 +++--- .../rapidata_client/signal/rapidata_signal.py | 18 +++--- .../signal/rapidata_signal_manager.py | 60 +++++++++++-------- 3 files changed, 54 insertions(+), 42 deletions(-) diff --git a/docs/signals.md b/docs/signals.md index 999797f00..d1a11aad8 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -11,7 +11,7 @@ A signal ties together three things: - an **audience** — who labels the data, - a **job definition** — what they are asked to do (the task, datapoints, settings), -- an **interval** — how often it fires (in seconds, minimum 60). +- an **interval** — how often it fires (in hours). Each time the interval elapses, the signal fires and produces a **run** (`SignalRun`). A run represents one execution: it spawns a single audience job @@ -58,18 +58,18 @@ job_definition = client.job.create_compare_job_definition( contexts=["A small blue book sitting on a large red book."], ) -signal = client.signals.create( +signal = client.signals.create_signal( name="Daily prompt alignment", audience_id=audience.id, job_definition_id=job_definition.id, - interval_seconds=86_400, # (1)! + interval_hours=24, # (1)! ) print(signal) print("First run scheduled for:", signal.next_run_at) ``` -1. Fires once per day. The minimum interval is 60 seconds. +1. Fires once per day. By default the signal follows the **latest** revision of the job definition at fire time. Pin it to a fixed revision with `revision_number=...`, and make it @@ -140,7 +140,7 @@ signal.resume() # resume the schedule signal.update( # change mutable fields (omit any you don't want to change) name="Hourly prompt alignment", - interval_seconds=3_600, + interval_hours=1, ) signal.refresh() # re-fetch the latest server-side state @@ -150,9 +150,9 @@ signal.delete() # stop the signal for good (existing runs/jobs are unaffected) Look signals up later through the manager: ```py -signal = client.signals.get("signal_id") # by id -signals = client.signals.list(page_size=20) # your signals + public ones -run = client.signals.get_run("run_id") # a run when you don't know its signal +signal = client.signals.get_signal_by_id("signal_id") # by id +signals = client.signals.find_signals(name="alignment") # your signals + public ones +run = client.signals.get_run_by_id("run_id") # a run when you don't know its signal ``` ## Property reference @@ -166,7 +166,7 @@ A `RapidataSignal` exposes: | `audience_id` | The audience each run targets. | | `job_definition_id` | The job definition each run is created from. | | `revision_number` | Pinned job-definition revision, or `None` for "latest at fire time". | -| `interval_seconds` | How often the signal fires. | +| `interval_hours` | How often the signal fires, in hours. | | `next_run_at` / `last_run_at` | Timestamps of the next and most recent runs. | | `is_paused` | Whether the scheduler is currently skipping this signal. | | `is_public` | Whether other users can discover and read it. | diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal.py b/src/rapidata/rapidata_client/signal/rapidata_signal.py index d5728d278..ea386e99f 100644 --- a/src/rapidata/rapidata_client/signal/rapidata_signal.py +++ b/src/rapidata/rapidata_client/signal/rapidata_signal.py @@ -78,8 +78,8 @@ def revision_number(self) -> int | None: return self._revision_number @property - def interval_seconds(self) -> int: - return self._interval_seconds + def interval_hours(self) -> float: + return self._interval_seconds / 3600 @property def next_run_at(self) -> datetime: @@ -164,19 +164,19 @@ def update( *, name: str | None = None, description: str | None = None, - interval_seconds: int | None = None, + interval_hours: float | None = None, ) -> RapidataSignal: """Update mutable fields on the signal. Args: name: New display name. Omit to leave unchanged. description: New description. Omit to leave unchanged. - interval_seconds: New scheduling interval. Omit to leave unchanged. + interval_hours: New scheduling interval, in hours. Omit to leave unchanged. Returns: RapidataSignal: ``self``, refreshed with the server's response. """ - if name is None and description is None and interval_seconds is None: + if name is None and description is None and interval_hours is None: # Nothing requested — avoid a useless round trip. return self @@ -190,7 +190,9 @@ def update( UpdateSignalEndpointInput( name=name, description=description, - intervalSeconds=interval_seconds, + intervalSeconds=( + None if interval_hours is None else round(interval_hours * 3600) + ), ), ) self.refresh() @@ -221,7 +223,7 @@ def get_runs( ) return [SignalRun._from_api(item) for item in result.items] - def get_run(self, run_id: str) -> SignalRun: + def get_run_by_id(self, run_id: str) -> SignalRun: """Get a specific run by ID. Args: @@ -230,7 +232,7 @@ def get_run(self, run_id: str) -> SignalRun: Returns: SignalRun: The requested run. """ - with tracer.start_as_current_span("RapidataSignal.get_run"): + with tracer.start_as_current_span("RapidataSignal.get_run_by_id"): data = self._openapi_service.signal.signal_api.signal_run_run_id_get(run_id) return SignalRun._from_api(data) diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py index 09763cd0c..ef2c31fd7 100644 --- a/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py +++ b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py @@ -15,7 +15,7 @@ class RapidataSignalManager: A signal binds an audience to a job definition and an interval. The signal service creates one audience job per interval tick (a :class:`SignalRun`). Use this manager - to create, look up, and list signals; use methods on the returned + to create, look up, and find signals; use methods on the returned :class:`RapidataSignal` to control a specific signal and observe its runs. Access this manager via :py:attr:`RapidataClient.signals`. @@ -25,12 +25,12 @@ def __init__(self, openapi_service: OpenAPIService): self._openapi_service = openapi_service logger.debug("RapidataSignalManager initialized") - def create( + def create_signal( self, name: str, audience_id: str, job_definition_id: str, - interval_seconds: int, + interval_hours: float, description: str | None = None, revision_number: int | None = None, is_public: bool = False, @@ -41,7 +41,7 @@ def create( name: Display name for the signal. audience_id: ID of the audience the signal targets each run. job_definition_id: ID of the job definition that backs each run. - interval_seconds: How often the signal fires, in seconds. Minimum 60. + interval_hours: How often the signal fires, in hours. description: Optional human-readable description. revision_number: Optional explicit revision of ``job_definition_id`` to pin this signal to. If omitted, the signal follows the latest revision. @@ -50,20 +50,20 @@ def create( Returns: RapidataSignal: The created signal. """ - if interval_seconds <= 0: - raise ValueError("interval_seconds must be positive") + if interval_hours <= 0: + raise ValueError("interval_hours must be positive") from rapidata.api_client.models.create_signal_endpoint_input import ( CreateSignalEndpointInput, ) - with tracer.start_as_current_span("RapidataSignalManager.create"): + with tracer.start_as_current_span("RapidataSignalManager.create_signal"): logger.debug( - "Creating signal '%s' (audience=%s, job_definition=%s, interval=%ds)", + "Creating signal '%s' (audience=%s, job_definition=%s, interval=%sh)", name, audience_id, job_definition_id, - interval_seconds, + interval_hours, ) created = self._openapi_service.signal.signal_api.signal_post( CreateSignalEndpointInput( @@ -72,15 +72,15 @@ def create( audienceId=audience_id, jobDefinitionId=job_definition_id, revisionNumber=revision_number, - intervalSeconds=interval_seconds, + intervalSeconds=round(interval_hours * 3600), isPublic=is_public, ) ) # The create endpoint only echoes the id; fetch the full signal to populate. - return self.get(created.id) + return self.get_signal_by_id(created.id) - def get(self, signal_id: str) -> RapidataSignal: - """Fetch a signal by ID. + def get_signal_by_id(self, signal_id: str) -> RapidataSignal: + """Get a signal by ID. Args: signal_id: The signal to fetch. @@ -88,37 +88,47 @@ def get(self, signal_id: str) -> RapidataSignal: Returns: RapidataSignal: The signal. """ - with tracer.start_as_current_span("RapidataSignalManager.get"): + with tracer.start_as_current_span("RapidataSignalManager.get_signal_by_id"): data = self._openapi_service.signal.signal_api.signal_signal_id_get( signal_id ) return RapidataSignal(self._openapi_service, data) - def list( + def find_signals( self, + name: str = "", + amount: int = 10, page: int = 1, - page_size: int = 50, ) -> list[RapidataSignal]: - """List signals visible to the caller (the caller's own signals plus public ones). + """Find signals visible to you (your own signals plus public ones). Args: - page: 1-indexed page number. - page_size: Number of signals per page. + name: Filter signals by name (matching signals will contain this string). + Defaults to "" for any signal. + amount: The maximum number of signals to return. Defaults to 10. + page: The page of signals to return. Defaults to 1. Returns: - list[RapidataSignal]: The signals on the requested page. + list[RapidataSignal]: The matching signals. """ - with tracer.start_as_current_span("RapidataSignalManager.list"): + from rapidata.api_client.models.audience_audience_id_jobs_get_job_id_parameter import ( + AudienceAudienceIdJobsGetJobIdParameter, + ) + + with tracer.start_as_current_span("RapidataSignalManager.find_signals"): + logger.debug("Finding signals: %s, %s", name, amount) result = self._openapi_service.signal.signal_api.signal_get( page=page, - page_size=page_size, + page_size=amount, + name=AudienceAudienceIdJobsGetJobIdParameter(contains=name), + sort=["-created_at"], ) return [ RapidataSignal(self._openapi_service, item) for item in result.items ] - def get_run(self, run_id: str) -> SignalRun: - """Fetch a single signal run by ID, when the signal it belongs to is unknown. + def get_run_by_id(self, run_id: str) -> SignalRun: + """Get a single signal run by ID, when the signal it belongs to is unknown. Args: run_id: The run ID. @@ -126,7 +136,7 @@ def get_run(self, run_id: str) -> SignalRun: Returns: SignalRun: The run. """ - with tracer.start_as_current_span("RapidataSignalManager.get_run"): + with tracer.start_as_current_span("RapidataSignalManager.get_run_by_id"): data = self._openapi_service.signal.signal_api.signal_run_run_id_get(run_id) return SignalRun._from_api(data) From 2ac1ad46cf7e828125103f3a53d37e480785ed8e Mon Sep 17 00:00:00 2001 From: RapidPoseidon Date: Thu, 25 Jun 2026 14:09:24 +0000 Subject: [PATCH 4/6] refactor(signal): produce live RapidataJobs, drop SignalRun snapshot Reshape the signal API around the SDK's existing job model instead of a parallel 'run' abstraction: - A signal now yields RapidataJob objects: signal.get_jobs() and signal.wait_for_next_job() return live jobs with the full get_results() / get_status() / display_progress_bar() surface. SignalRun (a frozen snapshot) is removed, along with the run-fetch methods. Skipped firings (no job) simply don't appear. - RapidataSignal is a live handle: identity stays cached, but mutable state (name, is_paused, next_run_at, interval_hours, ...) re-fetches on access, so it never goes stale. refresh() is gone (no longer needed). - create_signal accepts objects or ids: audience=RapidataAudience | RapidataFilteredAudience | str, job_definition=RapidataJobDefinition | str, resolved like create_leaderboard does. - Docs rewritten around jobs. Co-Authored-By: Claude Opus 4.8 Co-Authored-By: lino --- docs/signals.md | 120 +++++----- src/rapidata/__init__.py | 1 - src/rapidata/rapidata_client/__init__.py | 2 +- .../rapidata_client/signal/__init__.py | 1 - .../rapidata_client/signal/rapidata_signal.py | 211 ++++++++---------- .../signal/rapidata_signal_manager.py | 51 +++-- .../rapidata_client/signal/signal_run.py | 71 ------ 7 files changed, 173 insertions(+), 284 deletions(-) delete mode 100644 src/rapidata/rapidata_client/signal/signal_run.py diff --git a/docs/signals.md b/docs/signals.md index d1a11aad8..f5bcd9bf4 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -1,9 +1,9 @@ # Signals A **signal** runs a labeling job on a repeating schedule. Instead of assigning a -job once and waiting for results, you bind a [job definition](job_definition_parameters.md) -to an [audience](audiences.md) and an interval — and Rapidata creates a fresh -labeling job for you on every tick, automatically. +job once, you bind a [job definition](job_definition_parameters.md) to an +[audience](audiences.md) and an interval — and Rapidata creates a fresh +[job](examples/classify_job.md) for you on every tick, automatically. ## What a signal is @@ -13,15 +13,17 @@ A signal ties together three things: - a **job definition** — what they are asked to do (the task, datapoints, settings), - an **interval** — how often it fires (in hours). -Each time the interval elapses, the signal fires and produces a **run** -(`SignalRun`). A run represents one execution: it spawns a single audience job -from the job definition and tracks it through to completion. +Each time the interval elapses, the signal fires and creates one **job** — an +ordinary [`RapidataJob`](understanding_the_results.md), exactly like one you'd +assign by hand, with the same `get_status()`, `get_results()` and +`display_progress_bar()`. A signal is simply a scheduler that keeps producing +jobs for you. ```mermaid graph LR - S[Signal
audience + job definition + interval] -->|every interval| R1[Run 1 → audience job] - S -->|every interval| R2[Run 2 → audience job] - S -->|every interval| R3[Run 3 → audience job] + S[Signal
audience + job definition + interval] -->|every interval| J1[Job 1] + S -->|every interval| J2[Job 2] + S -->|every interval| J3[Job 3] ``` ## When to use it @@ -38,8 +40,8 @@ If you just need labels once, assign a job directly to an audience instead ## Creating a signal -You need an audience id and a job definition id. Both are created the same way -you would for a normal job: +Create an audience and a job definition the same way you would for a normal job, +then hand them to the signal: ```py from rapidata import RapidataClient @@ -60,82 +62,60 @@ job_definition = client.job.create_compare_job_definition( signal = client.signals.create_signal( name="Daily prompt alignment", - audience_id=audience.id, - job_definition_id=job_definition.id, - interval_hours=24, # (1)! + audience=audience, # (1)! + job_definition=job_definition, + interval_hours=24, # (2)! ) print(signal) -print("First run scheduled for:", signal.next_run_at) +print("First job scheduled for:", signal.next_run_at) ``` -1. Fires once per day. +1. Pass the `RapidataAudience` / `RapidataFilteredAudience` and `RapidataJobDefinition` objects directly, or their id strings if that's what you have. +2. Fires once per day. By default the signal follows the **latest** revision of the job definition at fire time. Pin it to a fixed revision with `revision_number=...`, and make it discoverable by other users in your organization with `is_public=True`. -## Runs and their lifecycle +## The jobs a signal creates -Every firing — scheduled or manual — produces a `SignalRun`. A run moves -through these statuses: - -| Status | Meaning | -|---|---| -| `Pending` | The run has been created but the audience job hasn't started yet. | -| `Running` | The audience job is live and collecting labels. | -| `Completed` | The audience job finished successfully. | -| `Failed` | The run failed (see `failure_message`). | -| `Skipped` | The firing was skipped without creating a job (see `skipped_reason`). | - -`Completed`, `Failed` and `Skipped` are **terminal**. Two convenience -properties make this easy to check: +Every firing creates a `RapidataJob`. List the jobs a signal has produced +(newest first by default) and work with them like any other job: ```py -run.is_terminal # True once the run has finished (any terminal status) -run.succeeded # True only when the run Completed -``` +for job in signal.get_jobs(page_size=10): + print(job, job.get_status()) -List the runs a signal has produced (newest first by default): - -```py -for run in signal.get_runs(page_size=10): - print(run.started_at, run.status, run.audience_job_id) +latest = signal.get_jobs(page_size=1)[0] +results = latest.get_results() # blocks until that job is complete ``` -## Triggering a run on demand +!!! note + A firing can occasionally be **skipped** (for example if the previous job + hasn't finished yet) — a skipped firing creates no job and therefore doesn't + appear in `get_jobs()`. + +## Triggering a job on demand -You don't have to wait for the schedule — fire one extra run immediately. This +You don't have to wait for the schedule — fire one extra job immediately. This is the easiest way to test a signal end to end: ```py -signal.trigger() # (1)! +signal.trigger() # (1)! -run = signal.wait_for_next_run(timeout=600) # (2)! -print("Run finished with status:", run.status) +job = signal.wait_for_next_job(timeout=600) # (2)! +job.display_progress_bar() +print(job.get_results()) ``` -1. `trigger()` returns `None` — the run is created asynchronously on the backend. -2. Blocks until the next run (the one you just triggered, or the next scheduled one) reaches a terminal status. Raises `TimeoutError` if none finishes in time. - -## Getting the labeling results of a run - -A run spawns a normal audience job. Once the run is terminal, use its -`audience_job_id` to fetch that job and read its [results](understanding_the_results.md): - -```py -run = signal.wait_for_next_run() - -if run.succeeded and run.audience_job_id: - job = client.job.get_job_by_id(run.audience_job_id) - results = job.get_results() - print(results) -``` +1. `trigger()` returns `None` — the job is created asynchronously on the backend. +2. Blocks until the next firing (the one you just triggered, or the next scheduled one) has created its job, then returns that live `RapidataJob`. Raises `TimeoutError` if none appears in time. ## Managing a signal ```py -signal.pause() # stop firing scheduled runs +signal.pause() # stop firing scheduled jobs signal.resume() # resume the schedule signal.update( # change mutable fields (omit any you don't want to change) @@ -143,37 +123,39 @@ signal.update( # change mutable fields (omit any you don't want to change) interval_hours=1, ) -signal.refresh() # re-fetch the latest server-side state -signal.delete() # stop the signal for good (existing runs/jobs are unaffected) +signal.delete() # stop the signal for good (jobs it already created are unaffected) ``` +A `RapidataSignal` is a **live handle** — reading a property like `is_paused` +or `next_run_at` always reflects the current server state, so there's nothing to +refresh after `pause()`, `update()`, or a scheduled firing. + Look signals up later through the manager: ```py signal = client.signals.get_signal_by_id("signal_id") # by id signals = client.signals.find_signals(name="alignment") # your signals + public ones -run = client.signals.get_run_by_id("run_id") # a run when you don't know its signal ``` ## Property reference -A `RapidataSignal` exposes: +A `RapidataSignal` exposes (mutable properties are re-fetched live on access): | Property | Description | |---|---| | `id` | The signal's unique id. | | `name` / `description` | Display name and optional description. | -| `audience_id` | The audience each run targets. | -| `job_definition_id` | The job definition each run is created from. | +| `audience_id` | The audience each job targets. | +| `job_definition_id` | The job definition each job is created from. | | `revision_number` | Pinned job-definition revision, or `None` for "latest at fire time". | | `interval_hours` | How often the signal fires, in hours. | -| `next_run_at` / `last_run_at` | Timestamps of the next and most recent runs. | +| `next_run_at` / `last_run_at` | Timestamps of the next and most recent firings. | | `is_paused` | Whether the scheduler is currently skipping this signal. | | `is_public` | Whether other users can discover and read it. | | `created_at` | When the signal was created. | ## Next Steps -- Set up the [audience](audiences.md) that will label each run. +- Set up the [audience](audiences.md) that will label each job. - Choose the task and tune the [job definition parameters](job_definition_parameters.md). -- Learn how to read a run's [results](understanding_the_results.md). +- Learn how to read a job's [results](understanding_the_results.md). diff --git a/src/rapidata/__init__.py b/src/rapidata/__init__.py index 3d68ff428..76f2b32ca 100644 --- a/src/rapidata/__init__.py +++ b/src/rapidata/__init__.py @@ -13,7 +13,6 @@ RapidataJobManager, RapidataSignal, RapidataSignalManager, - SignalRun, ValidationSetManager, RapidataValidationSet, Box, diff --git a/src/rapidata/rapidata_client/__init__.py b/src/rapidata/rapidata_client/__init__.py index 4d15e44d2..c375713ba 100644 --- a/src/rapidata/rapidata_client/__init__.py +++ b/src/rapidata/rapidata_client/__init__.py @@ -7,7 +7,7 @@ ) from .order import RapidataOrderManager, RapidataOrder from .job import RapidataJob, RapidataJobDefinition, RapidataJobManager -from .signal import RapidataSignal, RapidataSignalManager, SignalRun +from .signal import RapidataSignal, RapidataSignalManager from .validation import ValidationSetManager, RapidataValidationSet, Box from .results import RapidataResults from .selection import ( diff --git a/src/rapidata/rapidata_client/signal/__init__.py b/src/rapidata/rapidata_client/signal/__init__.py index 4680b31e9..948e8085f 100644 --- a/src/rapidata/rapidata_client/signal/__init__.py +++ b/src/rapidata/rapidata_client/signal/__init__.py @@ -1,3 +1,2 @@ from .rapidata_signal import RapidataSignal from .rapidata_signal_manager import RapidataSignalManager -from .signal_run import SignalRun diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal.py b/src/rapidata/rapidata_client/signal/rapidata_signal.py index ea386e99f..a57e6d879 100644 --- a/src/rapidata/rapidata_client/signal/rapidata_signal.py +++ b/src/rapidata/rapidata_client/signal/rapidata_signal.py @@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Union from rapidata.rapidata_client.config import logger, managed_print, tracer -from rapidata.rapidata_client.signal.signal_run import SignalRun if TYPE_CHECKING: from rapidata.api_client.models.get_signal_by_id_endpoint_output import ( @@ -14,19 +13,24 @@ from rapidata.api_client.models.query_signals_endpoint_output import ( QuerySignalsEndpointOutput, ) + from rapidata.rapidata_client.job.rapidata_job import RapidataJob from rapidata.service.openapi_service import OpenAPIService SignalOutput = Union[GetSignalByIdEndpointOutput, QuerySignalsEndpointOutput] class RapidataSignal: - """An instance of a Rapidata signal. + """A live handle to a Rapidata signal. - A signal runs an audience job creation on a recurring schedule. Each run produces a - :class:`SignalRun` that wraps the audience job that was created (or an explanation - of why the run was skipped or failed). Use the manager (:py:attr:`RapidataClient.signals`) - to create or look up signals; use the instance methods on this class to manage an - existing signal's lifecycle and observe its runs. + A signal creates one audience job on a recurring schedule — every interval tick + spawns a :class:`RapidataJob` built from the signal's job definition and audience. + Use the manager (:py:attr:`RapidataClient.signals`) to create or look up signals; + use the methods here to control a signal and reach the jobs it produces. + + This is a *live* handle: identity fields (``id``, ``audience_id``, + ``job_definition_id``, ``created_at`` …) are fixed at creation, while mutable state + (``name``, ``is_paused``, ``next_run_at`` …) is re-fetched from the server on every + access so it never goes stale. """ def __init__( @@ -35,85 +39,48 @@ def __init__( data: SignalOutput, ): self._openapi_service = openapi_service - self._apply(data) - logger.debug("RapidataSignal initialized: %s", self.id) - - def _apply(self, data: SignalOutput) -> None: - """Replace this instance's cached state with the latest API payload.""" - self._id: str = data.id + # Identity / immutable fields — safe to cache for the object's lifetime. + self.id: str = data.id + self.audience_id: str = data.audience_id + self.job_definition_id: str = data.job_definition_id + self.revision_number: int | None = data.revision_number + self.is_public: bool = data.is_public + self.created_at: datetime = data.created_at + # Last-known name, kept only as a display label for __str__/logging; the `name` + # property always reads the live value. self._name: str = data.name - self._description: str | None = data.description - self._audience_id: str = data.audience_id - self._job_definition_id: str = data.job_definition_id - self._revision_number: int | None = data.revision_number - self._interval_seconds: int = data.interval_seconds - self._next_run_at: datetime = data.next_run_at - self._last_run_at: datetime | None = data.last_run_at - self._is_paused: bool = data.is_paused - self._is_public: bool = data.is_public - self._created_at: datetime = data.created_at + logger.debug("RapidataSignal initialized: %s", self.id) - @property - def id(self) -> str: - return self._id + def _latest(self) -> GetSignalByIdEndpointOutput: + """Fetch the current server-side state of this signal.""" + return self._openapi_service.signal.signal_api.signal_signal_id_get(self.id) @property def name(self) -> str: - return self._name + return self._latest().name @property def description(self) -> str | None: - return self._description - - @property - def audience_id(self) -> str: - return self._audience_id - - @property - def job_definition_id(self) -> str: - return self._job_definition_id - - @property - def revision_number(self) -> int | None: - return self._revision_number + return self._latest().description @property def interval_hours(self) -> float: - return self._interval_seconds / 3600 + return self._latest().interval_seconds / 3600 @property def next_run_at(self) -> datetime: - return self._next_run_at + return self._latest().next_run_at @property def last_run_at(self) -> datetime | None: - return self._last_run_at + return self._latest().last_run_at @property def is_paused(self) -> bool: - return self._is_paused - - @property - def is_public(self) -> bool: - return self._is_public - - @property - def created_at(self) -> datetime: - return self._created_at - - def refresh(self) -> RapidataSignal: - """Re-fetch this signal from the server and update local state. - - Returns: - RapidataSignal: ``self`` for chaining. - """ - with tracer.start_as_current_span("RapidataSignal.refresh"): - data = self._openapi_service.signal.signal_api.signal_signal_id_get(self.id) - self._apply(data) - return self + return self._latest().is_paused def pause(self) -> RapidataSignal: - """Pause the signal. Scheduled runs stop until :py:meth:`resume` is called. + """Pause the signal. Scheduled jobs stop until :py:meth:`resume` is called. Returns: RapidataSignal: ``self`` for chaining. @@ -121,12 +88,11 @@ def pause(self) -> RapidataSignal: with tracer.start_as_current_span("RapidataSignal.pause"): logger.info("Pausing signal '%s'", self.id) self._openapi_service.signal.signal_api.signal_signal_id_pause_post(self.id) - self.refresh() managed_print(f"Signal '{self}' has been paused.") return self def resume(self) -> RapidataSignal: - """Resume a paused signal. Scheduled runs start firing again at the configured interval. + """Resume a paused signal. Scheduled jobs start firing again at the configured interval. Returns: RapidataSignal: ``self`` for chaining. @@ -136,22 +102,21 @@ def resume(self) -> RapidataSignal: self._openapi_service.signal.signal_api.signal_signal_id_resume_post( self.id ) - self.refresh() managed_print(f"Signal '{self}' has been resumed.") return self def delete(self) -> None: - """Delete the signal. Existing runs and their audience jobs are unaffected.""" + """Delete the signal. Jobs it already created are unaffected.""" with tracer.start_as_current_span("RapidataSignal.delete"): logger.info("Deleting signal '%s'", self.id) self._openapi_service.signal.signal_api.signal_signal_id_delete(self.id) managed_print(f"Signal '{self}' has been deleted.") def trigger(self) -> None: - """Trigger the signal manually, creating one extra run outside the schedule. + """Trigger the signal manually, creating one extra job outside the schedule. - The run is created asynchronously; use :py:meth:`wait_for_next_run` to block - until it appears and reaches a terminal status. + The job is created asynchronously; use :py:meth:`wait_for_next_job` to block + until it appears. """ with tracer.start_as_current_span("RapidataSignal.trigger"): logger.info("Triggering signal '%s'", self.id) @@ -174,7 +139,7 @@ def update( interval_hours: New scheduling interval, in hours. Omit to leave unchanged. Returns: - RapidataSignal: ``self``, refreshed with the server's response. + RapidataSignal: ``self`` for chaining. """ if name is None and description is None and interval_hours is None: # Nothing requested — avoid a useless round trip. @@ -195,102 +160,110 @@ def update( ), ), ) - self.refresh() + if name is not None: + self._name = name return self - def get_runs( + def get_jobs( self, page: int = 1, page_size: int = 20, sort_descending: bool = True, - ) -> list[SignalRun]: - """List historical runs of this signal. + ) -> list[RapidataJob]: + """List the jobs this signal has created (newest first by default). + + Each scheduled or manual firing that spawned a job is returned as a live + :class:`RapidataJob`. Firings that were skipped without creating a job are + not included. Args: page: 1-indexed page number. - page_size: Number of runs per page. - sort_descending: When ``True`` (default), newest runs come first. + page_size: Number of firings to inspect per page. + sort_descending: When ``True`` (default), newest jobs come first. Returns: - list[SignalRun]: The runs on the requested page. + list[RapidataJob]: The jobs on the requested page. """ - with tracer.start_as_current_span("RapidataSignal.get_runs"): - result = self._openapi_service.signal.signal_api.signal_signal_id_run_get( + from rapidata.rapidata_client.job.rapidata_job_manager import ( + RapidataJobManager, + ) + + with tracer.start_as_current_span("RapidataSignal.get_jobs"): + runs = self._openapi_service.signal.signal_api.signal_signal_id_run_get( self.id, page=page, page_size=page_size, sort=["-started_at" if sort_descending else "started_at"], - ) - return [SignalRun._from_api(item) for item in result.items] - - def get_run_by_id(self, run_id: str) -> SignalRun: - """Get a specific run by ID. - - Args: - run_id: The ID of the run to fetch. - - Returns: - SignalRun: The requested run. - """ - with tracer.start_as_current_span("RapidataSignal.get_run_by_id"): - data = self._openapi_service.signal.signal_api.signal_run_run_id_get(run_id) - return SignalRun._from_api(data) - - def wait_for_next_run( + ).items + job_manager = RapidataJobManager(openapi_service=self._openapi_service) + return [ + job_manager.get_job_by_id(run.audience_job_id) + for run in runs + if run.audience_job_id + ] + + def wait_for_next_job( self, timeout: float = 300, poll_interval: float = 5.0, - ) -> SignalRun: - """Block until the next run of this signal reaches a terminal status. + ) -> RapidataJob: + """Block until the signal's next firing creates a job, then return it. - Polls the signal's runs and waits for any run that started *after* this call was - made to reach a terminal status (``Completed``, ``Failed``, or ``Skipped``). - Useful after :py:meth:`trigger` or when watching a scheduled run come through. + Waits for a firing that started *after* this call and has spawned a job + (skipped firings are ignored). Handy right after :py:meth:`trigger`. The + returned :class:`RapidataJob` is live — call ``get_results()`` or + ``display_progress_bar()`` on it to follow the job to completion. Args: timeout: Maximum seconds to wait. Raises :py:class:`TimeoutError` on expiry. poll_interval: Seconds between polls. Returns: - SignalRun: The first new run to reach a terminal status. + RapidataJob: The job created by the next firing. Raises: - TimeoutError: If no new terminal run is observed within ``timeout`` seconds. + TimeoutError: If no new job is created within ``timeout`` seconds. """ if timeout <= 0: raise ValueError("timeout must be positive") if poll_interval <= 0: raise ValueError("poll_interval must be positive") - with tracer.start_as_current_span("RapidataSignal.wait_for_next_run"): + from rapidata.rapidata_client.job.rapidata_job_manager import ( + RapidataJobManager, + ) + + with tracer.start_as_current_span("RapidataSignal.wait_for_next_job"): started_waiting = datetime.now(timezone.utc) deadline = monotonic() + timeout + job_manager = RapidataJobManager(openapi_service=self._openapi_service) logger.info( - "Waiting up to %.0fs for the next terminal run of signal '%s'", + "Waiting up to %.0fs for the next job of signal '%s'", timeout, self.id, ) while True: - # Pull the most recent runs and look for any whose started_at is after we - # began waiting AND that has reached a terminal status. We re-look every - # poll because a Pending run may transition while we wait. - runs = self.get_runs(page=1, page_size=20, sort_descending=True) - candidates = [r for r in runs if r.started_at >= started_waiting] - terminal = [r for r in candidates if r.is_terminal] - if terminal: - # `runs` came back descending, so the last terminal in the list is - # the earliest new terminal — return that one (the next to finish). - return terminal[-1] + runs = self._openapi_service.signal.signal_api.signal_signal_id_run_get( + self.id, page=1, page_size=20, sort=["-started_at"] + ).items + # Newest-first, so the last match is the earliest new firing with a job. + fresh_job_ids = [ + run.audience_job_id + for run in runs + if run.started_at >= started_waiting and run.audience_job_id + ] + if fresh_job_ids: + return job_manager.get_job_by_id(fresh_job_ids[-1]) if monotonic() >= deadline: raise TimeoutError( - f"No new terminal run of signal '{self.id}' within {timeout} seconds." + f"No new job from signal '{self.id}' within {timeout} seconds." ) sleep(poll_interval) def __str__(self) -> str: - return f"RapidataSignal(name='{self.name}', id='{self.id}')" + return f"RapidataSignal(name='{self._name}', id='{self.id}')" def __repr__(self) -> str: return self.__str__() diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py index ef2c31fd7..128c86c6b 100644 --- a/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py +++ b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py @@ -4,9 +4,12 @@ from rapidata.rapidata_client.config import logger, tracer from rapidata.rapidata_client.signal.rapidata_signal import RapidataSignal -from rapidata.rapidata_client.signal.signal_run import SignalRun if TYPE_CHECKING: + from rapidata.rapidata_client.audience._audience_base import RapidataAudienceBase + from rapidata.rapidata_client.job.rapidata_job_definition import ( + RapidataJobDefinition, + ) from rapidata.service.openapi_service import OpenAPIService @@ -14,9 +17,9 @@ class RapidataSignalManager: """Manage signals: schedules that periodically create audience jobs. A signal binds an audience to a job definition and an interval. The signal service - creates one audience job per interval tick (a :class:`SignalRun`). Use this manager - to create, look up, and find signals; use methods on the returned - :class:`RapidataSignal` to control a specific signal and observe its runs. + creates one audience job per interval tick. Use this manager to create, look up, + and find signals; use methods on the returned :class:`RapidataSignal` to control a + specific signal and reach the jobs it produces. Access this manager via :py:attr:`RapidataClient.signals`. """ @@ -28,8 +31,8 @@ def __init__(self, openapi_service: OpenAPIService): def create_signal( self, name: str, - audience_id: str, - job_definition_id: str, + audience: str | RapidataAudienceBase, + job_definition: str | RapidataJobDefinition, interval_hours: float, description: str | None = None, revision_number: int | None = None, @@ -39,11 +42,13 @@ def create_signal( Args: name: Display name for the signal. - audience_id: ID of the audience the signal targets each run. - job_definition_id: ID of the job definition that backs each run. + audience: The audience each job targets — a :class:`RapidataAudience`, + a :class:`RapidataFilteredAudience`, or an audience id. + job_definition: The job definition each job is created from — a + :class:`RapidataJobDefinition` or a job definition id. interval_hours: How often the signal fires, in hours. description: Optional human-readable description. - revision_number: Optional explicit revision of ``job_definition_id`` to pin + revision_number: Optional explicit revision of ``job_definition`` to pin this signal to. If omitted, the signal follows the latest revision. is_public: Whether other users can discover and read this signal. @@ -53,10 +58,25 @@ def create_signal( if interval_hours <= 0: raise ValueError("interval_hours must be positive") + from rapidata.rapidata_client.audience._audience_base import ( + RapidataAudienceBase, + ) + from rapidata.rapidata_client.job.rapidata_job_definition import ( + RapidataJobDefinition, + ) from rapidata.api_client.models.create_signal_endpoint_input import ( CreateSignalEndpointInput, ) + audience_id = ( + audience.id if isinstance(audience, RapidataAudienceBase) else audience + ) + job_definition_id = ( + job_definition.id + if isinstance(job_definition, RapidataJobDefinition) + else job_definition + ) + with tracer.start_as_current_span("RapidataSignalManager.create_signal"): logger.debug( "Creating signal '%s' (audience=%s, job_definition=%s, interval=%sh)", @@ -127,19 +147,6 @@ def find_signals( RapidataSignal(self._openapi_service, item) for item in result.items ] - def get_run_by_id(self, run_id: str) -> SignalRun: - """Get a single signal run by ID, when the signal it belongs to is unknown. - - Args: - run_id: The run ID. - - Returns: - SignalRun: The run. - """ - with tracer.start_as_current_span("RapidataSignalManager.get_run_by_id"): - data = self._openapi_service.signal.signal_api.signal_run_run_id_get(run_id) - return SignalRun._from_api(data) - def __str__(self) -> str: return "RapidataSignalManager" diff --git a/src/rapidata/rapidata_client/signal/signal_run.py b/src/rapidata/rapidata_client/signal/signal_run.py deleted file mode 100644 index 68cc5ab98..000000000 --- a/src/rapidata/rapidata_client/signal/signal_run.py +++ /dev/null @@ -1,71 +0,0 @@ -from __future__ import annotations - -from dataclasses import dataclass -from datetime import datetime -from typing import TYPE_CHECKING, Literal, Union - -if TYPE_CHECKING: - from rapidata.api_client.models.get_signal_run_by_id_endpoint_output import ( - GetSignalRunByIdEndpointOutput, - ) - from rapidata.api_client.models.query_signal_runs_endpoint_output import ( - QuerySignalRunsEndpointOutput, - ) - - SignalRunOutput = Union[ - GetSignalRunByIdEndpointOutput, QuerySignalRunsEndpointOutput - ] - -SignalRunStatus = Literal["Pending", "Running", "Completed", "Failed", "Skipped"] -SignalRunTriggerSource = Literal["Scheduled", "Manual"] -_TERMINAL_STATUSES: frozenset[str] = frozenset({"Completed", "Failed", "Skipped"}) - - -@dataclass(frozen=True) -class SignalRun: - """A single execution of a Signal — the result of one audience job creation. - - A run is created either by the signal's scheduled interval or by an explicit - :py:meth:`RapidataSignal.trigger` call. It transitions through statuses until - it reaches a terminal one (``Completed``, ``Failed`` or ``Skipped``). - """ - - id: str - signal_id: str - trigger_source: SignalRunTriggerSource - started_at: datetime - status: SignalRunStatus - audience_job_id: str | None - completed_at: datetime | None - result_file_name: str | None - failure_message: str | None - skipped_reason: str | None - - @property - def is_terminal(self) -> bool: - """True when the run has finished (``Completed``, ``Failed`` or ``Skipped``).""" - return self.status in _TERMINAL_STATUSES - - @property - def succeeded(self) -> bool: - """True when the run finished successfully (``Completed``).""" - return self.status == "Completed" - - @classmethod - def _from_api(cls, model: SignalRunOutput) -> SignalRun: - """Build a SignalRun from a signal-service run model.""" - return cls( - id=model.id, - signal_id=model.signal_id, - trigger_source=model.trigger_source.value, - started_at=model.started_at, - status=model.status.value, - audience_job_id=model.audience_job_id, - completed_at=model.completed_at, - result_file_name=model.result_file_name, - failure_message=model.failure_message, - skipped_reason=model.skipped_reason, - ) - - def __str__(self) -> str: - return f"SignalRun(id='{self.id}', status='{self.status}')" From 46cd625d2d03e692dae6e2cfaf9c0e9ced8ce923 Mon Sep 17 00:00:00 2001 From: RapidPoseidon Date: Thu, 25 Jun 2026 14:58:02 +0000 Subject: [PATCH 5/6] docs(signal): fetch audience by id, drop speculative use cases, trim prose MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Use client.audience.get_audience_by_id(...) in the example instead of creating a fresh (empty) audience. - Remove the 'when to use it' use cases (model monitoring / scheduled evaluation) — a signal re-runs the same job definition on the same datapoints, so those framings were inaccurate. - Cut over-explained / obvious wording. Co-Authored-By: Claude Opus 4.8 Co-Authored-By: lino --- docs/signals.md | 112 +++++++++++++++--------------------------------- 1 file changed, 35 insertions(+), 77 deletions(-) diff --git a/docs/signals.md b/docs/signals.md index f5bcd9bf4..8f519b60c 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -1,23 +1,21 @@ # Signals -A **signal** runs a labeling job on a repeating schedule. Instead of assigning a -job once, you bind a [job definition](job_definition_parameters.md) to an -[audience](audiences.md) and an interval — and Rapidata creates a fresh -[job](examples/classify_job.md) for you on every tick, automatically. +A **signal** runs the same labeling job on a repeating schedule: bind a +[job definition](job_definition_parameters.md) to an [audience](audiences.md) +and an interval, and Rapidata creates a new [job](understanding_the_results.md) +on every tick. ## What a signal is A signal ties together three things: - an **audience** — who labels the data, -- a **job definition** — what they are asked to do (the task, datapoints, settings), -- an **interval** — how often it fires (in hours). +- a **job definition** — the task that gets run, +- an **interval** — how often it fires, in hours. -Each time the interval elapses, the signal fires and creates one **job** — an -ordinary [`RapidataJob`](understanding_the_results.md), exactly like one you'd -assign by hand, with the same `get_status()`, `get_results()` and -`display_progress_bar()`. A signal is simply a scheduler that keeps producing -jobs for you. +Each firing creates one `RapidataJob`, identical to a job you create directly. +A signal is just a scheduler that keeps producing those jobs; every job runs the +same job definition against the same audience. ```mermaid graph LR @@ -26,29 +24,14 @@ graph LR S -->|every interval| J3[Job 3] ``` -## When to use it - -Reach for a signal whenever you want labeling to happen **continuously and -unattended** rather than as a single one-off job: - -- **Recurring data collection** — re-run the same task on a fresh batch every day/hour. -- **Ongoing model monitoring** — periodically gather human judgments on your model's latest outputs. -- **Scheduled evaluation** — keep a benchmark fed with new human votes over time. - -If you just need labels once, assign a job directly to an audience instead -(see [Custom Audiences](audiences.md)) — you don't need a signal. - ## Creating a signal -Create an audience and a job definition the same way you would for a normal job, -then hand them to the signal: - ```py from rapidata import RapidataClient client = RapidataClient() -audience = client.audience.create_audience(name="Prompt Alignment Audience") +audience = client.audience.get_audience_by_id("aud_MU1GZYoESyO") job_definition = client.job.create_compare_job_definition( name="Prompt Alignment Job", @@ -62,85 +45,66 @@ job_definition = client.job.create_compare_job_definition( signal = client.signals.create_signal( name="Daily prompt alignment", - audience=audience, # (1)! + audience=audience, job_definition=job_definition, - interval_hours=24, # (2)! + interval_hours=24, ) - -print(signal) -print("First job scheduled for:", signal.next_run_at) ``` -1. Pass the `RapidataAudience` / `RapidataFilteredAudience` and `RapidataJobDefinition` objects directly, or their id strings if that's what you have. -2. Fires once per day. - -By default the signal follows the **latest** revision of the job definition at -fire time. Pin it to a fixed revision with `revision_number=...`, and make it -discoverable by other users in your organization with `is_public=True`. +`audience` and `job_definition` also accept id strings. By default the signal +uses the latest revision of the job definition at fire time; pin one with +`revision_number=...`. Set `is_public=True` to let others in your organization +read the signal. ## The jobs a signal creates -Every firing creates a `RapidataJob`. List the jobs a signal has produced -(newest first by default) and work with them like any other job: +Every firing creates a `RapidataJob`: ```py for job in signal.get_jobs(page_size=10): print(job, job.get_status()) -latest = signal.get_jobs(page_size=1)[0] -results = latest.get_results() # blocks until that job is complete +results = signal.get_jobs(page_size=1)[0].get_results() ``` -!!! note - A firing can occasionally be **skipped** (for example if the previous job - hasn't finished yet) — a skipped firing creates no job and therefore doesn't - appear in `get_jobs()`. +A firing can be skipped (for example if the previous job hasn't finished). A +skipped firing creates no job and won't appear in `get_jobs()`. ## Triggering a job on demand -You don't have to wait for the schedule — fire one extra job immediately. This -is the easiest way to test a signal end to end: +Fire one job immediately instead of waiting for the schedule: ```py -signal.trigger() # (1)! - -job = signal.wait_for_next_job(timeout=600) # (2)! -job.display_progress_bar() +signal.trigger() +job = signal.wait_for_next_job(timeout=600) print(job.get_results()) ``` -1. `trigger()` returns `None` — the job is created asynchronously on the backend. -2. Blocks until the next firing (the one you just triggered, or the next scheduled one) has created its job, then returns that live `RapidataJob`. Raises `TimeoutError` if none appears in time. +`trigger()` returns right away; the job is created in the background. +`wait_for_next_job()` blocks until the next firing has created its job and +returns it. ## Managing a signal ```py -signal.pause() # stop firing scheduled jobs -signal.resume() # resume the schedule - -signal.update( # change mutable fields (omit any you don't want to change) - name="Hourly prompt alignment", - interval_hours=1, -) - -signal.delete() # stop the signal for good (jobs it already created are unaffected) +signal.pause() +signal.resume() +signal.update(name="Hourly prompt alignment", interval_hours=1) +signal.delete() ``` -A `RapidataSignal` is a **live handle** — reading a property like `is_paused` -or `next_run_at` always reflects the current server state, so there's nothing to -refresh after `pause()`, `update()`, or a scheduled firing. +Reading a property re-fetches the current server state, so there's nothing to +refresh after a change or a scheduled firing. -Look signals up later through the manager: +Look signals up later: ```py -signal = client.signals.get_signal_by_id("signal_id") # by id -signals = client.signals.find_signals(name="alignment") # your signals + public ones +signal = client.signals.get_signal_by_id("signal_id") +signals = client.signals.find_signals(name="alignment") ``` ## Property reference -A `RapidataSignal` exposes (mutable properties are re-fetched live on access): - | Property | Description | |---|---| | `id` | The signal's unique id. | @@ -153,9 +117,3 @@ A `RapidataSignal` exposes (mutable properties are re-fetched live on access): | `is_paused` | Whether the scheduler is currently skipping this signal. | | `is_public` | Whether other users can discover and read it. | | `created_at` | When the signal was created. | - -## Next Steps - -- Set up the [audience](audiences.md) that will label each job. -- Choose the task and tune the [job definition parameters](job_definition_parameters.md). -- Learn how to read a job's [results](understanding_the_results.md). From 53bd28cdf77b304a809bd729cedee30d63369c1a Mon Sep 17 00:00:00 2001 From: Lino Giger <68745352+LinoGiger@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:02:42 +0200 Subject: [PATCH 6/6] small fix --- docs/signals.md | 3 --- uv.lock | 2 +- 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/docs/signals.md b/docs/signals.md index 8f519b60c..4109ce00f 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -93,9 +93,6 @@ signal.update(name="Hourly prompt alignment", interval_hours=1) signal.delete() ``` -Reading a property re-fetches the current server state, so there's nothing to -refresh after a change or a scheduled firing. - Look signals up later: ```py diff --git a/uv.lock b/uv.lock index 23c812895..9f10b5e1b 100644 --- a/uv.lock +++ b/uv.lock @@ -2570,7 +2570,7 @@ wheels = [ [[package]] name = "rapidata" -version = "3.15.0" +version = "3.15.1" source = { editable = "." } dependencies = [ { name = "authlib" },