add-a-source

Add a Source to BandiRadar

A source is the project's one extension point (ARCHITECTURE.md §5). Adding one is a new file + a fixture + a test — no changes to core, the matcher, or storage. The long tail of regional bandi is meant to be crowdsourced this way.

The Source contract

class Source(Protocol):
    id: str                         # unique, e.g. "regione_lazio"
    kind: Literal["tender", "grant", "incentive"]

    def fetch(self, since: datetime | None = None) -> Iterable[RawDoc]: ...
    def to_opportunities(self, raw: RawDoc, now=None) -> list[Opportunity]: ...

• fetch() pulls raw payloads (HTTP/feed/API) and yields RawDocs.
• to_opportunities() is a PURE mapping: raw → canonical Opportunity. No network, no I/O — so it is unit-testable offline against a fixture.

Steps

1. Skeleton adapter — src/bandiradar/sources/<name>.py

"""<Name> source adapter."""

from __future__ import annotations

import json
from collections.abc import Iterable
from datetime import datetime
from pathlib import Path
from typing import Any

from bandiradar.models import Kind, Opportunity, RawDoc, default_status
from bandiradar.sources.base import register

SOURCE_ID = "<name>"
SOURCE_KIND: Kind = "grant"  # or "tender" / "incentive"

# Do NOT invent a live endpoint. Leave empty + a TODO until confirmed.
SOURCE_URL = ""

FIXTURE_PATH = (
    Path(__file__).resolve().parents[3] / "data" / "fixtures" / "<name>.json"
)


def to_opportunities(raw: RawDoc, now: datetime | None = None) -> list[Opportunity]:
    """PURE: map one raw record (raw.payload) into Opportunity objects."""
    record: dict[str, Any] = raw.payload
    deadline = None  # parse from record, e.g. datetime.fromisoformat(...)
    return [
        Opportunity(
            id=f"{SOURCE_ID}:{record['id']}",
            source=SOURCE_ID,
            source_url=record.get("url") or raw.url or "",
            kind=SOURCE_KIND,
            title=record["title"],
            summary=record.get("description"),
            issuer_name=record.get("issuer"),
            issuer_region=record.get("region"),
            cpv=record.get("cpv", []),
            value_amount=record.get("amount"),
            geo_scope="regional" if record.get("region") else "national",
            region=record.get("region"),
            deadline=deadline,
            status=default_status(deadline, now),
            eligibility_text=record.get("eligibility"),
            raw_ref=raw.id,
            # content_hash auto-fills — do not set it.
        )
    ]


def load_fixture(path: Path | None = None) -> list[RawDoc]:
    """Recorded payloads -> RawDocs, for offline use and tests."""
    data = json.loads((path or FIXTURE_PATH).read_text(encoding="utf-8"))
    fetched_at = datetime.fromisoformat("1970-01-01T00:00:00+00:00")
    return [
        RawDoc(
            id=f"{SOURCE_ID}:{rec['id']}",
            source=SOURCE_ID,
            fetched_at=fetched_at,
            payload=rec,
            url=rec.get("url"),
        )
        for rec in data["records"]
    ]


class <Name>Source:
    id = SOURCE_ID
    kind: Kind = SOURCE_KIND

    def fetch(self, since: datetime | None = None) -> Iterable[RawDoc]:
        if not SOURCE_URL:
            raise NotImplementedError(
                "Live fetch not wired: confirm the endpoint, then implement. "
                "Use load_fixture() for offline runs."
            )
        raise NotImplementedError("Live fetch not implemented yet.")

    def to_opportunities(
        self, raw: RawDoc, now: datetime | None = None
    ) -> list[Opportunity]:
        return to_opportunities(raw, now=now)

    def load_fixture(self) -> list[RawDoc]:
        return load_fixture()


register(<Name>Source())