peter/schiphol
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

prefactor: extract poll() function from start()

Browse Source 
The inner fetch-and-publish sweep is moved into a standalone
async poll(session, stream_engine, api_settings, kafka_settings, tracer) -> int
function. start() calls it exactly once — observable behaviour is unchanged.
This creates the test seam that tickets #4 and #5 depend on.

Closes #3

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Peter 2026-07-19 09:03:12 +02:00
parent ff5c8ba548
commit 6dbc79202c
1 changed files with 40 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
src/main.py
View File

@ -12,37 +12,17 @@ from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.sdk.trace.export import BatchSpanProcessor
from requests_ratelimiter import LimiterSession from requests_ratelimiter import LimiterSession
from src.settings import KafkaSettings, SchipholApiSettings from settings import KafkaSettings, OtelSettings, SchipholApiSettings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
async def start(): async def poll(session, stream_engine, api_settings, kafka_settings, tracer) -> int:
# Initialize OpenTelemetry tracing (minimal, env-driven) """Fetch all Flights from the Schiphol API and publish a Snapshot for each to Kafka.
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",
}
)
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 next_url = api_settings.base_url
total_flights = 0 total_flights = 0
@ -80,6 +60,40 @@ async def start():
next_url = link.split(";")[0].strip().strip("<>") next_url = link.split(";")[0].strip().strip("<>")
break 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",
}
)
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}'") logger.info(f"Published {total_flights} flights to Kafka topic '{kafka_settings.topic}'")
await stream_engine.stop() await stream_engine.stop()