"""OpenF1 API Client - Handles all API calls to openF1.org"""
import asyncio
import aiohttp
from typing import Any, Dict, List, Optional
class OpenF1Client:
"""Client for interacting with openF1.org API"""
BASE_URL = "https://api.openf1.org/v1"
def __init__(self):
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make a GET request to the API"""
if not self.session:
self.session = aiohttp.ClientSession()
url = f"{self.BASE_URL}{endpoint}"
async with self.session.get(url, params=params) as response:
response.raise_for_status()
return await response.json()
async def get_drivers(self, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch drivers data"""
params = {}
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/drivers", params)
async def get_teams(self) -> List[Dict[str, Any]]:
"""Fetch teams data"""
return await self._get("/teams")
async def get_races(self, year: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch races data"""
params = {}
if year is not None:
params["year"] = year
return await self._get("/races", params)
async def get_meetings(self, year: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch meetings data (F1 race meetings)"""
params = {}
if year is not None:
params["year"] = year
return await self._get("/meetings", params)
async def get_results(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch results data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/results", params)
async def get_sessions(self, session_key: Optional[int] = None, meeting_key: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch sessions data (practice, qualifying, race)"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if meeting_key is not None:
params["meeting_key"] = meeting_key
return await self._get("/sessions", params)
async def get_laps(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch lap data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/laps", params)
async def get_stints(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch stint data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/stints", params)
async def get_pit_stops(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch pit stop data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/pit_stops", params)
async def get_weather(self, session_key: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch weather data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
return await self._get("/weather", params)
async def get_incidents(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch incident data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/incidents", params)
async def get_car_data(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch car telemetry data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/car_data", params)
async def get_positions(self, session_key: Optional[int] = None, driver_number: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch position data"""
params = {}
if session_key is not None:
params["session_key"] = session_key
if driver_number is not None:
params["driver_number"] = driver_number
return await self._get("/positions", params)
async def get_driver_standings(self, session_key: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch driver championship standings"""
params = {}
if session_key is not None:
params["session_key"] = session_key
return await self._get("/drivers_standings", params)
async def get_constructor_standings(self, session_key: Optional[int] = None) -> List[Dict[str, Any]]:
"""Fetch constructor/team championship standings"""
params = {}
if session_key is not None:
params["session_key"] = session_key
return await self._get("/constructor_standings", params)
