From 88cb7c9734b2db0cc42d3e6e198f8ee8391b0c40 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Thu, 23 Jul 2026 17:20:02 +0000 Subject: [PATCH] net: convert NetBackend contract from Protocol to ABC Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/net/backend/abstract.py | 37 +++++++++++++++++++++++----- kafka/net/backend/asyncio_backend.py | 3 ++- kafka/net/backend/selector.py | 3 ++- test/net/backend/test_abstract.py | 28 +++++++++++++++------ 4 files changed, 55 insertions(+), 16 deletions(-) diff --git a/kafka/net/backend/abstract.py b/kafka/net/backend/abstract.py index 6d39f8b7f..992f2cb2e 100644 --- a/kafka/net/backend/abstract.py +++ b/kafka/net/backend/abstract.py @@ -47,6 +47,7 @@ * **Future factory** -- ``create_future`` (see ``NetBackendFuture``). * **Cross-thread wake** -- ``wakeup``. """ +import abc import importlib from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable @@ -158,29 +159,43 @@ def resume_writing(self) -> None: ... ] -@runtime_checkable -class NetBackend(Protocol): - """Structural contract for a pluggable async event-loop backend. +class NetBackend(abc.ABC): + """Contract for a pluggable async event-loop backend. + + An ``abc.ABC`` (not a ``Protocol``): backends inherit it and are checked at + instantiation -- a missing method raises ``TypeError`` immediately, with no + type checker in the loop. ``NetworkSelector`` / ``AsyncioBackend`` subclass + it directly. The satellite surfaces this backend *produces* -- + :class:`NetBackendFuture`, :class:`NetTransport`, :class:`NetProtocol` -- + stay structural ``Protocol``\\ s, since the concrete objects that satisfy + them (``SelectorFuture``, ``KafkaTCPTransport``, ``KafkaConnection``) do so + by shape and shouldn't be forced to inherit. - ``runtime_checkable`` so conformance can be asserted with ``isinstance``; - note that only checks member *presence*, not signatures. ``NetworkSelector`` - satisfies this structurally (no explicit inheritance needed). + Every method below is abstract; the docstrings pin the cross-backend + semantics (which thread resolves futures, fan-out, callback timing). Derived + helpers that compose these primitives (e.g. :meth:`wait_for`) live here as + concrete methods, shared by every backend. """ # --- lifecycle -------------------------------------------------------- + @abc.abstractmethod def start(self) -> None: """Spawn/attach the IO thread that runs the loop. Idempotent.""" + @abc.abstractmethod def stop(self, timeout_ms: Optional[float] = None) -> None: """Stop the loop and join the IO thread. Idempotent.""" + @abc.abstractmethod def close(self) -> None: """Stop (if running) and release loop resources. Idempotent.""" + @abc.abstractmethod def on_io_thread(self) -> bool: """True if the caller is running on this backend's IO thread.""" # --- scheduling ------------------------------------------------------- + @abc.abstractmethod def call_soon(self, task: Any) -> Any: """Enqueue a coroutine/callable to run on the next loop iteration. @@ -190,26 +205,33 @@ def call_soon(self, task: Any) -> Any: deferred-handle box). """ + @abc.abstractmethod def call_soon_with_future(self, coro: Any, *args: Any) -> NetBackendFuture: """Schedule ``coro`` and return a future that resolves with its result.""" + @abc.abstractmethod def call_at(self, when: float, task: Any) -> Any: """Schedule ``task`` to run at absolute monotonic time ``when``.""" + @abc.abstractmethod def call_later(self, delay: float, task: Any) -> Any: """Schedule ``task`` to run after ``delay`` seconds.""" + @abc.abstractmethod def cancel(self, task: Any) -> None: """Cancel a scheduled task/timer previously returned by call_*.""" # --- timing (core coroutines await this) ------------------------------ + @abc.abstractmethod def sleep(self, delay: float) -> Any: """Awaitable that resolves after ``delay`` seconds.""" # --- connection seam -------------------------------------------------- + @abc.abstractmethod async def getaddrinfo(self, host: str, port: int) -> AddrInfoResult: """Resolve host/port via DNS""" + @abc.abstractmethod async def create_connection( self, protocol: NetProtocol, @@ -235,6 +257,7 @@ async def create_connection( """ # --- cross-thread bridge --------------------------------------------- + @abc.abstractmethod def run(self, coro: Any, *args: Any, timeout_ms: Optional[float] = None) -> Any: """Schedule ``coro`` on the loop, block the calling thread, return/raise. @@ -247,10 +270,12 @@ def run(self, coro: Any, *args: Any, timeout_ms: Optional[float] = None) -> Any: """ # --- future factory --------------------------------------------------- + @abc.abstractmethod def create_future(self) -> NetBackendFuture: """Create a loop-awaitable future (see ``NetBackendFuture``).""" # --- misc ------------------------------------------------------------- + @abc.abstractmethod def wakeup(self) -> None: """Interrupt the loop's select() from another thread.""" diff --git a/kafka/net/backend/asyncio_backend.py b/kafka/net/backend/asyncio_backend.py index f34d21868..d14c931f6 100644 --- a/kafka/net/backend/asyncio_backend.py +++ b/kafka/net/backend/asyncio_backend.py @@ -18,6 +18,7 @@ import kafka.errors as Errors from kafka.future import Future +from kafka.net.backend.abstract import NetBackend from kafka.version import __version__ log = logging.getLogger(__name__) @@ -82,7 +83,7 @@ def cancel(self): self._handle.cancel() -class AsyncioBackend: +class AsyncioBackend(NetBackend): DEFAULT_CONFIG = { 'client_id': 'kafka-python-' + __version__, # Default operation deadline for a cross-thread run() call that does not diff --git a/kafka/net/backend/selector.py b/kafka/net/backend/selector.py index 6d8efff98..ea977cc54 100644 --- a/kafka/net/backend/selector.py +++ b/kafka/net/backend/selector.py @@ -12,6 +12,7 @@ import kafka.errors as Errors from kafka.future import Future +from kafka.net.backend.abstract import NetBackend from kafka.net.backend.transport import KafkaTCPTransport from kafka.net.ssl import KafkaSSLTransport from kafka.version import __version__ @@ -194,7 +195,7 @@ def exception(self): return self._exc -class NetworkSelector: +class NetworkSelector(NetBackend): DEFAULT_CONFIG = { 'client_id': 'kafka-python-' + __version__, 'selector': selectors.DefaultSelector, diff --git a/test/net/backend/test_abstract.py b/test/net/backend/test_abstract.py index 1f76e442e..668a85590 100644 --- a/test/net/backend/test_abstract.py +++ b/test/net/backend/test_abstract.py @@ -1,9 +1,10 @@ """Conformance tests for the NetBackend contract (kafka/net/backend/abstract.py). -NetworkSelector is the reference implementation; these pin that it satisfies -the NetBackend Protocol structurally and that the shared lifecycle helper -``on_io_thread()`` behaves correctly. Step 4's AsyncioBackend will be held to -the same isinstance/method-presence checks. +NetworkSelector is the reference implementation; these pin that it inherits the +NetBackend ABC, that the ABC enforces the contract at instantiation (an +incomplete subclass raises TypeError), and that the shared lifecycle helper +``on_io_thread()`` behaves correctly. AsyncioBackend is held to the same +subclass/method-presence checks. """ import asyncio import threading @@ -31,18 +32,29 @@ class TestNetBackendContract: - def test_networkselector_satisfies_protocol(self): + def test_networkselector_is_netbackend_subclass(self): + assert issubclass(NetworkSelector, NetBackend) assert isinstance(NetworkSelector(), NetBackend) + def test_netbackend_is_an_abc(self): + import abc + assert isinstance(NetBackend, abc.ABCMeta) + # Every contract method is abstract, so the base itself can't instantiate. + with pytest.raises(TypeError): + NetBackend() + def test_plain_object_is_not_netbackend(self): assert not isinstance(object(), NetBackend) - def test_partial_impl_is_not_netbackend(self): - class Partial: + def test_incomplete_subclass_cannot_instantiate(self): + # ABC enforcement at instantiation -- no type checker required: a + # subclass missing any abstract method raises TypeError when built. + class Partial(NetBackend): def start(self): pass # missing everything else - assert not isinstance(Partial(), NetBackend) + with pytest.raises(TypeError): + Partial() def test_all_contract_methods_present_and_callable(self): net = NetworkSelector()