Skip to content

Defining Async Protocols for Dependency Injection

Async services depend on interfaces — a repository, a cache, an HTTP fetcher, a clock — and tests swap in fakes. typing.Protocol describes those interfaces structurally, so production classes and fakes need no shared base class. The async part adds a trap: a fake whose methods are plain def has the right names and the right signatures, and fails only when something awaits it. Checked with mypy 2.4.0 (--strict) and pyright 1.1.414: both rejected a synchronous fake passed where an async-method protocol was expected, and mypy listed each conflicting method with its expected Coroutine[...] return. At runtime, isinstance(SyncFake(), UserRepo) on a @runtime_checkable protocol returned True — and the first await repo.get(1) raised TypeError: 'dict' object can't be awaited. A fake missing one method returned False from isinstance but ran fine until that method was needed. This guide defines async protocols that catch these mistakes before tests run.

Prerequisites

1. Declare the interface with async methods

A protocol lists the methods a dependency must have, declared exactly as implementations will declare them — with async def:

from typing import Protocol


class UserRepo(Protocol):
    async def get(self, user_id: int) -> dict[str, object]: ...
    async def save(self, user: dict[str, object]) -> None: ...


class PgUserRepo:                                    # no inheritance needed
    async def get(self, user_id: int) -> dict[str, object]:
        return {"id": user_id}

    async def save(self, user: dict[str, object]) -> None:
        ...


async def handler(repo: UserRepo) -> object:
    user = await repo.get(1)
    return user["id"]

await handler(PgUserRepo())                          # accepted by both checkers

The protocol's async def get(...) -> dict[str, object] means "a method that returns Coroutine[Any, Any, dict[str, object]]". Any class whose methods match structurally satisfies it — the production repository, an in-memory fake, a caching wrapper — and handler depends only on the shape. That keeps service code free of imports from the database layer, which is what makes it testable without a database.

Verify: handler(PgUserRepo()) type-checks with no explicit subclassing of UserRepo.

One protocol, several implementations A flow of 5 stages. One protocol, several implementations Protocol UserRepo async get(), async save() PgUserRepo production, asyncpg InMemoryUserRepo tests, a dict CachedUserRepo wraps another repo handler(repo: UserRepo) awaits repo.get() The handler imports the protocol, never the implementations.

2. Let the checker reject synchronous fakes

The fake that causes trouble is the one written in a hurry, with plain methods:

class SyncFake:
    def get(self, user_id: int) -> dict[str, object]:
        return {"id": user_id}

    def save(self, user: dict[str, object]) -> None:
        pass


await handler(SyncFake())
# mypy:  Argument 1 to "handler" has incompatible type "SyncFake"; expected "UserRepo"
#        Following member(s) of "SyncFake" have conflicts:
#            Expected: def get(self, user_id: int) -> Coroutine[Any, Any, dict[str, object]]
#            Got:      def get(self, user_id: int) -> dict[str, object]
# pyright: Argument of type "SyncFake" cannot be assigned to parameter "repo" of type "UserRepo"

Measured at runtime without the checker: TypeError: 'dict' object can't be awaited, raised from inside handler, not where the fake was constructed. A fake missing a method is caught the same way — mypy reported "PartialFake" is missing following "UserRepo" protocol member: save — and at runtime it worked until the code path that called save(). Type-checking test code, not just production code, is what turns both into immediate errors.

Verify: run the type checker over the tests/ directory; a fake with a sync method or a missing method fails the run.

3. Do not rely on runtime_checkable for async correctness

@runtime_checkable makes isinstance() work with a protocol, which tempts code into validating dependencies at runtime. It checks only that the attributes exist:

from typing import runtime_checkable


@runtime_checkable
class UserRepo(Protocol):
    async def get(self, user_id: int) -> dict[str, object]: ...
    async def save(self, user: dict[str, object]) -> None: ...

isinstance(PgUserRepo(), UserRepo)     # True
isinstance(SyncFake(), UserRepo)       # True  - sync methods pass
isinstance(PartialFake(), UserRepo)    # False - only the missing name is detected

Measured: True for the synchronous fake that later crashed, False for the one missing save. Pyright warned at the isinstance call: Class overlaps "UserRepo" unsafely and could produce a match at runtime. If a runtime guard is genuinely needed — at a plugin boundary, for example — check the methods explicitly with inspect.iscoroutinefunction(obj.get); otherwise leave validation to the type checker, which inspects signatures, return types and async-ness.

Verify: no production code uses isinstance(x, SomeAsyncProtocol) as proof that x can be awaited.

