chore(pynumaflow): drop Python 3.10 and add Python 3.15 support

.github/workflows/coverage.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10"]
+        python-version: ["3.11"]
 
     steps:
     - uses: actions/checkout@v3

.github/workflows/lint.yml

@@ -20,7 +20,7 @@ jobs:
         uses: astral-sh/setup-uv@v4
 
       - name: Set up Python
-        run: uv python install 3.10
+        run: uv python install 3.11
 
       - name: Install dependencies
         run: uv sync --group dev

.github/workflows/pypi.yaml

@@ -15,7 +15,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: [ "3.10" ]
+        python-version: [ "3.11" ]
 
     name: Publish to PyPi
     steps:

.github/workflows/run-tests.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"]
 
     steps:
     - uses: actions/checkout@v3

.pre-commit-config.yaml

@@ -1,14 +1,14 @@
 default_language_version:
-    python: python3.10
+    python: python3.11
 repos:
 - repo: https://github.com/psf/black
-  rev: 23.3.0
+  rev: 26.10.1
   hooks:
   -   id: black
-      language_version: python3.10
+      language_version: python3.11
       args: [--config=pyproject.toml, --diff, --color ]
-- repo: https://github.com/charliermarsh/ruff-pre-commit
-  rev: 'v0.0.264'
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: 'v0.17.0'
   hooks:
     - id: ruff
 - repo: https://github.com/adamchainz/blacken-docs

packages/pynumaflow/docs/contributing.md

@@ -6,7 +6,7 @@ Thank you for your interest in contributing to pynumaflow! This guide will help
 
 ### Prerequisites
 
