BulkBeing · Oct 10, 2026 · Oct 10, 2026 · Oct 10, 2026
@@ -42,15 +42,10 @@ jobs:
     steps:
       - uses: actions/checkout@v7
 
-      - name: Set up Python
-        uses: actions/setup-python@v6
-        with:
-          python-version: "3.10"
-
       - name: Install uv
-        uses: astral-sh/setup-uv@v8.1.0
+        uses: astral-sh/setup-uv@v10.3.0
         with:
-          python-version: "3.10"
+          python-version: "3.11"
           working-directory: packages/pynumaflow-lite
 
       - name: Set up Rust
@@ -94,7 +89,7 @@ jobs:
       - uses: actions/checkout@v7
 
       - name: Install uv
-        uses: astral-sh/setup-uv@v8.1.0
+        uses: astral-sh/setup-uv@v10.3.0
         with:
           working-directory: packages/pynumaflow-lite
 
@@ -108,7 +103,7 @@ jobs:
         run: |
           set -euo pipefail
 
-          for python_version in 3.10 3.11 3.12 3.13 3.14; do
+          for python_version in 3.11 3.12 3.13 3.14 3.15; do
             echo "::group::Python ${python_version}"
             uv python install "${python_version}"
             export UV_PYTHON="${python_version}"
@@ -150,7 +145,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"]
         platform:
           - os: ubuntu-24.04
             target: x86_64-unknown-linux-gnu
@@ -166,10 +161,23 @@ jobs:
     steps:
       - uses: actions/checkout@v7
 
-      - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v6
+      - name: Install uv
+        uses: astral-sh/setup-uv@v10.3.0
         with:
-          python-version: ${{ matrix.python-version }}
+          working-directory: packages/pynumaflow-lite
+
+      # On Linux, maturin builds in the manylinux container and uses the
+      # python3.x of that container. On macOS, it builds on the host, so it
+      # needs an interpreter from uv.
+      - name: Set up Python ${{ matrix.python-version }}
+        if: ${{ !matrix.platform.linux }}
+        env:
+          PYTHON_VERSION: ${{ matrix.python-version }}
+        run: |
+          set -euo pipefail
+
+          uv python install "${PYTHON_VERSION}"
+          echo "PYTHON_BIN=$(uv python find --no-project "${PYTHON_VERSION}")" >> "${GITHUB_ENV}"
 
       - name: Set up Rust
         run: |
@@ -182,7 +190,7 @@ jobs:
           set -euo pipefail
 
           tag_version="${GITHUB_REF_NAME#pynumaflow-lite-v}"
-          package_version="$(cargo metadata --locked --no-deps --format-version 1 | python -c 'import json, sys; print(json.load(sys.stdin)["packages"][0]["version"])')"
+          package_version="$(cargo metadata --locked --no-deps --format-version 1 | uv run --no-project python -c 'import json, sys; print(json.load(sys.stdin)["packages"][0]["version"])')"
 
           if [ "${package_version}" != "${tag_version}" ]; then
             echo "Tag version ${tag_version} does not match Cargo.toml version ${package_version}"
@@ -194,15 +202,15 @@ jobs:
         run: |
           set -euo pipefail
 
-          base_version="$(cargo metadata --locked --no-deps --format-version 1 | python -c 'import json, sys; print(json.load(sys.stdin)["packages"][0]["version"])')"
+          base_version="$(cargo metadata --locked --no-deps --format-version 1 | uv run --no-project python -c 'import json, sys; print(json.load(sys.stdin)["packages"][0]["version"])')"
           if [[ "${base_version}" == *-* ]]; then
             manual_version="${base_version}.dev.${GITHUB_RUN_ID}${GITHUB_RUN_ATTEMPT}"
           else
             manual_version="${base_version}-dev.${GITHUB_RUN_ID}${GITHUB_RUN_ATTEMPT}"
           fi
           export MANUAL_VERSION="${manual_version}"
 
-          python - <<'PY'
+          uv run --no-project python - <<'PY'
           import os
           import re
           from pathlib import Path
