pyton312 and typing

.github/workflows/ci.yml

@@ -11,11 +11,11 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
-      - name: Set up Python 3.9
-        uses: actions/setup-python@v4
+      - uses: actions/checkout@v4
+      - name: Set up Python 3.12
+        uses: actions/setup-python@v5
         with:
-          python-version: 3.9
+          python-version: 3.12
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.9
+FROM python:3.12
 ADD . /code
 WORKDIR /code
 RUN pip install --no-cache-dir -r requirements.txt

lib/ipflow/field.py

@@ -1,3 +1,4 @@
+from typing import Union
 from .field_type import FIELD_TYPE_FMT
 from .field_type import FieldType
 
@@ -15,7 +16,7 @@ def __init__(self, field_id: int, length: int):
         self.length = length
 
     @property
-    def name(self) -> str:
+    def name(self) -> Union[str, None]:
         # used for serializing flows, unknown fields will be dropped later
         try:
             return FieldType(self.id).name.lower()

lib/ipflow/field_type.py

@@ -35,7 +35,7 @@ class FieldType(Enum):
     # natEvent = 230
 
 
-FIELD_TYPE_FMT = {
+FIELD_TYPE_FMT: dict[tuple[int, int], str] = {
     (1, 4): 'L',
     (1, 8): 'Q',  # also uint64
     (2, 4): 'L',

lib/ipflow/flow.py

@@ -1,6 +1,6 @@
 import struct
 from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network
-from typing import Dict, Iterator, Tuple, Union
+from typing import Iterator, Union, Any
 from .field_type import FIELD_TYPE_FUNC
 from .field_type import FieldType
 from .template import DataTemplate
@@ -10,7 +10,8 @@
 V5_TEMPLATE_FMT = '>4s4s4sHHLLLLHH2sBBB3s4s'
 V5_TEMPLATE_SIZE = struct.calcsize(V5_TEMPLATE_FMT)
 
-flowset_templates: Dict[Tuple[str, int, int], DataTemplate] = {
+DataTemplateKey = Any  # TODO v5 key should should be also tuple[str, int, int]
+flowset_templates: dict[Any, DataTemplate] = {
     V5_TEMPLATE_KEY: DataTemplate(
         V5_TEMPLATE_FMT,
         V5_TEMPLATE_SIZE,
@@ -45,12 +46,16 @@ class Flow:
         'values',
     )
 
-    def __init__(self, template: DataTemplate, values: Tuple[bytes]):
+    def __init__(self, template: DataTemplate, values: tuple[bytes, ...]):
         self.template = template
         self.values = values
 
     def serialize(self) -> dict:
-        fields = self.template.fields
+        '''
+        returns a serialized representation
+        fields without a formatter are omitted from the result
+        '''
+        fields = [f for f in self.template.fields if f.fmt]
         return {
             f.name: FIELD_TYPE_FUNC.get(f.id, lambda val: val)(val)
             for f, val in zip(fields, self.values)
@@ -59,7 +64,7 @@ def serialize(self) -> dict:
     def test_address(
         self,
         address: Union[IPv4Address, IPv6Address]
-    ) -> Iterator[Union[IPv4Address, IPv6Address]]:
+    ) -> bool:
         fields_idx = self.template.index
         for ft in (
             FieldType.IPV4_DST_ADDR,
@@ -80,11 +85,12 @@ def test_address(
                 addr = IPv6Address(self.values[fields_idx.index(ft.value)])
                 if addr == address:
                     return True
+        return False
 
     def test_network(
         self,
         network: Union[IPv4Network, IPv6Network]
-    ) -> Iterator[Union[IPv4Network, IPv6Network]]:
+    ) -> Iterator[Union[IPv4Address, IPv6Address]]:
         fields_idx = self.template.index
         for ft in (
             FieldType.IPV4_DST_ADDR,

lib/ipflow/template.py

@@ -1,6 +1,7 @@
 import struct
-from typing import List
+from typing import Union
 from .field import Field
+from .field_type import FieldType
 
 
 class DataTemplate:
@@ -12,8 +13,8 @@ class DataTemplate:
         'source_uptime',
     )
 
-    def __init__(self, fmt: str, length: int, fields: List[Field],
-                 index: List[int], source_uptime: int):
+    def __init__(self, fmt: str, length: int, fields: list[Field],
+                 index: list[Union[int, FieldType, None]], source_uptime: int):
         self.fmt = struct.Struct(fmt)
         self.fields = fields
         self.index = index

lib/server.py

@@ -45,12 +45,13 @@ def datagram_received(self, data, addr):
                 subs.on_flow(flow)
 
 
-async def start_server():
+def start_server(loop: asyncio.AbstractEventLoop):
     logging.info('Starting UDP server')
-    loop = asyncio.get_event_loop()
 
-    transport, protocol = await loop.create_datagram_endpoint(
-        ServerProtocol,
-        local_addr=('0.0.0.0', LISTEN_PORT))
+    transport, protocol = loop.run_until_complete(
+        loop.create_datagram_endpoint(
+            ServerProtocol,
+            local_addr=('0.0.0.0', LISTEN_PORT))
+    )
 
     return transport, protocol

lib/state.py

@@ -3,7 +3,7 @@
 import time
 import logging
 from ipaddress import IPv4Address, IPv6Address
-from typing import Dict, Optional, Tuple, Union
+from typing import Union
 from .subscription import Subscription
 
 
@@ -21,7 +21,7 @@ def subscribe_check(
     subscriptions[(asset_id, check_key, address)] = Subscription.make(address)
 
 
-def get_host_by_addr(address: str) -> Optional[str]:
+def get_host_by_addr(address: str) -> Union[str, None]:
     host, expire_ts = host_lk.get(address, (None, None))
 
     # request new name when no in lookup or aged
@@ -47,5 +47,6 @@ async def cleanup_subscriptions_loop():
         await asyncio.sleep(60)
 
 
-host_lk: Dict[str, Tuple[str, float]] = {}
-subscriptions: Dict[int, Subscription] = {}
+host_lk: dict[str, tuple[Union[str, None], float]] = {}
+subscriptions: dict[tuple[int, str, Union[IPv4Address, IPv6Address]],
+                    Subscription] = {}

lib/subscription.py

@@ -1,12 +1,12 @@
 import time
 from ipaddress import IPv4Address, IPv6Address
-from typing import List, NamedTuple, Union
+from typing import NamedTuple, Union
 from .ipflow.flow import Flow
 
 
 class Subscription(NamedTuple):
     address: Union[IPv4Address, IPv6Address]
-    result: List[Flow]
+    result: list[Flow]
     timestamp: int
 
     @classmethod

main.py

@@ -7,14 +7,14 @@
 
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    loop.run_until_complete(start_server())
-    asyncio.ensure_future(cleanup_subscriptions_loop())
+    loop = asyncio.new_event_loop()
+    start_server(loop)
+    asyncio.ensure_future(cleanup_subscriptions_loop(), loop=loop)
 
     checks = {
         'ipflow': check_ipflow,
     }
 
     probe = Probe("ipflow", version, checks)
 
-    probe.start()
+    probe.start(loop=loop)

requirements.txt

@@ -1,2 +1 @@
-aiohttp==3.10.3
-libprobe==0.2.36
+libprobe==0.3.0