What each check caught, for three implementations A grid of 3 rows by 4 columns. What each check caught, for three implementations implementation mypy / pyright isinstance() at runtime PgUserRepo (async methods) accepted True works SyncFake (plain def) rejected True TypeError when awaited PartialFake (no save) rejected False works until save() Only the static check caught both broken fakes.

4. Use callback protocols for async functions with keywords

When the dependency is a single function rather than an object — a fetcher, a clock, a notifier — Callable[[str], Awaitable[bytes]] cannot express keyword arguments. A protocol with __call__ can:

class Fetcher(Protocol):
    async def __call__(self, url: str, *, timeout: float = ...) -> bytes: ...


async def fetch(url: str, *, timeout: float = 5.0) -> bytes: ...
async def fetch_no_timeout(url: str) -> bytes: ...
def fetch_sync(url: str, *, timeout: float = 5.0) -> bytes: ...


async def crawl(fetcher: Fetcher) -> None:
    await fetcher("https://example.com", timeout=2.0)

await crawl(fetch)               # accepted
await crawl(fetch_no_timeout)    # rejected: no timeout keyword
await crawl(fetch_sync)          # rejected: not async

Both checkers accepted fetch and rejected the other two. The = ... default in the protocol means "has a default"; the implementation's actual default does not have to match. This is also the answer for decorators that add keyword parameters, which ParamSpec and Concatenate cannot describe.

Verify: replacing an injected function with one that lacks a keyword the caller passes is a type error.

5. Keep mocks honest with spec and autospec

AsyncMock objects type-check as anything, so the protocol does not protect tests that use bare mocks. spec and create_autospec bring part of the protection back at runtime:

from unittest.mock import AsyncMock, create_autospec

m = AsyncMock(spec=UserRepo)
m.get.return_value = {"id": 7}
await handler(m)                        # 7 - m.get is an AsyncMock, awaitable
m.delete(1)                             # AttributeError: Mock object has no attribute 'delete'

auto = create_autospec(UserRepo, instance=True)
auto.get.return_value = {"id": 9}
await handler(auto)                     # 9
await auto.get(1, 2)                    # TypeError: too many positional arguments

Measured: a bare AsyncMock() accepted any attribute and any arguments; spec=UserRepo rejected unknown attributes; create_autospec also checked call signatures. All three passed mypy without complaint, because mocks are typed as Any. For the strongest guarantee, prefer a small hand-written fake class that the checker verifies against the protocol — it costs a few lines and catches signature drift the moment the protocol changes.

Verify: changing a method signature on the protocol makes at least one test fake fail type checking.

How should this async dependency be described? A decision on What is being injected with 4 outcomes. How should this async dependency be described? What is being injected? an object with async methods Protocol with async def structural, no base class one async function with keywords Protocol with async __call__ keyword-aware one async function, positional only Callable[..., Awaitable[T]] simplest a test double checked fake class or autospec not a bare AsyncMock Let the type checker verify fakes; isinstance and bare mocks verify almost nothing.

Verification

Async protocols protect dependency injection when:

  • Interfaces are Protocols with async def methods, satisfied structurally.
  • Test code is type-checked, so synchronous or incomplete fakes fail before tests run.
  • No code treats isinstance on a runtime-checkable protocol as proof of async-ness.
  • Mocks use spec or create_autospec, or are replaced by checked fake classes.

Diagnostic Hook: grep the test suite for AsyncMock() with no spec argument. Each is a dependency whose interface the tests no longer check — if the real method is renamed, the mock keeps passing and production fails.

Pitfalls & edge cases

  • Sync fakes. Measured: they pass isinstance and crash with TypeError when awaited.
  • Untyped tests. The protocol only helps if the checker runs over test code.
  • Callable for keyword APIs. It cannot express keywords; use a callback protocol.
  • Bare AsyncMock. Typed as Any, it accepts every misuse.

Frequently Asked Questions

How do I define an interface with async methods in Python?

Use typing.Protocol and declare the methods with async def, as in async def get(self, user_id: int) -> User. Any class with matching async methods satisfies it without inheriting from it.

Does isinstance work with protocols that have async methods?

With @runtime_checkable it works, but only checks that the attribute names exist: a fake with synchronous methods returned True and then raised TypeError when awaited. Rely on the static type checker for async correctness.

How do I type an async callback that takes keyword arguments?

Define a Protocol with async def call(self, url: str, *, timeout: float = ...) -> bytes. Both mypy and pyright rejected a function missing the keyword and a synchronous function.

Do AsyncMock objects satisfy protocols?

They type-check as Any, so they satisfy everything statically. Use AsyncMock(spec=Protocol) or create_autospec to restrict attributes and signatures at runtime, or write a small fake class the checker can verify.