services/api/openf1.py

from __future__ import annotations

import os
from typing import Any

import requests

OPENF1_BASE_URL = os.getenv("OPENF1_BASE_URL", "https://api.openf1.org/v1")


class OpenF1Error(RuntimeError):
    pass


def _get(endpoint: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    url = f"{OPENF1_BASE_URL.rstrip('/')}/{endpoint.lstrip('/')}"
    response = requests.get(url, params=params, timeout=30)
    if response.status_code >= 400:
        raise OpenF1Error(f"OpenF1 request failed: {response.status_code} {response.text}")
    data = response.json()
    if not isinstance(data, list):
        raise OpenF1Error("OpenF1 response is not a list")
    return data


def fetch_sessions_for_date(date_iso: str) -> list[dict[str, Any]]:
    params = {
        "date_start>=": date_iso,
        "date_end<=": date_iso,
    }
    return _get("sessions", params=params)


def fetch_sessions(params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    return _get("sessions", params=params)


def fetch_car_data(
    *,
    session_key: int,
    driver_number: int | None = None,
    speed_min: float | None = None,
) -> list[dict[str, Any]]:
    params: dict[str, Any] = {"session_key": session_key}
    if driver_number is not None:
        params["driver_number"] = driver_number
    if speed_min is not None:
        params["speed>="] = speed_min
    return _get("car_data", params=params)