Use Python 3.11 syntax

examples/appv2/basic_packets/producer.py

@@ -15,7 +15,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # -----------------------------------------------------------------------------
-import typing
 import logging
 from ndn import appv2
 from ndn import encoding as enc
@@ -32,10 +31,10 @@
 
 
 @app.route('/example/testApp')
-def on_interest(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr],
+def on_interest(name: enc.FormalName, _app_param: enc.BinaryStr | None,
                 reply: appv2.ReplyFunc, context: appv2.PktContext):
     print(f'>> I: {enc.Name.to_str(name)}, {context["int_param"]}')
-    content = "Hello, world!".encode()
+    content = b"Hello, world!"
     reply(app.make_data(name, content=content, signer=keychain.get_signer({}),
                         freshness_period=10000))
     print(f'<< D: {enc.Name.to_str(name)}')

examples/appv2/forwarding_hint/producer.py

@@ -1,4 +1,3 @@
-import typing
 import logging
 from ndn import appv2
 from ndn import encoding as enc
@@ -15,10 +14,10 @@
 
 
 @app.route('/repo/command')
-def on_cmd(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr],
+def on_cmd(name: enc.FormalName, _app_param: enc.BinaryStr | None,
            reply: appv2.ReplyFunc, context: appv2.PktContext):
     print(f'>> I: {enc.Name.to_str(name)}, {context["int_param"]}')
-    content = "Hello, world!".encode()
+    content = b"Hello, world!"
     reply(app.make_data(name, content=content, signer=keychain.get_signer({}),
                         freshness_period=10000))
     print(f'<< D: {enc.Name.to_str(name)}')
@@ -30,7 +29,7 @@ def on_cmd(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr],
 # The following function catches all Interests that are not handled.
 # So we can dispatch by forwarding hints.
 @app.route('/')
