Skip to content

Client

pyopenf1.client.AsyncOpenF1Client

Asynchronous client for interacting with the OpenF1 API.

This is the main facade that should be used by consumers of the library. It manages the lifecycle of the underlying HTTP connection pool and exposes domain-specific API namespaces as properties.

Parameters:

Name Type Description Default
base_url str

Override the default OpenF1 API base URL.

 'https://api.openf1.org/v1'
timeout float

HTTP request timeout in seconds.

 30.0
headers dict[str, str] | None

Extra HTTP headers merged into every request.

 None
max_retries int

Maximum retry attempts for transient errors.

 3
cache_ttl float

TTL in seconds for response cache. 0 disables caching.

 0.0
max_per_second float

Rate limit: max requests per second.

 3.0
max_per_minute float

Rate limit: max requests per minute.

 30.0

Example::

async with AsyncOpenF1Client() as client:
    drivers = await client.drivers.get_drivers(session_key=9158)
    for d in drivers:
        print(f"{d.name_acronym} - {d.team_name}")
Source code in pyopenf1/client.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class AsyncOpenF1Client:
    """Asynchronous client for interacting with the OpenF1 API.

    This is the main facade that should be used by consumers of the
    library.  It manages the lifecycle of the underlying HTTP connection
    pool and exposes domain-specific API namespaces as properties.

    Args:
        base_url: Override the default OpenF1 API base URL.
        timeout: HTTP request timeout in seconds.
        headers: Extra HTTP headers merged into every request.
        max_retries: Maximum retry attempts for transient errors.
        cache_ttl: TTL in seconds for response cache.  ``0`` disables caching.
        max_per_second: Rate limit: max requests per second.
        max_per_minute: Rate limit: max requests per minute.

    Example::

        async with AsyncOpenF1Client() as client:
            drivers = await client.drivers.get_drivers(session_key=9158)
            for d in drivers:
                print(f"{d.name_acronym} - {d.team_name}")
    """

    def __init__(
        self,
        *,
        base_url: str = "https://api.openf1.org/v1",
        timeout: float = 30.0,
        headers: dict[str, str] | None = None,
        max_retries: int = 3,
        cache_ttl: float = 0.0,
        max_per_second: float = 3.0,
        max_per_minute: float = 30.0,
    ) -> None:
        cache = TTLCache(default_ttl=cache_ttl) if cache_ttl > 0 else None
        rate_limiter = RateLimiter(
            max_per_second=max_per_second,
            max_per_minute=max_per_minute,
        )

        self._http = BaseHTTPClient(
            base_url=base_url,
            timeout=timeout,
            headers=headers,
            max_retries=max_retries,
            rate_limiter=rate_limiter,
            cache=cache,
        )

        # Domain API namespaces
        self._telemetry = TelemetryAPI(self._http)
        self._sessions = SessionAPI(self._http)
        self._drivers = DriverAPI(self._http)
        self._timing = TimingAPI(self._http)
        self._race = RaceAPI(self._http)
        self._championship = ChampionshipAPI(self._http)
        self._results = ResultsAPI(self._http)
        self._weather = WeatherAPI(self._http)
        self._team_radio = TeamRadioAPI(self._http)

    # ------------------------------------------------------------------
    # Public API namespaces
    # ------------------------------------------------------------------

    @property
    def telemetry(self) -> TelemetryAPI:
        """Access telemetry / car-data / location endpoints."""
        return self._telemetry

    @property
    def sessions(self) -> SessionAPI:
        """Access session and meeting endpoints."""
        return self._sessions

    @property
    def drivers(self) -> DriverAPI:
        """Access driver information endpoints."""
        return self._drivers

    @property
    def timing(self) -> TimingAPI:
        """Access lap, interval, and position endpoints."""
        return self._timing

    @property
    def race(self) -> RaceAPI:
        """Access race control, pit, stint, and overtake endpoints."""
        return self._race

    @property
    def championship(self) -> ChampionshipAPI:
        """Access championship standings endpoints."""
        return self._championship

    @property
    def results(self) -> ResultsAPI:
        """Access session result and starting grid endpoints."""
        return self._results

    @property
    def weather(self) -> WeatherAPI:
        """Access weather endpoints."""
        return self._weather

    @property
    def team_radio(self) -> TeamRadioAPI:
        """Access team radio endpoints."""
        return self._team_radio

    # ------------------------------------------------------------------
    # Context-manager protocol
    # ------------------------------------------------------------------

    async def aclose(self) -> None:
        """Close the underlying HTTP client and release resources."""
        await self._http.aclose()

    async def __aenter__(self) -> AsyncOpenF1Client:
        """Enter the async context manager."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: Any,
    ) -> None:
        """Exit the async context manager, closing the HTTP pool."""
        await self.aclose()

championship property

Access championship standings endpoints.

drivers property

Access driver information endpoints.

race property

Access race control, pit, stint, and overtake endpoints.

results property

Access session result and starting grid endpoints.

sessions property

Access session and meeting endpoints.

team_radio property

Access team radio endpoints.

telemetry property

Access telemetry / car-data / location endpoints.

timing property

Access lap, interval, and position endpoints.

weather property

Access weather endpoints.

__aenter__() async

Enter the async context manager.

Source code in pyopenf1/client.py 
148
149
150
async def __aenter__(self) -> AsyncOpenF1Client:
    """Enter the async context manager."""
    return self

__aexit__(exc_type, exc_val, exc_tb) async

Exit the async context manager, closing the HTTP pool.

Source code in pyopenf1/client.py 
152
153
154
155
156
157
158
159
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None:
    """Exit the async context manager, closing the HTTP pool."""
    await self.aclose()

aclose() async

Close the underlying HTTP client and release resources.

Source code in pyopenf1/client.py 
144
145
146
async def aclose(self) -> None:
    """Close the underlying HTTP client and release resources."""
    await self._http.aclose()

pyopenf1.sync_client.OpenF1Client

Synchronous client for the OpenF1 API.

Wraps :class:AsyncOpenF1Client and runs all coroutines in an event loop. Supports standard with context manager.

Parameters:

Name Type Description Default
base_url str

Override the default OpenF1 API base URL.

 'https://api.openf1.org/v1'
timeout float

HTTP request timeout in seconds.

 30.0
headers dict[str, str] | None

Extra HTTP headers.

 None
max_retries int

Maximum retry attempts for transient errors.

 3
cache_ttl float

TTL in seconds for response cache. 0 disables.

 0.0
max_per_second float

Rate limit: max requests per second.

 3.0
max_per_minute float

Rate limit: max requests per minute.

 30.0

Example::

with OpenF1Client() as client:
    drivers = client.drivers.get_drivers(session_key=9158)
    for d in drivers:
        print(d.full_name)
Source code in pyopenf1/sync_client.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class OpenF1Client:
    """Synchronous client for the OpenF1 API.

    Wraps :class:`AsyncOpenF1Client` and runs all coroutines in an
    event loop.  Supports standard ``with`` context manager.

    Args:
        base_url: Override the default OpenF1 API base URL.
        timeout: HTTP request timeout in seconds.
        headers: Extra HTTP headers.
        max_retries: Maximum retry attempts for transient errors.
        cache_ttl: TTL in seconds for response cache.  ``0`` disables.
        max_per_second: Rate limit: max requests per second.
        max_per_minute: Rate limit: max requests per minute.

    Example::

        with OpenF1Client() as client:
            drivers = client.drivers.get_drivers(session_key=9158)
            for d in drivers:
                print(d.full_name)
    """

    def __init__(
        self,
        *,
        base_url: str = "https://api.openf1.org/v1",
        timeout: float = 30.0,
        headers: dict[str, str] | None = None,
        max_retries: int = 3,
        cache_ttl: float = 0.0,
        max_per_second: float = 3.0,
        max_per_minute: float = 30.0,
    ) -> None:
        self._async_client = AsyncOpenF1Client(
            base_url=base_url,
            timeout=timeout,
            headers=headers,
            max_retries=max_retries,
            cache_ttl=cache_ttl,
            max_per_second=max_per_second,
            max_per_minute=max_per_minute,
        )
        self._loop: asyncio.AbstractEventLoop | None = None

    def _get_loop(self) -> asyncio.AbstractEventLoop:
        """Get or create an event loop for sync execution."""
        if self._loop is None or self._loop.is_closed():
            self._loop = asyncio.new_event_loop()
        return self._loop

    def _run(self, coro: Any) -> Any:
        """Run an async coroutine synchronously."""
        return self._get_loop().run_until_complete(coro)

    @property
    def telemetry(self) -> _SyncNamespace:
        """Access telemetry endpoints synchronously."""
        return _SyncNamespace(self._async_client.telemetry, self._run)

    @property
    def sessions(self) -> _SyncNamespace:
        """Access session endpoints synchronously."""
        return _SyncNamespace(self._async_client.sessions, self._run)

    @property
    def drivers(self) -> _SyncNamespace:
        """Access driver endpoints synchronously."""
        return _SyncNamespace(self._async_client.drivers, self._run)

    @property
    def timing(self) -> _SyncNamespace:
        """Access timing endpoints synchronously."""
        return _SyncNamespace(self._async_client.timing, self._run)

    @property
    def race(self) -> _SyncNamespace:
        """Access race endpoints synchronously."""
        return _SyncNamespace(self._async_client.race, self._run)

    @property
    def championship(self) -> _SyncNamespace:
        """Access championship endpoints synchronously."""
        return _SyncNamespace(self._async_client.championship, self._run)

    @property
    def results(self) -> _SyncNamespace:
        """Access results endpoints synchronously."""
        return _SyncNamespace(self._async_client.results, self._run)

    @property
    def weather(self) -> _SyncNamespace:
        """Access weather endpoints synchronously."""
        return _SyncNamespace(self._async_client.weather, self._run)

    @property
    def team_radio(self) -> _SyncNamespace:
        """Access team radio endpoints synchronously."""
        return _SyncNamespace(self._async_client.team_radio, self._run)

    def close(self) -> None:
        """Close the underlying HTTP client."""
        self._run(self._async_client.aclose())
        if self._loop and not self._loop.is_closed():
            self._loop.close()

    def __enter__(self) -> OpenF1Client:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: Any,
    ) -> None:
        self.close()

championship property

Access championship endpoints synchronously.

drivers property

Access driver endpoints synchronously.

race property

Access race endpoints synchronously.

results property

Access results endpoints synchronously.

sessions property

Access session endpoints synchronously.

team_radio property

Access team radio endpoints synchronously.

telemetry property

Access telemetry endpoints synchronously.

timing property

Access timing endpoints synchronously.

weather property

Access weather endpoints synchronously.

close()

Close the underlying HTTP client.

Source code in pyopenf1/sync_client.py 
116
117
118
119
120
def close(self) -> None:
    """Close the underlying HTTP client."""
    self._run(self._async_client.aclose())
    if self._loop and not self._loop.is_closed():
        self._loop.close()