OPT project structure

.gitea/workflows/lint.yaml

@@ -1,25 +0,0 @@
-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

@@ -23,33 +23,9 @@ dependencies = [
 [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"
+[tool.setuptools]
+packages = ["src"]

src/main.py

@@ -1,36 +1,53 @@
-import asyncio
-import json
-import logging
-
-import aiorun
+from src.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.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
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.instrumentation.requests import RequestsInstrumentor
+import aiorun
+import json
+import logging
 
 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")
 
-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.
+    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",
+        }
+    )
 
-    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()
@@ -63,56 +80,11 @@ async def poll(session, stream_engine, api_settings, kafka_settings, tracer) ->
                     next_url = link.split(";")[0].strip().strip("<>")
                     break
 
-    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",
-        }
+    logger.info(
+        f"Published {total_flights} flights to Kafka topic '{kafka_settings.topic}'"
     )
 
-    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()
+    await stream_engine.stop()
 
 
 async def shutdown(loop):

src/middelware.py

@@ -1,14 +1,17 @@
 # middleware.py
-import logging
-
-from kstreams import ConsumerRecord, middleware
+from kstreams import middleware, ConsumerRecord
 from schema_registry.client import AsyncSchemaRegistryClient
 from schema_registry.serializers import AsyncJsonMessageSerializer
+from typing import Optional, Dict
+import logging
+
 
 logger = logging.getLogger(__name__)
 
 
-class ConfluentJsonSchemaMiddleware(middleware.BaseMiddleware, AsyncJsonMessageSerializer):
+class ConfluentJsonSchemaMiddleware(
+    middleware.BaseMiddleware, AsyncJsonMessageSerializer
+):
     """
     Middleware to deserialize JSON messages using Confluent Schema Registry
     """
@@ -17,13 +20,13 @@ class ConfluentJsonSchemaMiddleware(middleware.BaseMiddleware, AsyncJsonMessageS
         self,
         *,
         schema_registry_client: AsyncSchemaRegistryClient,
-        reader_schema: dict | None = None,
+        reader_schema: Optional[Dict] = 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,8 +59,12 @@ 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):
@@ -116,19 +120,35 @@ 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
@@ -145,12 +165,14 @@ 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.",
+        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.",
     )
     terminal: int | None = None
     transferPositions: TransferPositionsType | None = None

tests/test_poll.py

@@ -1,121 +0,0 @@
-"""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())