-def on_fwd_hint(name: enc.FormalName, app_param: typing.Optional[enc.BinaryStr],
+def on_fwd_hint(name: enc.FormalName, app_param: enc.BinaryStr | None,
                 reply: appv2.ReplyFunc, context: appv2.PktContext):
     fwd_hints = context["int_param"].forwarding_hint
     if fwd_hints:

examples/appv2/svs/sync_example.py

@@ -1,4 +1,3 @@
-import typing
 import logging
 import asyncio as aio
 from ndn import appv2

examples/dpdk_experimental/udp_producer.py

@@ -15,7 +15,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # -----------------------------------------------------------------------------
-import typing
 import logging
 import sys
 from ndn import appv2
@@ -48,10 +47,10 @@
 
 
 @app.route('/example/testApp')
-def on_interest(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr],
+def on_interest(name: enc.FormalName, _app_param: enc.BinaryStr | None,
                 reply: appv2.ReplyFunc, context: appv2.PktContext):
     print(f'>> I: {enc.Name.to_str(name)}, {context["int_param"]}')
-    content = "Hello, world!".encode()
+    content = b"Hello, world!"
     reply(app.make_data(name, content=content, signer=keychain.get_signer({}),
                         freshness_period=10000))
     print(f'<< D: {enc.Name.to_str(name)}')

examples/lvs/consumer.py

@@ -1,8 +1,7 @@
 import os
 import sys
 import logging
-from ndn.utils import timestamp
-from ndn.encoding import Name, Component, InterestParam
+from ndn.encoding import Name
 from ndn.security import TpmFile, KeychainSqlite3
 from ndn.app import NDNApp, InterestNack, InterestTimeout, InterestCanceled, ValidationFailure
 from ndn.app_support.light_versec import compile_lvs, Checker, DEFAULT_USER_FNS, lvs_validator

examples/lvs/producer.py

@@ -70,7 +70,7 @@ def main():
     @app.route('/lvs-test/article/xinyu/hello')
     def on_interest(name, param, _app_param):
         print(f'>> I: {Name.to_str(name)}, {param}')
-        content = "Hello,".encode()
+        content = b"Hello,"
         data_name = name + [Component.from_version(timestamp())]
         sign_cert_name = checker.suggest(data_name, app.keychain)
         print(f'        Suggested signing cert: {Name.to_str(sign_cert_name)}')
@@ -82,7 +82,7 @@ def on_interest(name, param, _app_param):
     @app.route('/lvs-test/article/xinyu/world')
     def on_interest(name, param, _app_param):
         print(f'>> I: {Name.to_str(name)}, {param}')
-        content = "world!".encode()
+        content = b"world!"
         data_name = name + [Component.from_version(timestamp())]
         sign_cert_name = checker.suggest(data_name, app.keychain)
         print(f'        Suggested signing cert: {Name.to_str(sign_cert_name)}')

examples/producer.py

@@ -15,7 +15,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # -----------------------------------------------------------------------------
-from typing import Optional
 from ndn.app import NDNApp
 from ndn.encoding import Name, InterestParam, BinaryStr, FormalName, MetaInfo
 import logging
@@ -31,9 +30,9 @@
 
 
 @app.route('/example/testApp')
-def on_interest(name: FormalName, param: InterestParam, _app_param: Optional[BinaryStr]):
+def on_interest(name: FormalName, param: InterestParam, _app_param: BinaryStr | None):
     print(f'>> I: {Name.to_str(name)}, {param}')
-    content = "Hello, world!".encode()
+    content = b"Hello, world!"
     app.put_data(name, content=content, freshness_period=10000)
     print(f'<< D: {Name.to_str(name)}')
     print(MetaInfo(freshness_period=10000))

examples/rpc_producer.py

@@ -15,7 +15,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # -----------------------------------------------------------------------------
-from typing import Optional
 from ndn.app import NDNApp
 from ndn.encoding import Name, InterestParam, BinaryStr, FormalName, MetaInfo
 import logging
@@ -31,7 +30,7 @@
 
 
 @app.route('/example/rpc')
-def on_interest(name: FormalName, param: InterestParam, app_param: Optional[BinaryStr]):
+def on_interest(name: FormalName, param: InterestParam, app_param: BinaryStr | None):
     app_param = bytes(app_param)
     print(f'>> I: {Name.to_str(name)}, {param}, {app_param}')
     if not app_param:

src/ndn/app.py

@@ -18,7 +18,8 @@
 import struct
 import logging
 import asyncio as aio
-from typing import Optional, Any, Awaitable, Coroutine, Tuple, List
+from typing import Any
+from collections.abc import Awaitable, Coroutine
 from .utils import gen_nonce
 from .encoding import BinaryStr, TypeNumber, LpTypeNumber, parse_interest, \
     parse_tl_num, parse_data, DecodeError, Name, NonStrictName, MetaInfo, \
@@ -47,7 +48,7 @@ class NDNApp:
     _prefix_tree: NameTrie = None
     int_validator: Validator = None
     data_validator: Validator = None
-    _autoreg_routes: List[Tuple[FormalName, Route, Optional[Validator], bool, bool]]
+    _autoreg_routes: list[tuple[FormalName, Route, Validator | None, bool, bool]]
     _prefix_register_semaphore: aio.Semaphore = None
     logger: logging.Logger
 
@@ -96,7 +97,7 @@ async def _receive(self, typ: int, data: BinaryStr):
                 self.logger.warning('Unable to decode the fragment of LpPacket')
                 return
             if self.logger.isEnabledFor(logging.DEBUG):
-                self.logger.debug('NetworkNack received %s, reason=%s' % (Name.to_str(name), nack_reason))
+                self.logger.debug('NetworkNack received %s, reason=%s', Name.to_str(name), nack_reason)
             self._on_nack(name, nack_reason)
         else:
             if typ == TypeNumber.INTEREST:
@@ -106,7 +107,7 @@ async def _receive(self, typ: int, data: BinaryStr):
                     self.logger.warning('Unable to decode received packet')
                     return
                 if self.logger.isEnabledFor(logging.DEBUG):
-                    self.logger.debug('Interest received %s' % Name.to_str(name))
+                    self.logger.debug('Interest received %s', Name.to_str(name))
                 await self._on_interest(name, param, app_param, sig, raw_packet=data)
             elif typ == TypeNumber.DATA:
                 try:
@@ -115,7 +116,7 @@ async def _receive(self, typ: int, data: BinaryStr):
                     self.logger.warning('Unable to decode received packet')
                     return
                 if self.logger.isEnabledFor(logging.DEBUG):
-                    self.logger.debug('Data received %s' % Name.to_str(name))
+                    self.logger.debug('Data received %s', Name.to_str(name))
                 await self._on_data(name, meta_info, content, sig, raw_packet=data)
             else:
                 self.logger.warning('Unable to decode received packet')
@@ -132,7 +133,7 @@ def put_raw_packet(self, data: BinaryStr):
             raise NetworkError('cannot send packet before connected')
         self.face.send(data)
 
-    def prepare_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, **kwargs):
+    def prepare_data(self, name: NonStrictName, content: BinaryStr | None = None, **kwargs):
         r"""
         Prepare a Data packet by generating, encoding and signing it.
 
@@ -155,7 +156,7 @@ def prepare_data(self, name: NonStrictName, content: Optional[BinaryStr] = None,
             meta_info = MetaInfo.from_dict(kwargs)
         return make_data(name, meta_info, content, signer=signer)
 
-    def put_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, **kwargs):
+    def put_data(self, name: NonStrictName, content: BinaryStr | None = None, **kwargs):
         r"""
         Publish a Data packet.
 