@@ -244,7 +252,7 @@ jobs:
             --release
             --locked
             --out dist
-            -i ${{ matrix.platform.linux && format('python{0}', matrix.python-version) || 'python' }}
+            -i ${{ matrix.platform.linux && format('python{0}', matrix.python-version) || env.PYTHON_BIN }}
 
       - name: Upload wheel artifact
         uses: actions/upload-artifact@v7
@@ -268,10 +276,10 @@ jobs:
 
     steps:
       - name: Install uv
-        uses: astral-sh/setup-uv@v8.1.0
+        uses: astral-sh/setup-uv@v10.3.0
 
       - name: Download wheel artifacts
-        uses: actions/download-artifact@v7
+        uses: actions/download-artifact@v8
         with:
           pattern: pynumaflow-lite-*
           path: dist

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bullseye AS builder
+FROM python:3.11-slim-trixie AS builder
 
 ENV PYTHONFAULTHANDLER=1 \
   PYTHONUNBUFFERED=1 \

@@ -8,7 +8,7 @@
 import asyncio
 import signal
 from collections.abc import AsyncIterator
-from datetime import datetime
+from datetime import UTC, datetime
 
 from pynumaflow_lite.accumulator import (
     Accumulator,
@@ -25,10 +25,8 @@ class StreamSorter(Accumulator):
     """
 
     def __init__(self):
-        from datetime import timezone
-
         # Initialize with a very old timestamp (timezone-aware)
-        self.latest_wm = datetime.fromtimestamp(-1, tz=timezone.utc)
+        self.latest_wm = datetime.fromtimestamp(-1, tz=UTC)
         self.sorted_buffer: list[Datum] = []
         print("StreamSorter initialized")
 

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bullseye AS builder
+FROM python:3.11-slim-trixie AS builder
 
 ENV PYTHONFAULTHANDLER=1 \
   PYTHONUNBUFFERED=1 \

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bullseye AS builder
+FROM python:3.11-slim-trixie AS builder
 
 ENV PYTHONFAULTHANDLER=1 \
   PYTHONUNBUFFERED=1 \

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bullseye AS builder
+FROM python:3.11-slim-trixie AS builder
 
 ENV PYTHONFAULTHANDLER=1 \
   PYTHONUNBUFFERED=1 \

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bullseye AS builder
+FROM python:3.11-slim-trixie AS builder
 
 ENV PYTHONFAULTHANDLER=1 \
   PYTHONUNBUFFERED=1 \

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bullseye AS builder
+FROM python:3.11-slim-trixie AS builder
 
 ENV PYTHONFAULTHANDLER=1 \
   PYTHONUNBUFFERED=1 \

@@ -2,7 +2,7 @@
 import logging
 import signal
 from collections.abc import AsyncIterator
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from pynumaflow_lite import sourcer
 from pynumaflow_lite._source_dtypes import Sourcer
@@ -43,7 +43,7 @@ async def read_handler(self, datum: sourcer.ReadRequest) -> AsyncIterator[source
             message = sourcer.Message(
                 payload=payload,
                 offset=offset,
-                event_time=datetime.now(timezone.utc),
+                event_time=datetime.now(UTC),
                 keys=["key1"],
                 headers={"source": "simple"},
             )

@@ -1,10 +1,10 @@
 import asyncio
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from pynumaflow_lite.sourcetransformer import Datum, Message, SourceTransformAsyncServer, SourceTransformer
 
-january_first_2022 = datetime(2022, 1, 1, tzinfo=timezone.utc)
-january_first_2023 = datetime(2023, 1, 1, tzinfo=timezone.utc)
+january_first_2022 = datetime(2022, 1, 1, tzinfo=UTC)
+january_first_2023 = datetime(2023, 1, 1, tzinfo=UTC)
 
 
 class EventFilter(SourceTransformer):

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import signal
+from collections.abc import Awaitable, Callable
+from types import TracebackType
+from typing import Protocol, Self
+
+_SHUTDOWN_SIGNALS = (signal.SIGINT, signal.SIGTERM)
+
+
+class _ServerCore(Protocol):
+    """The Rust server object that a `_AsyncServer` drives."""
+
+    def start(self, handler: Callable[..., object]) -> Awaitable[None]: ...
+    def wait_ready(self) -> Awaitable[None]: ...
+    def stop(self) -> None: ...
+
+
+class _AsyncServer:
+    """Lifecycle shared by the async UDF servers.
+
+    A subclass sets `_name` for error messages, and gives its Rust core object
+    and the user handler to `__init__`.
+    """
+
+    _name: str
+
+    def __init__(
+        self,
+        core: _ServerCore,
+        handler: Callable[..., object],
+        *,
+        install_signal_handlers: bool,
+    ) -> None:
+        self._core = core
+        self._handler = handler
+        self._install_signal_handlers = install_signal_handlers
+        self._task: asyncio.Task[None] | None = None
+        self._serving = False
+        self._installed_signals: list[signal.Signals] = []
+
+    async def serve(self) -> None:
+        """Run the server until it stops.
+
+        This is the entrypoint for an application that already runs an event
+        loop. It returns when a shutdown signal arrives or when `stop()` runs.
+        """
+        await self._serve(install_signal_handlers=self._install_signal_handlers)
+
+    async def _serve(self, *, install_signal_handlers: bool) -> None:
+        if self._serving:
+            raise RuntimeError(f"{self._name} server is already serving")
+        self._serving = True
+        try:
+            if install_signal_handlers:
+                self._add_signal_handlers()
+            await self._core.start(self._handler)
+        finally:
+            self._remove_signal_handlers()
+            self._serving = False
+
+    def stop(self) -> None:
+        self._core.stop()
+
+    async def wait_ready(self) -> None:
+        """Wait until the server answers the Numaflow readiness probe.
+
+        There is no time limit. Use `asyncio.timeout()` to set one.
+        """
+        await self._core.wait_ready()
+
+    async def wait_for_termination(self) -> None:
+        """Wait until the background server task ends.
+
+        Use this inside an `async with` block. It raises the handler error if
+        the server task failed.
+        """
+        if self._task is None:
+            raise RuntimeError(f"{self._name} server is not serving")
+        await asyncio.shield(self._task)
+
+    def _add_signal_handlers(self) -> None:
+        loop = asyncio.get_running_loop()
+        for sig in _SHUTDOWN_SIGNALS:
+            try:
+                loop.add_signal_handler(sig, self.stop)
+            except (NotImplementedError, RuntimeError, OSError):
+                continue
+            self._installed_signals.append(sig)
+
+    def _remove_signal_handlers(self) -> None:
+        if not self._installed_signals:
+            return
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self._installed_signals.clear()
+            return
+        for sig in self._installed_signals:
+            with contextlib.suppress(NotImplementedError, OSError):
+                loop.remove_signal_handler(sig)
+        self._installed_signals.clear()
+
+    async def __aenter__(self) -> Self:
+        """Start the server in a background task and wait until it is ready.
+
+        This form is for tests and for code that must run other work next to
+        the server. It never installs signal handlers. There is no time limit:
+        use `asyncio.timeout()` to set one. If the server stops before it is
+        ready, this raises the server error at once.
+        """
+        if self._task is not None and not self._task.done():
+            raise RuntimeError(f"{self._name} server is already serving")
+
+        task = self._task = asyncio.create_task(self._serve(install_signal_handlers=False))
+        ready = asyncio.create_task(self.wait_ready())
+        try:
+            await asyncio.wait((task, ready), return_when=asyncio.FIRST_COMPLETED)
+            if task.done():
+                task.result()
+                raise RuntimeError(f"{self._name} server stopped before it was ready")
+            ready.result()
+        except BaseException:
+            ready.cancel()
+            await asyncio.wait((ready,))
+            self.stop()
+            self._task = None
+            # Surface the server error, if there is one. It explains the
+            # failure better than a cancellation or a timeout does.
+            await task
+            raise
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        self.stop()
+        if self._task is not None:
+            try:
+                await self._task
+            finally:
+                self._task = None