yksa-source-poller

The machine that polls upstreams on a schedule: a source model, a run log, the dispatcher and the bookkeeping around one attempt. The cargo — what a fetched record is and what happens to it — stays in the service.

Extracted from tdas/yksa_tdas/sources/ and tle/yksa_tle/sources/, which were the same ~500 lines with different payload types.

Install

yksa-source-poller @ git+https://git.intra.yksa.space/web/yksa-source-poller.git@v0.1.0

Requires yksa-django-kit for RunStatus and read_secret. It is not added to INSTALLED_APPS: every model in it is abstract, so it ships no tables and needs no app config.

Use

# sources/models.py
class TelemetrySource(AbstractPollingSource):
    SUBSCRIPTION_ID_FIELDS = {"norad": "norad_cat_id", "satnogs": "satnogs_id"}

    kind = models.CharField(max_length=32, choices=SourceKind.choices)
    subscription_id_kind = models.CharField(..., default=SubscriptionIdKind.SATNOGS)
    tracked_satellites = models.ManyToManyField("satellites.Satellite", blank=True)


class SourceRun(AbstractSourceRun):
    source = models.ForeignKey(TelemetrySource, on_delete=models.CASCADE,
                               related_name="runs")

# sources/tasks.py
POLLER = Poller(
    source_model=TelemetrySource,
    run_model=SourceRun,
    adapters=ADAPTERS,
    persist=_persist_frame,          # -> True when the record is new
    track=_track_latest_rx,          # optional: extra source fields on success
    unpollable_kinds=(SourceKind.INTERNAL_PUSH,),
)


@shared_task(bind=True, max_retries=3, default_retry_delay=120)
def run_source_task(self, source_id):
    try:
        return run_source(POLLER, source_id)
    except Exception as exc:
        raise self.retry(exc=exc)

The Celery tasks stay in the service on purpose: CELERY_BEAT_SCHEDULE names them by dotted path, and a task that moved into this package would silently stop being scheduled.

Why the models are abstract

The two concrete source models differ where they have to — which satellite model the M2M points at, which identifiers the upstream accepts, and (in tdas) a last_rx_at watermark. Everything else was identical. Abstract bases share the identical part without pretending the rest is the same.

Field definitions here reproduce what the services already had, so adopting the package needs no migration. Where a service's own definition differed — a different poll_interval_sec default, different wording — the service overrides the field, which Django allows for abstract inheritance. Check with manage.py makemigrations --check --dry-run after wiring it up.

What the runner guarantees

Every exit path — success, failure, unconfigured adapter, missing adapter — writes last_status, updates a counter and re-arms next_poll_at. That is the reason to share it: a path that forgets leaves the dispatcher either hammering the source once a minute or ignoring it forever, and both shipped once.

NotImplementedError from an adapter is a skip, not a failure: a half-configured source must not look like an outage.

Tests

pip install -e ".[test]" ../yksa-django-kit
python -m pytest