@@ -170,10 +171,10 @@ def put_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, **k
 
     def express_interest(self,
                          name: NonStrictName,
-                         app_param: Optional[BinaryStr] = None,
-                         validator: Optional[Validator] = None,
+                         app_param: BinaryStr | None = None,
+                         validator: Validator | None = None,
                          need_raw_packet: bool = False,
-                         **kwargs) -> Coroutine[Any, None, Tuple[FormalName, MetaInfo, Optional[BinaryStr]]]:
+                         **kwargs) -> Coroutine[Any, None, tuple[FormalName, MetaInfo, BinaryStr | None]]:
         r"""
         Express an Interest packet.
 
@@ -226,9 +227,9 @@ def express_raw_interest(self,
                              final_name: NonStrictName,
                              interest_param: InterestParam,
                              raw_interest: BinaryStr,
-                             validator: Optional[Validator] = None,
+                             validator: Validator | None = None,
                              need_raw_packet: bool = False
-                             ) -> Coroutine[Any, None, Tuple[FormalName, MetaInfo, Optional[BinaryStr]]]:
+                             ) -> Coroutine[Any, None, tuple[FormalName, MetaInfo, BinaryStr | None]]:
         final_name = Name.normalize(final_name)
         future = aio.get_running_loop().create_future()
         if Component.get_type(final_name[-1]) == Component.TYPE_IMPLICIT_SHA256:
@@ -247,7 +248,7 @@ async def _wait_for_data(self, future: aio.Future, lifetime: int, node_name: For
         lifetime = 100 if lifetime is None else lifetime
         try:
             data_name, meta_info, content, sig, raw_packet = await aio.wait_for(future, timeout=lifetime/1000.0)
-        except aio.TimeoutError:
+        except TimeoutError:
             if node.timeout(future):
                 del self._int_tree[node_name]
             raise InterestTimeout()
@@ -338,7 +339,7 @@ def run_forever(self, after_start: Awaitable = None):
         except KeyboardInterrupt:
             self.logger.info('Receiving Ctrl+C, exit')
 
-    def route(self, name: NonStrictName, validator: Optional[Validator] = None,
+    def route(self, name: NonStrictName, validator: Validator | None = None,
               need_raw_packet: bool = False, need_sig_ptrs: bool = False):
         """
         A decorator used to register a permanent route for a specific prefix.
@@ -395,7 +396,7 @@ def decorator(func: Route):
             return func
         return decorator
 
-    async def register(self, name: NonStrictName, func: Optional[Route], validator: Optional[Validator] = None,
+    async def register(self, name: NonStrictName, func: Route | None, validator: Validator | None = None,
                        need_raw_packet: bool = False, need_sig_ptrs: bool = False) -> bool:
         """
         Register a route for a specific prefix dynamically.
@@ -431,15 +432,15 @@ async def register(self, name: NonStrictName, func: Optional[Route], validator:
                     lifetime=1000)
                 ret = parse_response(reply)
                 if ret['status_code'] != 200:
-                    self.logger.error(f'Registration for {Name.to_str(name)} failed: '
-                                      f'{ret["status_code"]} {ret["status_text"]}')
+                    self.logger.error('Registration for %s failed: %s %s',
+                                      Name.to_str(name), ret["status_code"], ret["status_text"])
                     return False
                 else:
-                    self.logger.debug(f'Registration for {Name.to_str(name)} succeeded: '
-                                      f'{ret["status_code"]} {ret["status_text"]}')
+                    self.logger.debug('Registration for %s succeeded: %s %s',
+                                      Name.to_str(name), ret["status_code"], ret["status_text"])
                     return True
             except (InterestNack, InterestTimeout, InterestCanceled, ValidationFailure) as e:
-                self.logger.error(f'Registration for {Name.to_str(name)} failed: {e.__class__.__name__}')
+                self.logger.error('Registration for %s failed: %s', Name.to_str(name), e.__class__.__name__)
                 return False
 
     async def unregister(self, name: NonStrictName) -> bool:
@@ -458,7 +459,7 @@ async def unregister(self, name: NonStrictName) -> bool:
             return False
 
     def set_interest_filter(self, name: NonStrictName, func: Route,
-                            validator: Optional[Validator] = None, need_raw_packet: bool = False,
+                            validator: Validator | None = None, need_raw_packet: bool = False,
                             need_sig_ptrs: bool = False):
         """
         Set the callback function for an Interest prefix without sending a register command to the forwarder.
@@ -500,7 +501,7 @@ def _on_nack(self, name: FormalName, nack_reason: int):
                 del self._int_tree[name]
 
     async def _on_data(self, name: FormalName, meta_info: MetaInfo,
-                       content: Optional[BinaryStr], sig: SignaturePtrs, raw_packet):
+                       content: BinaryStr | None, sig: SignaturePtrs, raw_packet):
         clean_list = []
         for prefix, node in self._int_tree.prefixes(name):
             if node.satisfy((name, meta_info, content, sig, raw_packet), prefix != name):
