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>
This commit is contained in:
Peter 2026-07-19 09:11:07 +02:00
parent 47b8fd29f8
commit cf6d66adb2

View File

@ -1,3 +1,4 @@
import asyncio
import json
import logging
@ -16,6 +17,8 @@ from settings import KafkaSettings, OtelSettings, SchipholApiSettings
logger = logging.getLogger(__name__)
POLL_INTERVAL_SECONDS = 300 # 5 minutes between Polls
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.
@ -93,10 +96,23 @@ async def start():
}
)
total_flights = await poll(session, stream_engine, api_settings, kafka_settings, tracer)
logger.info(f"Published {total_flights} flights to Kafka topic '{kafka_settings.topic}'")
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):