feat: implement continuous polling loop

start() now loops indefinitely, calling poll() every 5 minutes.
Each cycle is wrapped in a poll_cycle OTel span with a
flights.published attribute (0 on failure). Transient poll failures
are logged and the loop continues. The Kafka stream engine is stopped
cleanly via try/finally on shutdown.

Closes #5

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

.gitea/workflows/lint.yaml

@@ -0,0 +1,25 @@
+name: Lint and Format
+
+on:
+  pull_request:
+    branches:
+      - main
+  push:
+    branches:
+      - main
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+
+      - name: Check formatting with ruff
+        run: uvx ruff format --check src/
+
+      - name: Lint with ruff
+        run: uvx ruff check src/

pyproject.toml

@@ -19,3 +19,37 @@ dependencies = [
     "opentelemetry-exporter-otlp>=1.27.0",
     "opentelemetry-instrumentation-requests>=0.48b0",
 ]
+
+[project.scripts]
+schiphol = "src.main:main"
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0",
+    "pytest-asyncio>=0.24",
+]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+pythonpath = ["src"]
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["src*"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py314"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "SIM"]
+ignore = []
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"

src/__init__.py

@@ -0,0 +1 @@
+"""Schiphol flight data producer."""

src/main.py

@@ -1,53 +1,36 @@
-from settings import SchipholApiSettings, KafkaSettings
-from requests_ratelimiter import LimiterSession
-from kstreams import create_engine
-from kstreams.backends.kafka import Kafka
-from opentelemetry import trace
-from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import BatchSpanProcessor
-from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
-from opentelemetry.instrumentation.requests import RequestsInstrumentor
-import aiorun
+import asyncio
 import json
 import logging
 
+import aiorun
+from kstreams import create_engine
+from kstreams.backends.kafka import Kafka
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.instrumentation.requests import RequestsInstrumentor
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from requests_ratelimiter import LimiterSession
+
+from settings import KafkaSettings, OtelSettings, SchipholApiSettings
+
 logger = logging.getLogger(__name__)
 
+POLL_INTERVAL_SECONDS = 300  # 5 minutes between Polls
 
-async def start():
-    # Initialize OpenTelemetry tracing (minimal, env-driven)
-    resource = Resource.create({"service.name": "schiphol-producer"})
-    provider = TracerProvider(resource=resource)
-    trace.set_tracer_provider(provider)
-    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
-    RequestsInstrumentor().instrument()
-    tracer = trace.get_tracer("schiphol")
 
-    api_settings = SchipholApiSettings()
-    kafka_settings = KafkaSettings()
-
-    backend = Kafka(bootstrap_servers=kafka_settings.bootstrap_servers)
-    stream_engine = create_engine(title="schiphol-flights-engine", backend=backend)
-    await stream_engine.start()
-
-    session = LimiterSession(per_second=1)
-    session.headers.update(
-        {
-            "Accept": "application/json",
-            "app_id": api_settings.api_id,
-            "app_key": api_settings.api_key,
-            "ResourceVersion": "v4",
-        }
-    )
+async def poll(session, stream_engine, api_settings, kafka_settings, tracer) -> int:
+    """Fetch all Flights from the Schiphol API and publish a Snapshot for each to Kafka.
 
+    Follows pagination via ``link: rel="next"`` response headers until all pages
+    are consumed.  Returns the total number of Flights published.
+    """
     next_url = api_settings.base_url
     total_flights = 0
 
     while next_url:
-        with tracer.start_as_current_span(
-            "fetch_page", attributes={"http.url": next_url}
-        ):
+        with tracer.start_as_current_span("fetch_page", attributes={"http.url": next_url}):
             response = session.get(next_url)
         response.raise_for_status()
         data = response.json()
@@ -80,17 +63,66 @@ async def start():
                     next_url = link.split(";")[0].strip().strip("<>")
                     break
 
-    logger.info(
-        f"Published {total_flights} flights to Kafka topic '{kafka_settings.topic}'"
+    return total_flights
+
+
+async def start():
+    # Initialize OpenTelemetry tracing (minimal, env-driven)
+    otel_settings = OtelSettings()
+
+    resource = Resource.create({"service.name": "schiphol-producer"})
+    provider = TracerProvider(resource=resource)
+    trace.set_tracer_provider(provider)
+    provider.add_span_processor(
+        BatchSpanProcessor(OTLPSpanExporter(endpoint=otel_settings.endpoint))
+    )
+    RequestsInstrumentor().instrument()
+    tracer = trace.get_tracer("schiphol")
+
+    api_settings = SchipholApiSettings()
+    kafka_settings = KafkaSettings()
+
+    backend = Kafka(bootstrap_servers=kafka_settings.bootstrap_servers)
+    stream_engine = create_engine(title="schiphol-flights-engine", backend=backend)
+    await stream_engine.start()
+
+    session = LimiterSession(per_second=1)
+    session.headers.update(
+        {
+            "Accept": "application/json",
+            "app_id": api_settings.api_id,
+            "app_key": api_settings.api_key,
+            "ResourceVersion": "v4",
+        }
     )
 
-    await stream_engine.stop()
+    try:
+        while True:
+            with tracer.start_as_current_span("poll_cycle") as span:
+                try:
+                    total_flights = await poll(
+                        session, stream_engine, api_settings, kafka_settings, tracer
+                    )
+                    span.set_attribute("flights.published", total_flights)
+                    logger.info(
+                        f"Published {total_flights} flights to Kafka topic '{kafka_settings.topic}'"
+                    )
+                except Exception:
+                    span.set_attribute("flights.published", 0)
+                    logger.exception("Poll failed — will retry after %ds", POLL_INTERVAL_SECONDS)
+            await asyncio.sleep(POLL_INTERVAL_SECONDS)
+    finally:
+        await stream_engine.stop()
 
 
 async def shutdown(loop):
     logger.info("Shutdown")
 
 
-if __name__ == "__main__":
+def main():
     logging.basicConfig(level=logging.INFO)
     aiorun.run(start(), stop_on_unhandled_errors=True, shutdown_callback=shutdown)
+
+
+if __name__ == "__main__":
+    main()

src/middelware.py

@@ -1,17 +1,14 @@
 # middleware.py
-from kstreams import middleware, ConsumerRecord
-from schema_registry.client import AsyncSchemaRegistryClient
-from schema_registry.serializers import AsyncJsonMessageSerializer
-from typing import Optional, Dict
 import logging
 
+from kstreams import ConsumerRecord, middleware
+from schema_registry.client import AsyncSchemaRegistryClient
+from schema_registry.serializers import AsyncJsonMessageSerializer
 
 logger = logging.getLogger(__name__)
 
 
-class ConfluentJsonSchemaMiddleware(
-    middleware.BaseMiddleware, AsyncJsonMessageSerializer
-):
+class ConfluentJsonSchemaMiddleware(middleware.BaseMiddleware, AsyncJsonMessageSerializer):
     """
     Middleware to deserialize JSON messages using Confluent Schema Registry
     """
@@ -20,13 +17,13 @@ class ConfluentJsonSchemaMiddleware(
         self,
         *,
         schema_registry_client: AsyncSchemaRegistryClient,
-        reader_schema: Optional[Dict] = None,
+        reader_schema: dict | None = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
         self.schemaregistry_client = schema_registry_client
         self.reader_schema = reader_schema
-        self.id_to_decoder: Dict = {}
+        self.id_to_decoder: dict = {}
 
     async def __call__(self, cr: ConsumerRecord):
         """

src/models.py

@@ -59,12 +59,8 @@ class FlightDirection(Enum):
 
 class RouteType(BaseModel):
     destinations: list[str] | None = None
-    eu: str | None = Field(
-        None, description="S (Schengen), E (Europe) or N (non-Europe)"
-    )
-    visa: bool | None = Field(
-        None, description="Indicates if a visum is required for destination"
-    )
+    eu: str | None = Field(None, description="S (Schengen), E (Europe) or N (non-Europe)")
+    visa: bool | None = Field(None, description="Indicates if a visum is required for destination")
 
 
 class AircraftTypeType(BaseModel):
@@ -120,35 +116,19 @@ class CheckinAllocationsType(BaseModel):
 
 class Flight(BaseModel):
     lastUpdatedAt: datetime | None = None
-    actualLandingTime: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
-    actualOffBlockTime: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
+    actualLandingTime: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    actualOffBlockTime: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
     aircraftRegistration: str | None = None
     aircraftType: AircraftTypeType | None = None
     baggageClaim: BaggageClaimType | None = None
     checkinAllocations: CheckinAllocationsType | None = None
     codeshares: CodesharesType | None = None
-    estimatedLandingTime: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
-    expectedTimeBoarding: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
-    expectedTimeGateClosing: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
-    expectedTimeGateOpen: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
-    expectedTimeOnBelt: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
-    expectedSecurityFilter: str | None = Field(
-        None, description="expected security filter"
-    )
+    estimatedLandingTime: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    expectedTimeBoarding: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    expectedTimeGateClosing: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    expectedTimeGateOpen: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    expectedTimeOnBelt: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    expectedSecurityFilter: str | None = Field(None, description="expected security filter")
     flightDirection: FlightDirection | None = None
     flightName: str | None = None
     flightNumber: int | None = None
@@ -165,14 +145,12 @@ class Flight(BaseModel):
     )
     publicFlightState: PublicFlightStateType | None = None
     route: RouteType | None = None
-    scheduleDateTime: datetime | None = Field(
-        None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ"
-    )
+    scheduleDateTime: datetime | None = Field(None, description="yyyy-MM-dd'T'HH:mm:ss.SSSZ")
     scheduleDate: date | None = Field(None, description="yyyy-MM-dd")
     scheduleTime: str | None = Field(None, description="hh:mm:ss")
     serviceType: str | None = Field(
         None,
-        description="The service type category of the commercial flight. For example: J = Passenger Line, C=Passenger Charter, F = Freight Line and H = Freight Charter etc.",
+        description="The service type category of the commercial flight.",
     )
     terminal: int | None = None
     transferPositions: TransferPositionsType | None = None

tests/test_poll.py

@@ -0,0 +1,121 @@
+"""Tests for poll() — the Schiphol API fetch-and-publish sweep."""
+
+import pytest
+import requests
+
+from main import poll
+
+# ---------------------------------------------------------------------------
+# Test doubles
+# ---------------------------------------------------------------------------
+
+
+class _NoOpSpan:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        pass
+
+
+class StubTracer:
+    def start_as_current_span(self, name, attributes=None):
+        return _NoOpSpan()
+
+
+class StubResponse:
+    """Minimal requests.Response lookalike."""
+
+    def __init__(self, flights, *, next_url=None, raise_on_status=False):
+        self._flights = flights
+        self._raise = raise_on_status
+        self.headers = {}
+        if next_url:
+            self.headers["link"] = f'<{next_url}>; rel="next"'
+
+    def raise_for_status(self):
+        if self._raise:
+            raise requests.HTTPError("500 Server Error")
+
+    def json(self):
+        return {"flights": self._flights}
+
+
+class StubSession:
+    """Returns pre-configured StubResponse objects keyed by URL."""
+
+    def __init__(self, responses_by_url: dict):
+        self._responses = responses_by_url
+
+    def get(self, url):
+        return self._responses[url]
+
+
+class MockEngine:
+    """Captures every send() call so tests can assert on Kafka output."""
+
+    def __init__(self):
+        self.sent: list[dict] = []
+
+    async def send(self, topic, *, value, key):
+        self.sent.append({"topic": topic, "value": value, "key": key})
+
+
+class FakeApiSettings:
+    base_url = "https://api.example.com/flights"
+
+
+class FakeKafkaSettings:
+    topic = "flights-test"
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+async def test_single_page_publishes_all_flights_and_returns_count():
+    """poll() publishes every Flight on a single-page response and returns the count."""
+    flights = [{"id": "1", "flightName": "KL100"}, {"id": "2", "flightName": "KL200"}]
+    session = StubSession({FakeApiSettings.base_url: StubResponse(flights)})
+    engine = MockEngine()
+
+    count = await poll(session, engine, FakeApiSettings(), FakeKafkaSettings(), StubTracer())
+
+    assert count == 2
+    assert len(engine.sent) == 2
+    published_keys = {msg["key"] for msg in engine.sent}
+    assert published_keys == {"1", "2"}
+
+
+async def test_pagination_publishes_flights_across_all_pages():
+    """poll() follows link: rel="next" headers and publishes Flights from every page."""
+    page1_url = FakeApiSettings.base_url
+    page2_url = "https://api.example.com/flights?page=2"
+
+    page1_flights = [{"id": "A"}, {"id": "B"}]
+    page2_flights = [{"id": "C"}, {"id": "D"}, {"id": "E"}]
+
+    session = StubSession(
+        {
+            page1_url: StubResponse(page1_flights, next_url=page2_url),
+            page2_url: StubResponse(page2_flights),
+        }
+    )
+    engine = MockEngine()
+
+    count = await poll(session, engine, FakeApiSettings(), FakeKafkaSettings(), StubTracer())
+
+    assert count == 5
+    assert len(engine.sent) == 5
+
+
+async def test_http_error_propagates_out_of_poll():
+    """poll() lets HTTP errors from raise_for_status() propagate to the caller."""
+    session = StubSession(
+        {FakeApiSettings.base_url: StubResponse([], raise_on_status=True)}
+    )
+    engine = MockEngine()
+
+    with pytest.raises(requests.HTTPError):
+        await poll(session, engine, FakeApiSettings(), FakeKafkaSettings(), StubTracer())