ezri/dbus-fast
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore: drop Python 3.8 support as it has reached EOL (#338)

Browse Source 
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
J. Nick Koston 2025-01-07 09:22:28 -10:00 committed by GitHub
parent 471e680354
commit 42a786b23f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 471 additions and 456 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.github/workflows/ci.yml vendored
View File

@ -36,7 +36,6 @@ jobs:
fail-fast: false
matrix:
python-version:
- "3.8"
- "3.9"
- "3.10"
- "3.11"
@ -145,7 +144,7 @@ jobs:
fetch-depth: 0
- name: Install cibuildwheel
run: python -m pip install cibuildwheel==2.20.0
run: python -m pip install cibuildwheel==2.22.0
- name: Build wheels
run: python -m cibuildwheel --output-dir wheelhouse

2
.pre-commit-config.yaml
View File

@ -36,7 +36,7 @@ repos:
rev: v3.19.1
hooks:
- id: pyupgrade
args: [--py37-plus]
args: [--py39-plus]
- repo: https://github.com/PyCQA/isort
rev: 5.13.2
hooks:

746
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

2
pyproject.toml
View File

@ -34,7 +34,7 @@ script = "build_ext.py"
"Changelog" = "https://github.com/bluetooth-devices/dbus-fast/blob/main/CHANGELOG.md"
[tool.poetry.dependencies]
python = "^3.8"
python = "^3.9"
# duplicated in docs/requirements.txt for readthedocs compatibility
[tool.poetry.group.docs.dependencies]

6
src/dbus_fast/_private/address.py
View File

@ -11,9 +11,9 @@ invalid_address_chars_re = re.compile(r"[^-0-9A-Za-z_/.%]")
str_ = str
def parse_address(address_str: str_) -> List[Tuple[str, Dict[str, str]]]:
def parse_address(address_str: str_) -> list[tuple[str, dict[str, str]]]:
"""Parse a dbus address string into a list of addresses."""
addresses: List[Tuple[str, Dict[str, str]]] = []
addresses: list[tuple[str, dict[str, str]]] = []
for address in address_str.split(";"):
if not address:
@ -22,7 +22,7 @@ def parse_address(address_str: str_) -> List[Tuple[str, Dict[str, str]]]:
raise InvalidAddressError("address did not contain a transport")
transport, opt_string = address.split(":", 1)
options: Dict[str, str] = {}
options: dict[str, str] = {}
for kv in opt_string.split(","):
if not kv:

16
src/dbus_fast/_private/marshaller.py
View File

@ -20,7 +20,7 @@ class Marshaller:
__slots__ = ("signature_tree", "_buf", "body")
def __init__(self, signature: str, body: List[Any]) -> None:
def __init__(self, signature: str, body: list[Any]) -> None:
"""Marshaller constructor."""
self.signature_tree = get_signature_tree(signature)
self._buf = bytearray()
@ -89,12 +89,12 @@ class Marshaller:
return written
def write_array(
self, array: Union[List[Any], Dict[Any, Any]], type_: SignatureType
self, array: Union[list[Any], dict[Any, Any]], type_: SignatureType
) -> int:
return self._write_array(array, type_)
def _write_array(
self, array: Union[List[Any], Dict[Any, Any]], type_: SignatureType
self, array: Union[list[Any], dict[Any, Any]], type_: SignatureType
) -> int:
# TODO max array size is 64MiB (67108864 bytes)
written = self._align(4)
@ -137,19 +137,19 @@ class Marshaller:
return written + array_len
def write_struct(
self, array: Union[Tuple[Any], List[Any]], type_: SignatureType
self, array: Union[tuple[Any], list[Any]], type_: SignatureType
) -> int:
return self._write_struct(array, type_)
def _write_struct(
self, array: Union[Tuple[Any], List[Any]], type_: SignatureType
self, array: Union[tuple[Any], list[Any]], type_: SignatureType
) -> int:
written = self._align(8)
for i, value in enumerate(array):
written += self._write_single(type_.children[i], value)
return written
def write_dict_entry(self, dict_entry: List[Any], type_: SignatureType) -> int:
def write_dict_entry(self, dict_entry: list[Any], type_: SignatureType) -> int:
written = self._align(8)
written += self._write_single(type_.children[0], dict_entry[0])
written += self._write_single(type_.children[1], dict_entry[1])
@ -201,9 +201,9 @@ class Marshaller:
self._write_single(type_, body[i])
return self._buf
_writers: Dict[
_writers: dict[
str,
Tuple[
tuple[
Optional[Callable[[Any, Any, SignatureType], int]],
Optional[Callable[[Any], bytes]],
int,

27
src/dbus_fast/_private/unmarshaller.py
View File

@ -3,8 +3,9 @@ import errno
import io
import socket
import sys
from collections.abc import Iterable
from struct import Struct
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from ..constants import MESSAGE_FLAG_MAP, MESSAGE_TYPE_MAP, MessageFlag
from ..errors import InvalidMessageError
@ -157,9 +158,9 @@ def unpack_parser_factory(unpack_from: Callable, size: int) -> READER_TYPE:
def build_simple_parsers(
endian: int,
) -> Dict[str, Callable[["Unmarshaller", SignatureType], Any]]:
) -> dict[str, Callable[["Unmarshaller", SignatureType], Any]]:
"""Build a dict of parsers for simple types."""
parsers: Dict[str, READER_TYPE] = {}
parsers: dict[str, READER_TYPE] = {}
for dbus_type, ctype_size in DBUS_TO_CTYPE.items():
ctype, size = ctype_size
size = ctype_size[1]
@ -231,12 +232,12 @@ class Unmarshaller:
sock: Optional[socket.socket] = None,
negotiate_unix_fd: bool = True,
) -> None:
self._unix_fds: List[int] = []
self._unix_fds: list[int] = []
self._buf = bytearray() # Actual buffer
self._stream = stream
self._sock = sock
self._message: Optional[Message] = None
self._readers: Dict[str, READER_TYPE] = {}
self._readers: dict[str, READER_TYPE] = {}
self._pos = 0
self._body_len = 0
self._serial = 0
@ -494,14 +495,14 @@ class Unmarshaller:
False,
)
def read_struct(self, type_: _SignatureType) -> List[Any]:
def read_struct(self, type_: _SignatureType) -> list[Any]:
self._pos += -self._pos & 7 # align 8
readers = self._readers
return [
readers[child_type.token](self, child_type) for child_type in type_.children
]
def read_dict_entry(self, type_: _SignatureType) -> Tuple[Any, Any]:
def read_dict_entry(self, type_: _SignatureType) -> tuple[Any, Any]:
self._pos += -self._pos & 7 # align 8
return self._readers[type_.children[0].token](
self, type_.children[0]
@ -537,7 +538,7 @@ class Unmarshaller:
return self._buf[self._pos - array_length : self._pos]
if token_as_int == TOKEN_LEFT_CURLY_AS_INT:
result_dict: Dict[Any, Any] = {}
result_dict: dict[Any, Any] = {}
beginning_pos = self._pos
children = child_type.children
child_0 = children[0]
@ -595,7 +596,7 @@ class Unmarshaller:
result_list.append(reader(self, child_type))
return result_list
def _header_fields(self, header_length: _int) -> Dict[str, Any]:
def _header_fields(self, header_length: _int) -> dict[str, Any]:
"""Header fields are always a(yv)."""
beginning_pos = self._pos
headers = {}
@ -696,7 +697,7 @@ class Unmarshaller:
signature = header_fields.pop("signature", "")
if not self._body_len:
tree = SIGNATURE_TREE_EMPTY
body: List[Any] = []
body: list[Any] = []
else:
token_as_int = ord(signature[0])
if len(signature) == 1:
@ -778,7 +779,7 @@ class Unmarshaller:
return None
return self._message
_complex_parsers_unpack: Dict[
_complex_parsers_unpack: dict[
str, Callable[["Unmarshaller", SignatureType], Any]
] = {
"b": read_boolean,
@ -795,11 +796,11 @@ class Unmarshaller:
UINT16_DBUS_TYPE: read_uint16_unpack,
}
_ctype_by_endian: Dict[int, Dict[str, READER_TYPE]] = {
_ctype_by_endian: dict[int, dict[str, READER_TYPE]] = {
endian: build_simple_parsers(endian) for endian in (LITTLE_ENDIAN, BIG_ENDIAN)
}
_readers_by_type: Dict[int, Dict[str, READER_TYPE]] = {
_readers_by_type: dict[int, dict[str, READER_TYPE]] = {
LITTLE_ENDIAN: {
**_ctype_by_endian[LITTLE_ENDIAN],
**_complex_parsers_unpack,

12
src/dbus_fast/_private/util.py
View File

@ -6,7 +6,7 @@ from ..signature import SignatureTree, Variant, get_signature_tree
def signature_contains_type(
signature: Union[str, SignatureTree], body: List[Any], token: str
signature: Union[str, SignatureTree], body: list[Any], token: str
) -> bool:
"""For a given signature and body, check to see if it contains any members
with the given token"""
@ -49,8 +49,8 @@ def signature_contains_type(
def replace_fds_with_idx(
signature: Union[str, SignatureTree], body: List[Any]
) -> Tuple[List[Any], List[int]]:
signature: Union[str, SignatureTree], body: list[Any]
) -> tuple[list[Any], list[int]]:
"""Take the high level body format and convert it into the low level body
format. Type 'h' refers directly to the fd in the body. Replace that with
an index and return the corresponding list of unix fds that can be set on
@ -76,8 +76,8 @@ def replace_fds_with_idx(
def replace_idx_with_fds(
signature: Union[str, SignatureTree], body: List[Any], unix_fds: List[int]
) -> List[Any]:
signature: Union[str, SignatureTree], body: list[Any], unix_fds: list[int]
) -> list[Any]:
"""Take the low level body format and return the high level body format.
Type 'h' refers to an index in the unix_fds array. Replace those with the
actual file descriptor or `None` if one does not exist."""
@ -128,7 +128,7 @@ def parse_annotation(annotation: str) -> str:
return annotation
def _replace_fds(body_obj: List[Any], children, replace_fn):
def _replace_fds(body_obj: list[Any], children, replace_fn):
"""Replace any type 'h' with the value returned by replace_fn() given the
value of the fd field. This is used by the high level interfaces which
allow type 'h' to be the fd directly instead of an index in an external

6
src/dbus_fast/aio/message_bus.py
View File

@ -57,7 +57,7 @@ class _MessageWriter:
def __init__(self, bus: "MessageBus") -> None:
"""A class to handle writing messages to the message bus."""
self.messages: deque[
Tuple[bytearray, Optional[List[int]], Optional[asyncio.Future]]
tuple[bytearray, Optional[list[int]], Optional[asyncio.Future]]
] = deque()
self.negotiate_unix_fd = bus._negotiate_unix_fd
self.bus = bus
@ -66,7 +66,7 @@ class _MessageWriter:
self.buf: Optional[memoryview] = None
self.fd = bus._fd
self.offset = 0
self.unix_fds: Optional[List[int]] = None
self.unix_fds: Optional[list[int]] = None
self.fut: Optional[asyncio.Future] = None
def write_callback(self, remove_writer: bool = True) -> None:
@ -208,7 +208,7 @@ class MessageBus(BaseMessageBus):
self._auth = auth
self._disconnect_future = self._loop.create_future()
self._pending_futures: Set[asyncio.Future] = set()
self._pending_futures: set[asyncio.Future] = set()
async def connect(self) -> "MessageBus":
"""Connect this message bus to the DBus daemon.

2
src/dbus_fast/aio/proxy_object.py
View File

@ -201,5 +201,5 @@ class ProxyObject(BaseProxyObject):
def get_interface(self, name: str) -> ProxyInterface:
return super().get_interface(name)
def get_children(self) -> List["ProxyObject"]:
def get_children(self) -> list["ProxyObject"]:
return super().get_children()

2
src/dbus_fast/auth.py
View File

@ -25,7 +25,7 @@ class _AuthResponse(enum.Enum):
AGREE_UNIX_FD = "AGREE_UNIX_FD"
@classmethod
def parse(klass, line: str) -> Tuple["_AuthResponse", List[str]]:
def parse(klass, line: str) -> tuple["_AuthResponse", list[str]]:
args = line.split(" ")
response = klass(args[0])
return response, args[1:]

2
src/dbus_fast/glib/proxy_object.py
View File

@ -316,5 +316,5 @@ class ProxyObject(BaseProxyObject):
def get_interface(self, name: str) -> ProxyInterface:
return super().get_interface(name)
def get_children(self) -> List["ProxyObject"]:
def get_children(self) -> list["ProxyObject"]:
return super().get_children()

14
src/dbus_fast/introspection.py
View File

@ -31,7 +31,7 @@ class Arg:
def __init__(
self,
signature: Union[SignatureType, str],
direction: Optional[List[ArgDirection]] = None,
direction: Optional[list[ArgDirection]] = None,
name: Optional[str] = None,
):
if name is not None:
@ -105,7 +105,7 @@ class Signal:
- :class:`InvalidMemberNameError <dbus_fast.InvalidMemberNameError>` - If the name of the signal is not a valid member name.
"""
def __init__(self, name: Optional[str], args: Optional[List[Arg]] = None):
def __init__(self, name: Optional[str], args: Optional[list[Arg]] = None):
if name is not None:
assert_member_name_valid(name)
@ -168,7 +168,7 @@ class Method:
- :class:`InvalidMemberNameError <dbus_fast.InvalidMemberNameError>` - If the name of this method is not valid.
"""
def __init__(self, name: str, in_args: List[Arg] = [], out_args: List[Arg] = []):
def __init__(self, name: str, in_args: list[Arg] = [], out_args: list[Arg] = []):
assert_member_name_valid(name)
self.name = name
@ -312,9 +312,9 @@ class Interface:
def __init__(
self,
name: str,
methods: Optional[List[Method]] = None,
signals: Optional[List[Signal]] = None,
properties: Optional[List[Property]] = None,
methods: Optional[list[Method]] = None,
signals: Optional[list[Signal]] = None,
properties: Optional[list[Property]] = None,
):
assert_interface_name_valid(name)
@ -397,7 +397,7 @@ class Node:
def __init__(
self,
name: Optional[str] = None,
interfaces: Optional[List[Interface]] = None,
interfaces: Optional[list[Interface]] = None,
is_root: bool = True,
):
if not is_root and not name:

12
src/dbus_fast/message.py
View File

@ -109,9 +109,9 @@ class Message:
error_name: Optional[Union[str, ErrorType]] = None,
reply_serial: int = 0,
sender: Optional[str] = None,
unix_fds: List[int] = [],
unix_fds: list[int] = [],
signature: Optional[Union[SignatureTree, str]] = None,
body: List[Any] = [],
body: list[Any] = [],
serial: int = 0,
validate: bool = True,
) -> None:
@ -203,8 +203,8 @@ class Message:
def new_method_return(
msg: "Message",
signature: str = "",
body: List[Any] = [],
unix_fds: List[int] = [],
body: list[Any] = [],
unix_fds: list[int] = [],
) -> "Message":
"""A convenience constructor to create a method return to the given method call message.
@ -238,8 +238,8 @@ class Message:
interface: str,
member: str,
signature: str = "",
body: Optional[List[Any]] = None,
unix_fds: Optional[List[int]] = None,
body: Optional[list[Any]] = None,
unix_fds: Optional[list[int]] = None,
) -> "Message":
"""A convenience constructor to create a new signal message.

18
src/dbus_fast/message_bus.py
View File

@ -121,7 +121,7 @@ class BaseMessageBus:
self,
bus_address: Optional[str] = None,
bus_type: BusType = BusType.SESSION,
ProxyObject: Optional[Type[BaseProxyObject]] = None,
ProxyObject: Optional[type[BaseProxyObject]] = None,
negotiate_unix_fd: bool = False,
) -> None:
self.unique_name: Optional[str] = None
@ -132,20 +132,20 @@ class BaseMessageBus:
# the main loop.
self._user_disconnect = False
self._method_return_handlers: Dict[
self._method_return_handlers: dict[
int, Callable[[Optional[Message], Optional[Exception]], None]
] = {}
self._serial = 0
self._user_message_handlers: List[
self._user_message_handlers: list[
Callable[[Message], Union[Message, bool, None]]
] = []
# the key is the name and the value is the unique name of the owner.
# This cache is kept up to date by the NameOwnerChanged signal and is
# used to route messages to the correct proxy object. (used for the
# high level client only)
self._name_owners: Dict[str, str] = {}
self._name_owners: dict[str, str] = {}
# used for the high level service
self._path_exports: Dict[str, list[ServiceInterface]] = {}
self._path_exports: dict[str, list[ServiceInterface]] = {}
self._bus_address = (
parse_address(bus_address)
if bus_address
@ -156,7 +156,7 @@ class BaseMessageBus:
self._name_owner_match_rule = "sender='org.freedesktop.DBus',interface='org.freedesktop.DBus',path='/org/freedesktop/DBus',member='NameOwnerChanged'"
# _match_rules: the keys are match rules and the values are ref counts
# (used for the high level client only)
self._match_rules: Dict[str, int] = {}
self._match_rules: dict[str, int] = {}
self._high_level_client_initialized = False
self._ProxyObject = ProxyObject
@ -350,7 +350,7 @@ class BaseMessageBus:
ServiceInterface._get_all_property_values(interface, get_properties_callback)
def _emit_interface_removed(self, path: str, removed_interfaces: List[str]) -> None:
def _emit_interface_removed(self, path: str, removed_interfaces: list[str]) -> None:
"""Emit the ``org.freedesktop.DBus.ObjectManager.InterfacesRemoved` signal.
This signal is intended to be used to alert clients when
@ -622,8 +622,8 @@ class BaseMessageBus:
interface_name: str,
member: str,
signature: str,
body: List[Any],
unix_fds: List[int] = [],
body: list[Any],
unix_fds: list[int] = [],
) -> None:
path = None
for p, ifaces in self._path_exports.items():

9
src/dbus_fast/proxy_object.py
View File

@ -3,9 +3,10 @@ import inspect
import logging
import re
import xml.etree.ElementTree as ET
from collections.abc import Coroutine
from dataclasses import dataclass
from functools import lru_cache
from typing import Callable, Coroutine, Dict, List, Optional, Type, Union
from typing import Callable, Dict, List, Optional, Type, Union
from . import introspection as intr
from . import message_bus
@ -62,7 +63,7 @@ class BaseProxyInterface:
self.path = path
self.introspection = introspection
self.bus = bus
self._signal_handlers: Dict[str, List[SignalHandler]] = {}
self._signal_handlers: dict[str, list[SignalHandler]] = {}
self._signal_match_rule = f"type='signal',sender={bus_name},interface={introspection.name},path={path}"
_underscorer1 = re.compile(r"(.)([A-Z][a-z]+)")
@ -239,7 +240,7 @@ class BaseProxyObject:
path: str,
introspection: Union[intr.Node, str, ET.Element],
bus: "message_bus.BaseMessageBus",
ProxyInterface: Type[BaseProxyInterface],
ProxyInterface: type[BaseProxyInterface],
) -> None:
assert_object_path_valid(path)
assert_bus_name_valid(bus_name)
@ -330,7 +331,7 @@ class BaseProxyObject:
self._interfaces[name] = interface
return interface
def get_children(self) -> List["BaseProxyObject"]:
def get_children(self) -> list["BaseProxyObject"]:
"""Get the child nodes of this proxy object according to the introspection data."""
if self._children is None:
self._children = [

4
src/dbus_fast/send_reply.py
View File

@ -28,7 +28,7 @@ class SendReply:
def _exit(
self,
exc_type: Optional[Type[Exception]],
exc_type: Optional[type[Exception]],
exc_value: Optional[Exception],
tb: Optional[TracebackType],
) -> bool:
@ -49,7 +49,7 @@ class SendReply:
def __exit__(
self,
exc_type: Optional[Type[Exception]],
exc_type: Optional[type[Exception]],
exc_value: Optional[Exception],
tb: Optional[TracebackType],
) -> bool:

32
src/dbus_fast/service.py
View File

@ -317,7 +317,7 @@ def _real_fn_result_to_body(
result: Optional[Any],
signature_tree: SignatureTree,
replace_fds: bool,
) -> Tuple[List[Any], List[int]]:
) -> tuple[list[Any], list[int]]:
out_len = len(signature_tree.types)
if result is None:
final_result = []
@ -362,13 +362,13 @@ class ServiceInterface:
def __init__(self, name: str) -> None:
# TODO cannot be overridden by a dbus member
self.name = name
self.__methods: List[_Method] = []
self.__properties: List[_Property] = []
self.__signals: List[_Signal] = []
self.__methods: list[_Method] = []
self.__properties: list[_Property] = []
self.__signals: list[_Signal] = []
self.__buses = set()
self.__handlers: Dict[
self.__handlers: dict[
BaseMessageBus,
Dict[_Method, Callable[[Message, Callable[[Message], None]], None]],
dict[_Method, Callable[[Message, Callable[[Message], None]], None]],
] = {}
for name, member in inspect.getmembers(type(self)):
@ -404,7 +404,7 @@ class ServiceInterface:
)
def emit_properties_changed(
self, changed_properties: Dict[str, Any], invalidated_properties: List[str] = []
self, changed_properties: dict[str, Any], invalidated_properties: list[str] = []
):
"""Emit the ``org.freedesktop.DBus.Properties.PropertiesChanged`` signal.
@ -464,26 +464,26 @@ class ServiceInterface:
)
@staticmethod
def _get_properties(interface: "ServiceInterface") -> List[_Property]:
def _get_properties(interface: "ServiceInterface") -> list[_Property]:
return interface.__properties
@staticmethod
def _get_methods(interface: "ServiceInterface") -> List[_Method]:
def _get_methods(interface: "ServiceInterface") -> list[_Method]:
return interface.__methods
@staticmethod
def _c_get_methods(interface: "ServiceInterface") -> List[_Method]:
def _c_get_methods(interface: "ServiceInterface") -> list[_Method]:
# _c_get_methods is used by the C code to get the methods for an
# interface
# https://github.com/cython/cython/issues/3327
return interface.__methods
@staticmethod
def _get_signals(interface: "ServiceInterface") -> List[_Signal]:
def _get_signals(interface: "ServiceInterface") -> list[_Signal]:
return interface.__signals
@staticmethod
def _get_buses(interface: "ServiceInterface") -> Set["BaseMessageBus"]:
def _get_buses(interface: "ServiceInterface") -> set["BaseMessageBus"]:
return interface.__buses
@staticmethod
@ -520,11 +520,11 @@ class ServiceInterface:
del interface.__handlers[bus]
@staticmethod
def _msg_body_to_args(msg: Message) -> List[Any]:
def _msg_body_to_args(msg: Message) -> list[Any]:
return ServiceInterface._c_msg_body_to_args(msg)
@staticmethod
def _c_msg_body_to_args(msg: Message) -> List[Any]:
def _c_msg_body_to_args(msg: Message) -> list[Any]:
# https://github.com/cython/cython/issues/3327
if not signature_contains_type(msg.signature_tree, msg.body, "h"):
return msg.body
@ -541,7 +541,7 @@ class ServiceInterface:
result: Optional[Any],
signature_tree: SignatureTree,
replace_fds: bool = True,
) -> Tuple[List[Any], List[int]]:
) -> tuple[list[Any], list[int]]:
return _real_fn_result_to_body(result, signature_tree, replace_fds)
@staticmethod
@ -549,7 +549,7 @@ class ServiceInterface:
result: Optional[Any],
signature_tree: SignatureTree,
replace_fds: bool,
) -> Tuple[List[Any], List[int]]:
) -> tuple[list[Any], list[int]]:
"""The high level interfaces may return single values which may be
wrapped in a list to be a message body. Also they may return fds
directly for type 'h' which need to be put into an external list."""

10
src/dbus_fast/signature.py
View File

@ -26,7 +26,7 @@ class SignatureType:
def __init__(self, token: str) -> None:
"""Init a new SignatureType."""
self.token: str = token
self.children: List[SignatureType] = []
self.children: list[SignatureType] = []
self._signature: Optional[str] = None
def __eq__(self, other: Any) -> bool:
@ -60,7 +60,7 @@ class SignatureType:
return self._signature
@staticmethod
def _parse_next(signature: str) -> Tuple["SignatureType", str]:
def _parse_next(signature: str) -> tuple["SignatureType", str]:
if not signature:
raise InvalidSignatureError("Cannot parse an empty signature")
@ -295,7 +295,7 @@ class SignatureType:
return True
validators: Dict[str, Callable[["SignatureType", Any], None]] = {
validators: dict[str, Callable[["SignatureType", Any], None]] = {
"y": _verify_byte,
"b": _verify_boolean,
"n": _verify_int16,
@ -336,7 +336,7 @@ class SignatureTree:
def __init__(self, signature: str = "") -> None:
self.signature = signature
self.types: List[SignatureType] = []
self.types: list[SignatureType] = []
if len(signature) > 0xFF:
raise InvalidSignatureError("A signature must be less than 256 characters")
@ -350,7 +350,7 @@ class SignatureTree:
return self.signature == other.signature
return super().__eq__(other)
def verify(self, body: List[Any]) -> bool:
def verify(self, body: list[Any]) -> bool:
"""Verifies that the give body matches this signature tree
:param body: the body to verify for this tree

2
tests/test_marshaller.py
View File

@ -30,7 +30,7 @@ with open(os.path.dirname(__file__) + "/data/get_managed_objects.hex") as fp:
get_managed_objects_msg = fp.read()
def json_to_message(message: Dict[str, Any]) -> Message:
def json_to_message(message: dict[str, Any]) -> Message:
copy = dict(message)
if "message_type" in copy:
copy["message_type"] = MessageType(copy["message_type"])