refactor(interface): make the registry-entry helpers public in pcapkit.interface.registry (#1519)

docs/source/pcapkit/interface/index.rst

@@ -11,12 +11,15 @@ designed to help and simplify the usage of :mod:`pcapkit`.
 
    core
    misc
+   registry
 
 For wrapper interface functions and constants, please
 refer to the documentation of :mod:`pcapkit.interface.core`
 module, where the core interfaces are defined. And for the
 auxiliary interfaces, please refer to the documentation
 of :mod:`pcapkit.interface.misc` module.
+The helpers the registrars share live in the
+:mod:`pcapkit.interface.registry` module.
 
 .. note::
 
@@ -69,3 +72,9 @@ Following is a table of all interfaces defined in this module:
 +------------------+---------------------+--------------------------------------------+
 | Miscellaneous    | :func:`pcapkit.interface.misc.follow_tcp_stream`                 |
 +------------------+---------------------+--------------------------------------------+
+|                  | :func:`pcapkit.interface.registry.same_entry`                    |
++                  +---------------------+--------------------------------------------+
+| Registry Helpers | :func:`pcapkit.interface.registry.restore_shipped`               |
++                  +---------------------+--------------------------------------------+
+|                  | :func:`pcapkit.interface.registry.restore_shipped_dumper`        |
++------------------+---------------------+--------------------------------------------+

docs/source/pcapkit/interface/registry.rst

@@ -0,0 +1,29 @@
+Registry Helpers
+================
+
+.. module:: pcapkit.interface.registry
+
+:mod:`pcapkit.interface.registry` contains the helpers the
+registrars use to compare registry entries and to put back
+the entry a registry shipped with, so that an override of a
+built-in can be undone (:issue:`1363`, :issue:`1364`).
+
+They back the registrars of
+:class:`~pcapkit.foundation.extraction.Extractor` --
+:meth:`~pcapkit.foundation.extraction.Extractor.register_engine`,
+:meth:`~pcapkit.foundation.extraction.Extractor.register_reassembly`,
+:meth:`~pcapkit.foundation.extraction.Extractor.register_traceflow` and
+:meth:`~pcapkit.foundation.extraction.Extractor.register_dumper` -- and
+:meth:`TraceFlow.register_dumper <pcapkit.foundation.traceflow.traceflow.TraceFlow.register_dumper>`,
+which the functions in :mod:`pcapkit.foundation.registry` call in turn.
+
+A registry entry is a class or a
+:class:`~pcapkit.corekit.module.ModuleDescriptor` naming one; an
+``__output__`` entry is a ``(dumper, ext)`` pair whose ``dumper``
+is such an entry.
+
+.. autofunction:: pcapkit.interface.registry.same_entry
+
+.. autofunction:: pcapkit.interface.registry.restore_shipped
+
+.. autofunction:: pcapkit.interface.registry.restore_shipped_dumper

pcapkit/foundation/extraction.py

@@ -37,6 +37,7 @@
 from pcapkit.foundation.traceflow import TraceFlowManager
 from pcapkit.foundation.traceflow.data import TraceFlowData
 from pcapkit.foundation.traceflow.traceflow import TraceFlow, TraceFlowBase
+from pcapkit.interface.registry import restore_shipped, restore_shipped_dumper, same_entry
 from pcapkit.utilities.exceptions import (CallableError, FileNotFound, FormatError, IterableError,
                                           RegistryError, UnsupportedCall, stacklevel)
 from pcapkit.utilities.logging import get_logger
@@ -95,93 +96,6 @@
 _P = TypeVar('_P')
 
 
-def _same_entry(entry: 'Any', other: 'Any') -> 'bool':
-    """Tell whether two registry entries name the same class.
-
-    An entry is a class or a :class:`~pcapkit.corekit.module.ModuleDescriptor`.
-    Two descriptors are the same entry when they are equal, and a descriptor
-    and a class are when the descriptor names that class. The class is looked
-    up in :data:`sys.modules` rather than imported: a class someone holds has
-    its module loaded already, so a descriptor whose module is not loaded
-    cannot name it, and an optional engine that is not installed is never
-    imported to answer the question.
-
-    Arguments:
-        entry: registry entry
-        other: registry entry to compare against
-
-    Returns:
-        :data:`True` if both name the same class.
-
-    """
-    if entry is other:
-        return True
-    if isinstance(entry, ModuleDescriptor) and isinstance(other, ModuleDescriptor):
-        return entry == other
-    if isinstance(other, ModuleDescriptor):
-        entry, other = other, entry
-    if not isinstance(entry, ModuleDescriptor) or not isinstance(entry.name, str):
-        return False
-    module = sys.modules.get(entry.module)
-    return module is not None and getattr(module, entry.name, None) is other
-
-
-def _restore_shipped(registry: 'dict[str, Any]', key: 'str', shipped: 'Any', kind: 'str') -> 'None':
-    """Put the entry ``registry`` shipped with back under ``key``.
-
-    The public registrars call this when handed the built-in that shipped under
-    ``key``, so an override can be undone (:issue:`1363`). It stores the shipped
-    object itself, a descriptor if a descriptor shipped, and it is a silent
-    no-op when the incumbent already names the shipped class.
-
-    Arguments:
-        registry: registry to restore
-        key: registry key
-        shipped: entry the registry shipped with under ``key``
-        kind: entry kind, for the warning message
-
-    Warns:
-        RegistryWarning: If a different class is registered under ``key``; it
-            is overwritten.
-
-    """
-    incumbent = registry.get(key)
-    if incumbent is not None and _same_entry(shipped, incumbent):
-        return
-    if incumbent is not None:
-        warn(f'{kind} {key} already registered, overwriting', RegistryWarning)
-    registry[key] = shipped
-
-
-def _restore_shipped_dumper(registry: 'dict[str, Any]', format: 'str',  # pylint: disable=redefined-builtin
-                            shipped: 'tuple[Any, str | None]', ext: 'str') -> 'None':
-    """Put the dumper ``registry`` shipped with back under ``format``.
-
-    The ``__output__`` counterpart of :func:`_restore_shipped`: an entry is a
-    ``(dumper, ext)`` pair, so the shipped pair itself is stored when ``ext``
-    matches it, and the shipped dumper with the new ``ext`` otherwise. A
-    re-registration that only changes ``ext`` stays silent, as the dumper
-    registrars document.
-
-    Arguments:
-        registry: ``__output__`` registry to restore
-        format: format name
-        shipped: ``(dumper, ext)`` pair the registry shipped with under ``format``
-        ext: file extension
-
-    Warns:
-        RegistryWarning: If a different dumper is registered for ``format``;
-            it is overwritten.
-
-    """
-    incumbent = registry.get(format)
-    if incumbent is not None and not _same_entry(shipped[0], incumbent[0]):
-        warn(f'dumper {format} already registered, overwriting', RegistryWarning)
-    elif incumbent is not None and incumbent[1] == ext:
-        return
-    registry[format] = shipped if ext == shipped[1] else (shipped[0], ext)
-
-
 #: Engines whose library opens the input by *path*, ``Extractor._ifnm``, rather
 #: than reading ``Extractor._ifile``, so they cannot take an unnamed stream.
 _PATH_ENGINES = frozenset({'scapy', 'pyshark', 'pypcap', 'pcap_ct'})
@@ -526,8 +440,8 @@ class it names, restores the shipped entry rather than storing
         shipped = cls._shipped_output.get(format)
         if isinstance(dumper, ModuleDescriptor) and (shipped is None or dumper != shipped[0]):
             dumper = dumper.klass
-        if shipped is not None and _same_entry(shipped[0], dumper):
-            _restore_shipped_dumper(cls.__output__, format, shipped, ext)
+        if shipped is not None and same_entry(shipped[0], dumper):
+            restore_shipped_dumper(cls.__output__, format, shipped, ext)
             return
         if not isinstance(dumper, type):
             raise RegistryError(f'dumper must be a class, not {dumper!r}')
@@ -579,8 +493,8 @@ def register_engine(cls, name: 'str', engine: 'ModuleDescriptor[Engine] | Type[E
         shipped = cls._shipped_engine.get(name)
         if isinstance(engine, ModuleDescriptor) and engine != shipped:
             engine = engine.klass
-        if shipped is not None and _same_entry(shipped, engine):
-            _restore_shipped(cls.__engine__, name, shipped, 'engine')
+        if shipped is not None and same_entry(shipped, engine):
+            restore_shipped(cls.__engine__, name, shipped, 'engine')
             return
         if not isinstance(engine, type):
             raise RegistryError(f'engine must be a class, not {engine!r}')
@@ -665,8 +579,8 @@ def register_reassembly(cls, protocol: 'str', reassembly: 'ModuleDescriptor[Reas
         shipped = cls._shipped_reassembly.get(protocol)
         if isinstance(reassembly, ModuleDescriptor) and reassembly != shipped:
             reassembly = reassembly.klass
-        if shipped is not None and _same_entry(shipped, reassembly):
-            _restore_shipped(cls.__reassembly__, protocol, shipped, 'reassembly')
+        if shipped is not None and same_entry(shipped, reassembly):
+            restore_shipped(cls.__reassembly__, protocol, shipped, 'reassembly')
             return
         if not isinstance(reassembly, type):
             raise RegistryError(f'reassembly must be a class, not {reassembly!r}')
@@ -746,8 +660,8 @@ def register_traceflow(cls, protocol: 'str', traceflow: 'ModuleDescriptor[TraceF
         shipped = cls._shipped_traceflow.get(protocol)
         if isinstance(traceflow, ModuleDescriptor) and traceflow != shipped:
             traceflow = traceflow.klass
-        if shipped is not None and _same_entry(shipped, traceflow):
-            _restore_shipped(cls.__traceflow__, protocol, shipped, 'traceflow')
+        if shipped is not None and same_entry(shipped, traceflow):
+            restore_shipped(cls.__traceflow__, protocol, shipped, 'traceflow')
             return
         if not isinstance(traceflow, type):
             raise RegistryError(f'traceflow must be a class, not {traceflow!r}')

pcapkit/foundation/traceflow/traceflow.py

@@ -243,14 +243,14 @@ class it names, restores the shipped entry rather than storing
             ext: file extension
 
         """
-        from pcapkit.foundation.extraction import (  # pylint: disable=import-outside-toplevel
-            _restore_shipped_dumper, _same_entry)
+        from pcapkit.interface.registry import (  # pylint: disable=import-outside-toplevel
+            restore_shipped_dumper, same_entry)
 
         shipped = cls._shipped_output.get(format)
         if isinstance(dumper, ModuleDescriptor) and (shipped is None or dumper != shipped[0]):
             dumper = dumper.klass
-        if shipped is not None and _same_entry(shipped[0], dumper):
-            _restore_shipped_dumper(cls.__output__, format, shipped, ext)
+        if shipped is not None and same_entry(shipped[0], dumper):
+            restore_shipped_dumper(cls.__output__, format, shipped, ext)
             return
         if not isinstance(dumper, type):
             raise RegistryError(f'dumper must be a class, not {dumper!r}')

pcapkit/interface/registry.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""Registry Helpers
+======================
+
+.. module:: pcapkit.interface.registry
+
+:mod:`pcapkit.interface.registry` contains the helpers the
+registrars use to compare registry entries and to put back
+the entry a registry shipped with, so that an override of a
+built-in can be undone (:issue:`1363`, :issue:`1364`).
+
+"""
+import sys
+from typing import TYPE_CHECKING
+
+from pcapkit.corekit.module import ModuleDescriptor
+from pcapkit.utilities.warnings import RegistryWarning, warn
+
+if TYPE_CHECKING:
+    from typing import Any
+
+__all__ = ['same_entry', 'restore_shipped', 'restore_shipped_dumper']
+
+
+def same_entry(entry: 'Any', other: 'Any') -> 'bool':
+    """Tell whether two registry entries name the same class.
+
+    An entry is a class or a :class:`~pcapkit.corekit.module.ModuleDescriptor`.
+    Two descriptors are the same entry when they are equal, and a descriptor
+    and a class are when the descriptor names that class. The class is looked
+    up in :data:`sys.modules` rather than imported: a class someone holds has
+    its module loaded already, so a descriptor whose module is not loaded
+    cannot name it, and an optional engine that is not installed is never
+    imported to answer the question.
+
+    The comparison is symmetric, and two classes are the same entry only when
+    they are the same object.
+
+    Arguments:
+        entry: registry entry
+        other: registry entry to compare against
+
+    Returns:
+        :data:`True` if both name the same class.
+
+    """
+    if entry is other:
+        return True
+    if isinstance(entry, ModuleDescriptor) and isinstance(other, ModuleDescriptor):
+        return entry == other
+    if isinstance(other, ModuleDescriptor):
+        entry, other = other, entry
+    if not isinstance(entry, ModuleDescriptor) or not isinstance(entry.name, str):
+        return False
+    module = sys.modules.get(entry.module)
+    return module is not None and getattr(module, entry.name, None) is other
+
+
+def restore_shipped(registry: 'dict[str, Any]', key: 'str', shipped: 'Any', kind: 'str') -> 'None':
+    """Put the entry ``registry`` shipped with back under ``key``.
+
+    The public registrars call this when handed the built-in that shipped under
+    ``key``, so an override can be undone (:issue:`1363`). It stores the shipped
+    object itself, a descriptor if a descriptor shipped, and it is a silent
+    no-op when the incumbent already names the shipped class, as decided by
+    :func:`same_entry`.
+
+    Arguments:
+        registry: registry to restore
+        key: registry key
+        shipped: entry the registry shipped with under ``key``
+        kind: entry kind, for the warning message
+
+    Warns:
+        RegistryWarning: If a different class is registered under ``key``; it
+            is overwritten.
+
+    """
+    incumbent = registry.get(key)
+    if incumbent is not None and same_entry(shipped, incumbent):
+        return
+    if incumbent is not None:
+        warn(f'{kind} {key} already registered, overwriting', RegistryWarning)
+    registry[key] = shipped
+
+
+def restore_shipped_dumper(registry: 'dict[str, Any]', format: 'str',  # pylint: disable=redefined-builtin
+                           shipped: 'tuple[Any, str | None]', ext: 'str') -> 'None':
+    """Put the dumper ``registry`` shipped with back under ``format``.
+
+    The ``__output__`` counterpart of :func:`restore_shipped`: an entry is a
+    ``(dumper, ext)`` pair, so the shipped pair itself is stored when ``ext``
+    matches it, and the shipped dumper with the new ``ext`` otherwise. A
+    re-registration that only changes ``ext`` stays silent, as the dumper
+    registrars document.
+
+    The incumbent is read with :meth:`dict.get`, so an ``__output__`` that is a
+    :class:`collections.defaultdict` gains no entry from the lookup.
+
+    Arguments:
+        registry: ``__output__`` registry to restore
+        format: format name
+        shipped: ``(dumper, ext)`` pair the registry shipped with under ``format``
+        ext: file extension
+
+    Warns:
+        RegistryWarning: If a different dumper is registered for ``format``;
+            it is overwritten.
+
+    """
+    incumbent = registry.get(format)
+    if incumbent is not None and not same_entry(shipped[0], incumbent[0]):
+        warn(f'dumper {format} already registered, overwriting', RegistryWarning)
+    elif incumbent is not None and incumbent[1] == ext:
+        return
+    registry[format] = shipped if ext == shipped[1] else (shipped[0], ext)