Rapidata signal

RapidataSignal

RapidataSignal(
    openapi_service: OpenAPIService, data: SignalOutput
)

A live handle to a Rapidata signal.

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 (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.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def __init__(
    self,
    openapi_service: OpenAPIService,
    data: SignalOutput,
):
    self._openapi_service = openapi_service
    # 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
    logger.debug("RapidataSignal initialized: %s", self.id)

pause

pause() -> RapidataSignal

Pause the signal. Scheduled jobs stop until meth:resume is called.

Returns:

Name	Type	Description
RapidataSignal	RapidataSignal	self for chaining.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def pause(self) -> RapidataSignal:
    """Pause the signal. Scheduled jobs 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)
        managed_print(f"Signal '{self}' has been paused.")
        return self

resume

resume() -> RapidataSignal

Resume a paused signal. Scheduled jobs start firing again at the configured interval.

Returns:

Name	Type	Description
RapidataSignal	RapidataSignal	self for chaining.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def resume(self) -> RapidataSignal:
    """Resume a paused signal. Scheduled jobs 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
        )
        managed_print(f"Signal '{self}' has been resumed.")
        return self

delete

delete() -> None

Delete the signal. Jobs it already created are unaffected.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def delete(self) -> None:
    """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.")

trigger

trigger() -> None

Trigger the signal manually, creating one extra job outside the schedule.

The job is created asynchronously; use meth:wait_for_next_job to block until it appears.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def trigger(self) -> None:
    """Trigger the signal manually, creating one extra job outside the schedule.

    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)
        self._openapi_service.signal.signal_api.signal_signal_id_trigger_post(
            self.id
        )

update

update(
    *,
    name: str | None = None,
    description: str | None = None,
    interval_hours: float | None = None
) -> RapidataSignal

Update mutable fields on the signal.

Parameters:

Name	Type	Description	Default
name	str | None	New display name. Omit to leave unchanged.	None
description	str | None	New description. Omit to leave unchanged.	None
interval_hours	float | None	New scheduling interval, in hours. Omit to leave unchanged.	None

Returns:

Name	Type	Description
RapidataSignal	RapidataSignal	self for chaining.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def update(
    self,
    *,
    name: str | None = None,
    description: str | 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_hours: New scheduling interval, in hours. Omit to leave unchanged.

    Returns:
        RapidataSignal: ``self`` for chaining.
    """
    if name is None and description is None and interval_hours 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=(
                    None if interval_hours is None else round(interval_hours * 3600)
                ),
            ),
        )
        if name is not None:
            self._name = name
        return self

get_jobs

get_jobs(
    page: int = 1,
    page_size: int = 20,
    sort_descending: bool = True,
) -> 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.

Parameters:

Name	Type	Description	Default
page	int	1-indexed page number.	1
page_size	int	Number of firings to inspect per page.	20
sort_descending	bool	When True (default), newest jobs come first.	True

Returns:

Type	Description
list[RapidataJob]	list[RapidataJob]: The jobs on the requested page.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def get_jobs(
    self,
    page: int = 1,
    page_size: int = 20,
    sort_descending: bool = True,
) -> 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 firings to inspect per page.
        sort_descending: When ``True`` (default), newest jobs come first.

    Returns:
        list[RapidataJob]: The jobs on the requested page.
    """
    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"],
        ).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
        ]

wait_for_next_job

wait_for_next_job(
    timeout: float = 300, poll_interval: float = 5.0
) -> RapidataJob

Block until the signal's next firing creates a job, then return it.

Waits for a firing that started after this call and has spawned a job (skipped firings are ignored). Handy right after meth:trigger. The returned :class:RapidataJob is live — call get_results() or display_progress_bar() on it to follow the job to completion.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum seconds to wait. Raises class:TimeoutError on expiry.	300
poll_interval	float	Seconds between polls.	5.0

Returns:

Name	Type	Description
RapidataJob	RapidataJob	The job created by the next firing.

Raises:

Type	Description
TimeoutError	If no new job is created within timeout seconds.

Source code in src/rapidata/rapidata_client/signal/rapidata_signal.py

def wait_for_next_job(
    self,
    timeout: float = 300,
    poll_interval: float = 5.0,
) -> RapidataJob:
    """Block until the signal's next firing creates a job, then return it.

    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:
        RapidataJob: The job created by the next firing.

    Raises:
        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")

    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 job of signal '%s'",
            timeout,
            self.id,
        )
        while True:
            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 job from signal '{self.id}' within {timeout} seconds."
                )
            sleep(poll_interval)