Defining Typed Repository Contracts with Protocol

Hello! Last lesson, you separated the incident API into a router, an application service, a domain model, and an in-memory adapter. The service accepted a single retrieval function, which was enough for one operation. As a service needs more storage operations, that boundary can become clearer as a typed repository contract.

In this lesson, you’ll define that contract with Python’s Protocol, adapt the existing code to use it, and check what a type checker can—and cannot—guarantee. Allow about 40 minutes, including the selected video, readings, and code walkthrough.

---

Specify what the service needs

A repository contract describes operations available to a service without specifying how they work. For the incident service, that might mean: retrieve an incident by ID, and retrieve open incidents. Neither requirement says “run a SQL query.” An in-memory adapter can satisfy it today; a PostgreSQL adapter can satisfy it later.

Watch protocol-based abstraction in ArjanCodes’ Stop Hardcoding Everything: Use Dependency Injection. The example is a data pipeline rather than an incident repository. Focus on how the pipeline names the behavior it needs, while the concrete components retain their own construction details.

Python calls this structural subtyping: a class is compatible with a protocol if it supplies the required members with compatible types. It need not inherit from the protocol. Contrast that with ordinary class inheritance, where the declared family relationship matters.

The practical design question is who decides which methods belong in the contract? Start with the service that calls them. A single, application-wide repository interface can accumulate methods unrelated to any one service. A small contract makes the service’s requirements visible during review.

---

Define the incident repository Protocol

Keep the previous lesson’s domain/incident.py unchanged. Add application/ports.py beside the service:

from typing import Protocol

from incident_api.domain.incident import Incident


class IncidentRepository(Protocol):
    async def get(self, incident_id: int) -> Incident | None: ...

    async def list_open(self) -> list[Incident]: ...

Each signature makes a useful promise. get accepts an integer ID and, when awaited, produces either a domain Incident or None. The None gives the service an explicit way to handle a missing incident. list_open produces domain objects rather than database rows or HTTP response models.

Both methods are async because the service will await them. That choice accommodates a future adapter performing non-blocking database I/O; it does not make every implementation non-blocking automatically. Notice, too, what the contract omits: connection strings, SQLAlchemy sessions, database model classes, and an initializer. Those are implementation details, not operations the service calls.

Now update application/incidents.py. The previous IncidentNotFound exception remains; the single retrieval callable becomes a repository:

from incident_api.application.ports import IncidentRepository
from incident_api.domain.incident import Incident


class IncidentNotFound(Exception):
    pass


class IncidentService:
    def __init__(self, repository: IncidentRepository) -> None:
        self._repository = repository

    async def read_incident(self, incident_id: int) -> Incident:
        incident = await self._repository.get(incident_id)
        if incident is None:
            raise IncidentNotFound(f"Incident {incident_id} not found")
        return incident

    async def list_incidents_needing_attention(self) -> list[Incident]:
        open_incidents = await self._repository.list_open()
        return [
            incident
            for incident in open_incidents
            if incident.needs_attention()
        ]

The second use case shows why a repository is useful here. The adapter supplies open incidents; the application service applies the existing domain rule, needs_attention(). The service knows neither where records live nor how “open” is queried. This is a fixed, small use case—not yet a general client-configurable filtering endpoint.

Pay particular attention to the imports in these two files. The application layer imports the domain model and its own contract, but does not import the in-memory adapter or PostgreSQL. That import direction is the separation we want.

---

Make the existing adapter satisfy the contract

Replace infrastructure/memory_incidents.py with an implementation of both required methods:

from incident_api.domain.incident import Incident, Status


class MemoryIncidentRepository:
    def __init__(self, incidents: list[Incident]) -> None:
        self._incidents = {incident.id: incident for incident in incidents}

    async def get(self, incident_id: int) -> Incident | None:
        return self._incidents.get(incident_id)

    async def list_open(self) -> list[Incident]:
        return [
            incident
            for incident in self._incidents.values()
            if incident.status is Status.OPEN
        ]

There is no IncidentRepository base class here. Compatibility is structural: MemoryIncidentRepository has matching get and list_open methods. Its dictionary access is immediate in-memory work; the async declarations match the awaited contract but do not perform asynchronous I/O.

The router from the previous lesson can stay unchanged because its read_incident service call has not changed. Update main.py at the composition point, where choosing a concrete implementation is appropriate:

from fastapi import FastAPI

from incident_api.api.incidents import create_incident_router
from incident_api.application.incidents import IncidentService
from incident_api.application.ports import IncidentRepository
from incident_api.domain.incident import Incident, Severity, Status
from incident_api.infrastructure.memory_incidents import MemoryIncidentRepository

repository: IncidentRepository = MemoryIncidentRepository(
    [
        Incident(id=42, severity=Severity.CRITICAL, status=Status.OPEN),
        Incident(id=43, severity=Severity.LOW, status=Status.CLOSED),
    ]
)
service = IncidentService(repository)

app = FastAPI()
app.include_router(create_incident_router(service))

The assignment annotated IncidentRepository gives your static type checker an explicit place to verify the adapter. Try your project’s configured type-check command. If you remove list_open from MemoryIncidentRepository, or change get to accept a str ID, the implementation no longer satisfies this contract and the checker should report a mismatch.

The existing /incidents/{incident_id} route should still return incident 42, incident 43, or a 404 for an unknown ID. The new list_incidents_needing_attention use case is available to call from application code; we have not added a route for it. Keeping the HTTP surface unchanged makes it easier to see that this refactor concerns the service-to-storage boundary.

A type checker examines declared types; Protocol does not automatically validate an adapter when the application runs. Likewise, a matching method signature does not prove that list_open returns only open incidents. Tests remain necessary for behavior. The contract helps catch wiring and signature mistakes early, while focused tests check what the implementation actually does.

---

Keep the contract independent of the database

When a PostgreSQL adapter arrives, it can implement these same operations using its own session and queries, then convert database results into domain Incident objects. The application service need not change merely because storage changes. main.py or another composition point will choose and supply the concrete adapter.

This does not mean every callable needs a Protocol. The previous single-method callable was a sensible small boundary. Introduce a repository contract when naming several related operations makes the consumer’s requirements clearer. Equally, avoid putting every possible incident query into this one interface: a different service may need a different, smaller view of storage.

---

Takeaways

A Protocol lets the incident service state which typed operations it requires without inheriting from, constructing, or importing a PostgreSQL implementation. The in-memory adapter satisfies that contract through matching methods, and the composition point selects the adapter. Static checking can catch incompatible signatures; tests still establish behavioral correctness.

Next, you’ll connect services and repositories through FastAPI dependencies, so the choice of implementation can be wired at the application boundary without hardcoding it into a route.