@@ -509,18 +510,18 @@ async def _on_data(self, name: FormalName, meta_info: MetaInfo,
             del self._int_tree[prefix]
 
     async def _on_interest(self, name: FormalName, param: InterestParam,
-                           app_param: Optional[BinaryStr], sig: SignaturePtrs, raw_packet: BinaryStr):
+                           app_param: BinaryStr | None, sig: SignaturePtrs, raw_packet: BinaryStr):
         trie_step = self._prefix_tree.longest_prefix(name)
         if not trie_step:
-            self.logger.warning('No route: %s' % name)
+            self.logger.warning('No route: %s', name)
             return
         node = trie_step.value
         if node.callback is None:
-            self.logger.warning('No callback: %s' % name)
+            self.logger.warning('No callback: %s', name)
             return
         if app_param is not None or sig.signature_info is not None:
             if not await params_sha256_checker(name, sig):
-                self.logger.warning('Drop malformed Interest: %s' % name)
+                self.logger.warning('Drop malformed Interest: %s', name)
                 return
 
         # In case the validator blocks the pipeline, create a task
@@ -531,7 +532,7 @@ async def submit_interest():
             else:
                 valid = True
             if not valid:
-                self.logger.warning('Drop unvalidated Interest: %s' % name)
+                self.logger.warning('Drop unvalidated Interest: %s', name)
                 return
             if node.extra_param:
                 kwargs = {}

src/ndn/app_support/dispatcher.py

@@ -15,7 +15,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # -----------------------------------------------------------------------------
-from typing import Optional
 from ..encoding import NonStrictName, Name, BinaryStr, InterestParam, FormalName
 from ..types import Route
 from ..name_tree import NameTrie, PrefixTreeNode
@@ -54,7 +53,7 @@ def unregister(self, name: NonStrictName):
         name = Name.normalize(name)
         del self._tree[name]
 
-    def dispatch(self, name: FormalName, param: InterestParam, app_param: Optional[BinaryStr]) -> bool:
+    def dispatch(self, name: FormalName, param: InterestParam, app_param: BinaryStr | None) -> bool:
         """
         Dispatch the Interest to registered callbacks using longest match.
 

src/ndn/app_support/keychain_register.py

@@ -33,7 +33,7 @@ def on_int(self, int_name: enc.FormalName, _app_param, reply: ReplyFunc, pkt_ctx
         can_be_prefix = pkt_ctx['int_param'].can_be_prefix
         # can_be_prefix = True if using KEY name, False if using CERT name
         if len(int_name) != len(id_name) + (2 if can_be_prefix else 4):
-            logger.warning(f'Invalid key fetching Interest: {enc.Name.to_str(int_name)}')
+            logger.warning('Invalid key fetching Interest: %s', enc.Name.to_str(int_name))
             return
         try:
             key_name = int_name[:len(id_name)+2]
@@ -47,12 +47,12 @@ def on_int(self, int_name: enc.FormalName, _app_param, reply: ReplyFunc, pkt_ctx
             else:
                 cert = key[int_name]
             if cert is not None:
-                logger.info(f'KeychainRegister replied with: {enc.Name.to_str(cert.name)}')
+                logger.info('KeychainRegister replied with: %s', enc.Name.to_str(cert.name))
                 reply(cert.data)
             else:
-                logger.warning(f'No certificate for key: {enc.Name.to_str(int_name)}')
+                logger.warning('No certificate for key: %s', enc.Name.to_str(int_name))
         except KeyError:
-            logger.warning(f'Fetching not existing key/cert: {enc.Name.to_str(int_name)}')
+            logger.warning('Fetching not existing key/cert: %s', enc.Name.to_str(int_name))
 
     def __init__(self, ident: sec.AbstractIdentity):
         self.ident = ident

src/ndn/app_support/light_versec/checker.py

@@ -22,7 +22,7 @@
 # -----------------------------------------------------------------------------
 from __future__ import annotations
 
-from typing import Callable, Iterator
+from collections.abc import Callable, Iterator
 
 from ...encoding import BinaryStr, Component, FormalName, Name, NonStrictName
 from ...security import Keychain