Skip to content

refactor(protocols): import ProtocolBase under its own name in the last 54 sites #752

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Sep 24, 2026
Merged

refactor(protocols): import ProtocolBase under its own name in the last 54 sites #752

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions pcapkit/foundation/extraction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@
from pcapkit.foundation.traceflow.tcp import TCP as TCP_TraceFlow
from pcapkit.protocols.misc.pcap.frame import Frame
from pcapkit.protocols.misc.pcapng import PCAPNG
from pcapkit.protocols.protocol import ProtocolBase as Protocol
from pcapkit.protocols.protocol import ProtocolBase

#: Every key registered in :attr:`Extractor.__output__` and in
#: :attr:`TraceFlowBase.__output__
Expand All @@ -82,7 +82,7 @@
Packet = Union[Frame, PCAPNG, ScapyPacket, DPKTPacket, PySharkPacket,
PCAPFilePacket, tuple[float, bytes]]

Protocols = Union[str, Protocol, Type[Protocol]]
Protocols = Union[str, ProtocolBase, Type[ProtocolBase]]
VerboseHandler = Callable[['Extractor', Packet], Any]

__all__ = ['Extractor']
Expand Down
70 changes: 35 additions & 35 deletions pcapkit/foundation/registry/protocols.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
from pcapkit.protocols.link.link import Link
from pcapkit.protocols.misc.pcap.frame import Frame
from pcapkit.protocols.misc.pcapng import PCAPNG
from pcapkit.protocols.protocol import ProtocolBase as Protocol
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema.application.httpv2 import FrameType as Schema_HTTP_FrameType
from pcapkit.protocols.schema.internet.hip import Parameter as Schema_HIP_Parameter
from pcapkit.protocols.schema.internet.hopopt import Option as Schema_HOPOPT_Option
Expand Down Expand Up @@ -144,7 +144,7 @@


# NOTE: pcapkit.protocols.__proto__
def register_protocol(protocol: 'Type[Protocol]') -> 'None':
def register_protocol(protocol: 'Type[ProtocolBase]') -> 'None':
"""Registered protocol class.

The protocol class must be a subclass of
Expand Down Expand Up @@ -213,7 +213,7 @@ class under two codes, a supported and documented thing to do, reaches
same class under the same name is silent.

"""
if not issubclass(protocol, Protocol):
if not issubclass(protocol, ProtocolBase):
raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}')

