net: convert NetBackend contract from Protocol to ABC

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."""
 

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

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,

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()