From cf6d66adb2c7b4ebc04d9e0fde45f8ce004e12ca Mon Sep 17 00:00:00 2001 From: Peter Date: Sun, 19 Jul 2026 09:11:07 +0200 Subject: [PATCH] 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> --- src/main.py | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/src/main.py b/src/main.py index 1fc8f94..2b72054 100644 --- a/src/main.py +++ b/src/main.py @@ -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):