name = protocol.__name__.upper()
Expand Down Expand Up @@ -249,15 +249,15 @@ class under two codes, a supported and documented thing to do, reaches
#: hand, moved into one table so :func:`register_protocol_code` can consult
#: it. ``LinkType`` naming two classes is deliberate, not ambiguous --
#: :func:`register_linktype` already fans out to both.
_CODE_DESTINATIONS: 'dict[type, tuple[Type[Protocol], ...]]' = {
_CODE_DESTINATIONS: 'dict[type, tuple[Type[ProtocolBase], ...]]' = {
Enum_EtherType: (Link,),
Enum_TransType: (Internet,),
Enum_PayloadProtocolIdentifier: (SCTP,),
Enum_LinkType: (Frame, PCAPNG),
}


def _iter_code_targets(code: 'Any') -> 'Iterator[tuple[Type[Protocol], Any]]':
def _iter_code_targets(code: 'Any') -> 'Iterator[tuple[Type[ProtocolBase], Any]]':
"""Flatten a ``code=`` argument into ``(destination, key)`` pairs.

Args:
Expand Down Expand Up @@ -300,7 +300,7 @@ def _iter_code_targets(code: 'Any') -> 'Iterator[tuple[Type[Protocol], Any]]':
f'explicit destination, e.g. code={{TCP: {code!r}}}')


def register_protocol_code(protocol: 'Type[Protocol]', code: 'Any') -> 'None':
def register_protocol_code(protocol: 'Type[ProtocolBase]', code: 'Any') -> 'None':
r"""Register ``protocol`` into the next-layer dispatch registry (or
registries) named by ``code``.

Expand Down Expand Up @@ -377,12 +377,12 @@ def register_protocol_code(protocol: 'Type[Protocol]', code: 'Any') -> 'None':


@overload
def register_linktype(code: 'LinkType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_linktype(code: 'LinkType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_linktype(code: 'LinkType', module: 'str', class_: 'str') -> 'None': ...


def register_linktype(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_linktype(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -408,7 +408,7 @@ def register_linktype(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol

"""
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

Frame.register(code, module)
PCAPNG.register(code, module)
Expand All @@ -421,13 +421,13 @@ def register_linktype(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol


@overload
def register_pcap(code: 'LinkType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_pcap(code: 'LinkType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_pcap(code: 'LinkType', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.misc.pcap.frame.Frame.__proto__
def register_pcap(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_pcap(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -446,7 +446,7 @@ def register_pcap(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol] |

"""
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

Frame.register(code, module)
logger.debug('registered PCAP linktype protocol: %s', code.name)
Expand All @@ -458,13 +458,13 @@ def register_pcap(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol] |


@overload
def register_pcapng(code: 'LinkType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_pcapng(code: 'LinkType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_pcapng(code: 'LinkType', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.misc.pcapng.PCAPNG.__proto__
def register_pcapng(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_pcapng(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -483,7 +483,7 @@ def register_pcapng(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol]

"""
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

PCAPNG.register(code, module)
logger.debug('registered PCAP-NG linktype protocol: %s', code.name)
Expand All @@ -500,13 +500,13 @@ def register_pcapng(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol]


@overload
def register_ethertype(code: 'EtherType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_ethertype(code: 'EtherType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_ethertype(code: 'EtherType', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.link.link.Link.__proto__
def register_ethertype(code: 'EtherType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_ethertype(code: 'EtherType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -525,7 +525,7 @@ def register_ethertype(code: 'EtherType', module: 'str | ModuleDescriptor[Protoc

"""
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

Link.register(code, module)
logger.debug('registered ethertype protocol: %s', code.name)
Expand All @@ -542,13 +542,13 @@ def register_ethertype(code: 'EtherType', module: 'str | ModuleDescriptor[Protoc


@overload
def register_transtype(code: 'TransType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_transtype(code: 'TransType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_transtype(code: 'TransType', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.internet.internet.Internet.__proto__
def register_transtype(code: 'TransType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_transtype(code: 'TransType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -567,7 +567,7 @@ def register_transtype(code: 'TransType', module: 'str | ModuleDescriptor[Protoc

"""
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

Internet.register(code, module)
logger.debug('registered transtype protocol: %s', code.name)
Expand Down Expand Up @@ -784,16 +784,16 @@ def register_mh_extension(code: 'MH_CGAExtension', meth: 'str | tuple[MH_Extensi


@overload
def register_apptype(code: 'int', module: 'ModuleDescriptor[Protocol] | Type[Protocol]', *, proto: 'TransportProtocol | str') -> 'None': ...
def register_apptype(code: 'int', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]', *, proto: 'TransportProtocol | str') -> 'None': ...
@overload
def register_apptype(code: 'Enum_AppType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]', *, proto: 'TransportProtocol | str' = ...) -> 'None': ...
def register_apptype(code: 'Enum_AppType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]', *, proto: 'TransportProtocol | str' = ...) -> 'None': ...
@overload
def register_apptype(code: 'int', module: 'str', class_: 'str', *, proto: 'TransportProtocol | str') -> 'None': ...
@overload
def register_apptype(code: 'Enum_AppType', module: 'str', class_: 'str', *, proto: 'TransportProtocol | str' = ...) -> 'None': ...


def register_apptype(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_apptype(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL, *, proto: 'TransportProtocol | str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand Down Expand Up @@ -833,13 +833,13 @@ def register_apptype(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor
proto = code.proto
code = code.port
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

_reg = False
if isinstance(proto, str):
proto = TransportProtocol.get(proto.lower())

for test, cls in cast('dict[TransportProtocol, Type[Protocol]]', {
for test, cls in cast('dict[TransportProtocol, Type[ProtocolBase]]', {
TransportProtocol.tcp: TCP,
TransportProtocol.udp: UDP,
}).items():
Expand All @@ -860,13 +860,13 @@ def register_apptype(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor


@overload
def register_tcp(code: 'int | Enum_AppType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_tcp(code: 'int | Enum_AppType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_tcp(code: 'int | Enum_AppType', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.transport.tcp.TCP.__proto__
def register_tcp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_tcp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -887,7 +887,7 @@ def register_tcp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[Pro
if isinstance(code, Enum_AppType):
code = code.port
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

TCP.register(code, module)
logger.debug('registered TCP port: %s', code)
Expand Down Expand Up @@ -949,13 +949,13 @@ def register_tcp_mp_option(code: 'TCP_MPTCPOption', meth: 'str | tuple[TCP_MPOpt


@overload
def register_udp(code: 'int | Enum_AppType', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_udp(code: 'int | Enum_AppType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_udp(code: 'int | Enum_AppType', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.transport.udp.UDP.__proto__
def register_udp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_udp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -976,7 +976,7 @@ def register_udp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[Pro
if isinstance(code, Enum_AppType):
code = code.port
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

UDP.register(code, module)
logger.debug('registered UDP port: %s', code)
Expand All @@ -988,13 +988,13 @@ def register_udp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[Pro


@overload
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'ModuleDescriptor[Protocol] | Type[Protocol]') -> 'None': ...
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'str', class_: 'str') -> 'None': ...


# NOTE: pcapkit.protocols.transport.sctp.SCTP.__proto__
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'str | ModuleDescriptor[Protocol] | Type[Protocol]',
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str' = NULL) -> 'None':
r"""Register a new protocol class.

Expand All @@ -1020,7 +1020,7 @@ def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'str | M

"""
if isinstance(module, str):
module = cast('ModuleDescriptor[Protocol]', ModuleDescriptor(module, class_))
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))

SCTP.register(code, module)
logger.debug('registered SCTP payload protocol identifier: %s', code)
Expand Down
10 changes: 5 additions & 5 deletions pcapkit/foundation/traceflow/traceflow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@
from typing_extensions import Literal, Self

from pcapkit.corekit.infoclass import Info
from pcapkit.protocols.protocol import ProtocolBase as Protocol
from pcapkit.protocols.protocol import ProtocolBase

CallbackFn = Callable[[_IT], None]

Expand All @@ -65,7 +65,7 @@ class TraceFlowMeta(abc.ABCMeta):
#: Protocol name of current object.
__protocol_name__: 'str'
#: Protocol of current object.
__protocol_type__: 'Type[Protocol]'
__protocol_type__: 'Type[ProtocolBase]'

@property
def name(cls) -> 'str':
Expand All @@ -75,7 +75,7 @@ def name(cls) -> 'str':
return cls.__name__

@property
def protocol(cls) -> 'Type[Protocol]':
def protocol(cls) -> 'Type[ProtocolBase]':
"""Protocol of current object."""
if hasattr(cls, '__protocol_type__'):
return cls.__protocol_type__
Expand Down Expand Up @@ -131,7 +131,7 @@ class TraceFlowBase(Generic[_DT, _BT, _IT, _PT], metaclass=TraceFlowMeta):
#: Protocol name of current reassembly object.
__protocol_name__: 'str'
#: Protocol of current reassembly object.
__protocol_type__: 'Type[Protocol]'
__protocol_type__: 'Type[ProtocolBase]'

#: List of callback functions upon reassembled datagram.
__callback_fn__: 'list[CallbackFn]'
Expand Down Expand Up @@ -179,7 +179,7 @@ def name(self) -> 'str':
return type(self).name # type: ignore[return-value]

@property
def protocol(self) -> 'Type[Protocol]':
def protocol(self) -> 'Type[ProtocolBase]':
"""Protocol of current flow tracing object.

Note:
Expand Down
5 changes: 2 additions & 3 deletions pcapkit/protocols/application/application.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,7 @@

from pcapkit.corekit.protochain import ProtoChain
from pcapkit.protocols.misc.null import NoPayload
from pcapkit.protocols.protocol import _PT, _ST
from pcapkit.protocols.protocol import ProtocolBase as Protocol
from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase
from pcapkit.utilities.exceptions import IntError, UnsupportedCall

if TYPE_CHECKING:
Expand All @@ -28,7 +27,7 @@
__all__ = ['Application']


class Application(Protocol[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-method
class Application(ProtocolBase[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-method
"""Abstract base class for transport layer protocol family."""

##########################################################################
Expand Down
4 changes: 2 additions & 2 deletions pcapkit/protocols/data/data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
if TYPE_CHECKING:
from typing import Type

from pcapkit.protocols.protocol import ProtocolBase as Protocol
from pcapkit.protocols.protocol import ProtocolBase


class Data(Info):
Expand All @@ -22,4 +22,4 @@ class Data(Info):
#: Next field name, i.e., the name of the payload field.
__next_name__: 'str'
#: Next field type, i.e., the type of the payload field.
__next_type__: 'Type[Protocol]'
__next_type__: 'Type[ProtocolBase]'
Loading
Loading