Your live data source just broke. The API provider changed their auth overnight — no warning, no migration path. Now your entire app is blocked.

When the live feed dies

I'm coaching a developer building an F1 live race dashboard. The app streams driver positions in real time via Server-Sent (SSE) Events, using Datastar.

The original plan was to use OpenF1's MQTT feed. Then, one day, F1 locked down their SignalR stream so this feed broke overnight. No warning, no migration path. If the app had been hardwired to that one data source, development would have stopped cold.

If your code is coupled to one source, you inherit all of its instability.

The fix: abstract your data source behind an interface. This allows you to swap implementations without touching your app code. This is the repository pattern.

Define the contract

Start with an abstract base class. Keep it minimal: just the methods your app actually calls.

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class Driver:
    name: str
    number: int
    team: str


@dataclass
class Position:
    driver_name: str
    driver_number: int
    position: int
    change: int = 0


class RaceDataSource(ABC):
    @abstractmethod
    async def get_positions(self, session_key: str) -> list[Position]:
        ...

    @abstractmethod
    async def get_drivers(self, session_key: str) -> list[Driver]:
        ...

This is your contract. Every data source returns the same types. Your app code never knows or cares where the data comes from. (For a deeper dive into the pattern itself, the Cosmic Python chapter on repositories is excellent.)

One honest caveat on naming: a textbook repository abstracts a collection of domain objects you persist to and query (add, get, list), like the expense repository I cover elsewhere. This live, read-only feed is strictly closer to a gateway, an abstraction over an external system. I use "repository" here in the looser, everyday sense of a swappable data-source boundary; the swap mechanics are identical either way.

ABC vs Protocol: An ABC enforces the contract through inheritance — subclasses must implement the abstract methods or they fail at instantiation. Python's typing.Protocol offers an alternative: structural subtyping. A class satisfies the protocol if it has the right methods, no inheritance required. For this example, ABC works well because we control all implementations. Protocol shines when you want third-party code to conform to your interface without inheriting from your base class.

Build the implementations

After the OpenF1 breakage, we evaluated alternatives. Sportmonks turned out to be the most reliable option: REST-based, 10-15 second latency, and it handles the SignalR complexity on their end.

class SportmonksDataSource(RaceDataSource):
    async def get_positions(self, session_key: str) -> list[Position]:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{BASE_URL}/fixtures/{session_key}/positions",
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
            data = resp.json()
            return [Position(**p) for p in data["positions"]]

    async def get_drivers(self, session_key: str) -> list[Driver]:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{BASE_URL}/fixtures/{session_key}/drivers",
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
            data = resp.json()
            return [Driver(**d) for d in data["drivers"]]

But F1 races happen on weekends. Development happens anytime. You can't iterate on a live dashboard without live-shaped data.

So we built a fake that simulates a race from recorded data:

class FakeDataSource(RaceDataSource):
    def __init__(self, data_file: Path, delay_ms: int = 100):
        self.delay_ms = delay_ms
        drivers = self._load_drivers(data_file)
        self.simulator = RaceSimulator(drivers=drivers)

    async def get_positions(self, session_key: str) -> list[Position]:
        self.simulator.tick()
        return self.simulator.get_current_positions()

    async def get_drivers(self, session_key: str) -> list[Driver]:
        return self.simulator.drivers

Notice that FakeDataSource has an __init__ while the ABC doesn't. That's intentional: the ABC defines the contract (which methods your app can call), not how each implementation is constructed. Each data source needs different setup (file paths, API keys, simulators) and no super().__init__() is needed because the base class has no state.

The RaceSimulator randomly swaps adjacent driver positions every few seconds. The dashboard looks alive even without a real race. You can develop the UI, test the SSE streaming, and demo to stakeholders, all without a live event.

Wire it up with a factory

A factory function selects the right implementation at startup based on an environment variable. We use the python-decouple package to read from .env files, but you could use any config system.

from decouple import config

def get_data_source() -> RaceDataSource:
    source_type = config("DATA_SOURCE", default="fake")

    if source_type == "fake":
        return FakeDataSource(
            data_file=Path("data/replay.json"),
            delay_ms=config("FAKE_DELAY_MS", default=100, cast=int),
        )
    elif source_type == "sportmonks":
        return SportmonksDataSource()
    else:
        raise ValueError(f"Unknown data source: {source_type}")

And for local development, I have this in my .env:

# Development
DATA_SOURCE=fake
FAKE_DELAY_MS=100

# Race day (swap the above)
# DATA_SOURCE=sportmonks
# API_KEY=your_sportmonks_api_key

The FastAPI endpoint stays the same regardless:

@app.get("/live/stream")
async def live_endpoint(
    request: Request,
    session_key: str,
    data_source: RaceDataSource = Depends(get_data_source),
):
    async def stream():
        while True:
            positions = await data_source.get_positions(session_key)
            yield SSE.patch_elements(render_positions(positions))
            await asyncio.sleep(SLEEP_INTERVAL)

    return DatastarResponse(stream())

Same endpoint, same rendering code, same SSE streaming. The only thing that changes is where the positions come from.

Why not just mock?

Mocks verify that your code calls the right methods with the right arguments. They don't verify that your app works with real-shaped data.

A fake implementation exercises the same code paths as the real one. Your SSE rendering, your position sorting, your lap counter — all running against data that looks like production. And unlike mocks, you can actually see it in action. Open the dashboard UI and watch fake drivers swap positions. It feels real, even if it's not. This is invaluable for development and demos.

Adding a third source is now trivial

When OpenF1 eventually comes back, or a new provider appears, you add one class and one elif. No endpoint changes, no rendering changes, no test rewiring. That's the payoff: the pattern absorbs instability so your application code doesn't have to.

The repository pattern isn't an abstract architecture concept. It's what kept a developer I'm coaching shipping features the week their data source died. Get the abstraction right early and external instability becomes a config change, not a rewrite.