-- Python 3.10 or higher
+- Python 3.11 or higher
 - [uv](https://docs.astral.sh/uv/) for dependency management
 - Git
 

packages/pynumaflow/examples/batchmap/concurrent_sink/example.py

@@ -121,7 +121,8 @@ async def submit_task(datum: Datum):
             """Submit a single message to worker pool."""
             task_data = (datum.id, datum.keys, datum.value)
             fut = loop.run_in_executor(self.executor, _process_single_message_task, task_data)
-            return await asyncio.wait_for(fut, timeout=30.0)
+            async with asyncio.timeout(30.0):
+                return await fut
 
         # Wait for all results
         results = await asyncio.gather(

packages/pynumaflow/pynumaflow/_constants.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from enum import Enum
+from enum import StrEnum
 
 from pynumaflow import setup_logging
 
@@ -73,7 +73,7 @@
     _LOGGER.setLevel(logging.DEBUG)
 
 
-class UDFType(str, Enum):
+class UDFType(StrEnum):
     """
     Enumerate the type of UDF.
     """

packages/pynumaflow/pynumaflow/accumulator/_dtypes.py

@@ -3,15 +3,13 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import IntEnum
-from typing import TypeAlias, TypeVar
+from typing import Self, TypeAlias
 from collections.abc import AsyncIterable, Callable
 
 from pynumaflow.shared.asynciter import NonBlockingIterator
 from pynumaflow._constants import DROP
 from pynumaflow._validate import _validate_message_fields
 
-M = TypeVar("M", bound="Message")
-
 
 class WindowOperation(IntEnum):
     """
@@ -37,13 +35,13 @@ class Datum:
     Example usage
     ```py
     from pynumaflow.accumulator import Datum
-    from datetime import datetime, timezone
+    from datetime import UTC, datetime
 
     d = Datum(
           keys=["test_key"],
           value=b"test_mock_message",
-          event_time=datetime.fromtimestamp(1662998400, timezone.utc),
-          watermark=datetime.fromtimestamp(1662998460, timezone.utc),
+          event_time=datetime.fromtimestamp(1662998400, UTC),
+          watermark=datetime.fromtimestamp(1662998460, UTC),
           headers={"key1": "value1", "key2": "value2"},
        )
     ```
@@ -418,7 +416,7 @@ def __init__(
         self._id = id or ""
 
     @classmethod
-    def to_drop(cls: type[M], datum: Datum) -> M:
+    def to_drop(cls, datum: Datum) -> Self:
         """Creates a Message from the given Datum with the DROP tag set, so the message is not
         forwarded to the next vertex but still allows the accumulator to advance the watermark
         and release the tracked state.
@@ -431,7 +429,7 @@ def to_drop(cls: type[M], datum: Datum) -> M:
             datum: The Datum to drop the results for.
 
         Returns:
-            M: A Message instance with empty value and the DROP tag set, carrying the datum's
+            Message: A Message instance with empty value and the DROP tag set, carrying the datum's
                watermark, event time, id and headers.
         """
         return cls(

packages/pynumaflow/pynumaflow/accumulator/servicer/task_manager.py

@@ -1,6 +1,6 @@
 import asyncio
 from collections.abc import AsyncIterable
-from datetime import datetime, timezone
+from datetime import datetime, UTC
 
 from google.protobuf import timestamp_pb2
 from pynumaflow._constants import (
@@ -322,7 +322,7 @@ async def write_to_global_queue(
                 watermark_pb.FromDatetime(msg.watermark)
 
             start_dt_pb = timestamp_pb2.Timestamp()
-            start_dt_pb.FromDatetime(datetime.fromtimestamp(0, tz=timezone.utc))
+            start_dt_pb.FromDatetime(datetime.fromtimestamp(0, tz=UTC))
 
             end_dt_pb = timestamp_pb2.Timestamp()
             end_dt_pb.FromDatetime(wm)
@@ -365,7 +365,7 @@ async def write_to_global_queue(
             # stream_send_eof on SIGTERM): synthesize the window from epoch(0) and the
             # latest watermark, preserving prior behavior.
             start_eof_pb = timestamp_pb2.Timestamp()
-            start_eof_pb.FromDatetime(datetime.fromtimestamp(0, tz=timezone.utc))
+            start_eof_pb.FromDatetime(datetime.fromtimestamp(0, tz=UTC))
 
             end_eof_pb = timestamp_pb2.Timestamp()
             end_eof_pb.FromDatetime(wm)

packages/pynumaflow/pynumaflow/batchmapper/_dtypes.py

@@ -2,7 +2,7 @@
 from collections.abc import Iterator, Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import TypeAlias, TypeVar
+from typing import Self, TypeAlias, TypeVar
 from collections.abc import AsyncIterable, Callable
 
 from pynumaflow._constants import DROP, FAIL, NACK
@@ -42,17 +42,17 @@ def __init__(self, value: bytes, keys: list[str] | None = None, tags: list[str]
 
     # returns the Message Object which will be dropped
     @classmethod
-    def to_drop(cls: type[M]) -> M:
+    def to_drop(cls) -> Self:
         return cls(b"", None, [DROP])
 
     @classmethod
-    def to_nack(cls: type[M], opts: NackOptions | None = None) -> M:
+    def to_nack(cls, opts: NackOptions | None = None) -> Self:
         m = cls(b"", None, [NACK])
         m._nack_options = opts
         return m
 
     @classmethod
-    def to_fail(cls: type[M]) -> M:
+    def to_fail(cls) -> Self:
         return cls(b"", None, [FAIL])
 
     @property

packages/pynumaflow/pynumaflow/info/types.py

@@ -1,7 +1,7 @@
 import logging
 import os
 from dataclasses import dataclass, field
-from enum import Enum
+from enum import StrEnum
 from importlib.metadata import version
 from typing import TypeVar
 
@@ -30,7 +30,7 @@
 SI = TypeVar("SI", bound="ServerInfo")
 
 
-class Protocol(str, Enum):
+class Protocol(StrEnum):
     """
     Enumerate grpc server connection protocol.
     """
@@ -39,15 +39,15 @@ class Protocol(str, Enum):
     TCP = "tcp"
 
 
-class Language(str, Enum):
+class Language(StrEnum):
     """
     Enumerate Numaflow SDK language.
     """
 
     PYTHON = "python"
 
 
-class MapMode(str, Enum):
+class MapMode(StrEnum):
     """
     Enumerate Map Mode to be enabled
     """
@@ -57,7 +57,7 @@ class MapMode(str, Enum):
     BatchMap = "batch-map"
 
 
-class ContainerType(str, Enum):
+class ContainerType(StrEnum):
     """
     Enumerate Numaflow container type.
     """

packages/pynumaflow/pynumaflow/mapper/_dtypes.py

@@ -2,7 +2,7 @@
 from collections.abc import Iterator, Sequence, Awaitable
 from dataclasses import dataclass
 from datetime import datetime
-from typing import TypeAlias, TypeVar
+from typing import Self, TypeAlias, TypeVar
 from collections.abc import Callable
 from warnings import warn
 
@@ -52,17 +52,17 @@ def __init__(
 
     # returns the Message Object which will be dropped
     @classmethod
-    def to_drop(cls: type[M]) -> M:
+    def to_drop(cls) -> Self:
         return cls(b"", None, [DROP])
 
     @classmethod
-    def to_nack(cls: type[M], opts: NackOptions | None = None) -> M:
+    def to_nack(cls, opts: NackOptions | None = None) -> Self:
         m = cls(b"", None, [NACK])
         m._nack_options = opts
         return m
 
     @classmethod
-    def to_fail(cls: type[M]) -> M:
+    def to_fail(cls) -> Self:
         return cls(b"", None, [FAIL])
 
     @property
@@ -144,13 +144,13 @@ class Datum:
     Example usage
     ```py
     from pynumaflow.mapper import Datum
-    from datetime import datetime, timezone
+    from datetime import UTC, datetime
 
     d = Datum(
           keys=["test_key"],
           value=b'test_mock_message'
-          event_time=datetime.fromtimestamp(1662998400, timezone.utc),
-          watermark=datetime.fromtimestamp(1662998460, timezone.utc),
+          event_time=datetime.fromtimestamp(1662998400, UTC),
+          watermark=datetime.fromtimestamp(1662998460, UTC),
           headers={"key1": "value1", "key2": "value2"},
        )
     ```

packages/pynumaflow/pynumaflow/mapstreamer/_dtypes.py

@@ -2,7 +2,7 @@
 from collections.abc import Iterator, Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import TypeAlias, TypeVar
+from typing import Self, TypeAlias, TypeVar
 from collections.abc import AsyncIterable, Callable
 from warnings import warn
 
@@ -42,17 +42,17 @@ def __init__(self, value: bytes, keys: list[str] | None = None, tags: list[str]
 
     # returns the Message Object which will be dropped
     @classmethod
-    def to_drop(cls: type[M]) -> M:
+    def to_drop(cls) -> Self:
         return cls(b"", None, [DROP])
 
     @classmethod
-    def to_nack(cls: type[M], opts: NackOptions | None = None) -> M:
+    def to_nack(cls, opts: NackOptions | None = None) -> Self:
         m = cls(b"", None, [NACK])
         m._nack_options = opts
         return m
 
     @classmethod
-    def to_fail(cls: type[M]) -> M:
+    def to_fail(cls) -> Self:
         return cls(b"", None, [FAIL])
 
     @property
@@ -130,13 +130,13 @@ class Datum:
     Example:
     ```py
     from pynumaflow.mapstreamer import Datum
-    from datetime import datetime, timezone
+    from datetime import UTC, datetime
 
     d = Datum(
           keys=["test_key"],
           value=b"test_mock_message",
-          event_time=datetime.fromtimestamp(1662998400, timezone.utc),
-          watermark=datetime.fromtimestamp(1662998460, timezone.utc),
+          event_time=datetime.fromtimestamp(1662998400, UTC),
+          watermark=datetime.fromtimestamp(1662998460, UTC),
           headers={"key1": "value1", "key2": "value2"},
        )
     ```

packages/pynumaflow/pynumaflow/reducer/_dtypes.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import IntEnum
-from typing import TypeAlias, TypeVar
+from typing import Self, TypeAlias, TypeVar
 from collections.abc import AsyncIterable, Callable
 from warnings import warn
 
@@ -56,7 +56,7 @@ def __init__(self, value: bytes, keys: list[str] | None = None, tags: list[str]
 
     # returns the Message Object which will be dropped
     @classmethod
-    def to_drop(cls: type[M]) -> M:
+    def to_drop(cls) -> Self:
         return cls(b"", None, [DROP])
 
     @property
@@ -130,13 +130,13 @@ class Datum:
     Example usage
     ```py
     from pynumaflow.reducer import Datum
-    from datetime import datetime, timezone
+    from datetime import UTC, datetime
 
     d = Datum(
           keys=["test_key"],
           value=b"test_mock_message",
-          event_time=datetime.fromtimestamp(1662998400, timezone.utc),
-          watermark=datetime.fromtimestamp(1662998460, timezone.utc),
+          event_time=datetime.fromtimestamp(1662998400, UTC),
+          watermark=datetime.fromtimestamp(1662998460, UTC),
           headers={"key1": "value1", "key2": "value2"},
        )
     ```

packages/pynumaflow/pynumaflow/reducer/servicer/task_manager.py

@@ -1,5 +1,5 @@
 import asyncio
-from datetime import datetime, timezone
+from datetime import datetime, UTC
 from collections.abc import AsyncIterable
 
 from pynumaflow.exceptions import UDFError
@@ -153,8 +153,8 @@ async def __invoke_reduce(
         new_instance = self.__reduce_handler
 
         # Convert the window to a datetime object
-        start_dt = datetime.fromtimestamp(int(window.start.ToMilliseconds()) / 1e3, timezone.utc)
-        end_dt = datetime.fromtimestamp(int(window.end.ToMilliseconds()) / 1e3, timezone.utc)
+        start_dt = datetime.fromtimestamp(int(window.start.ToMilliseconds()) / 1e3, UTC)
+        end_dt = datetime.fromtimestamp(int(window.end.ToMilliseconds()) / 1e3, UTC)
         interval_window = IntervalWindow(start_dt, end_dt)
         md = Metadata(interval_window=interval_window)
         # If the reduce handler is a class instance, create a new instance of it.