python 3.14 deprecates eager event loop binding

.gitignore

@@ -161,3 +161,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+# uv 
+uv.lock

.python-version

@@ -1 +1 @@
-3.11.1
+3.14

pyproject.toml

@@ -10,7 +10,8 @@ dependencies = [
     "aiohttp-jinja2>=1.6",
     "jinja2>=3.1.4",
     "rich",
-    "textual>=0.66.0",
+    "textual>=0.66.0; python_version < '3.14'",
+    "textual>=6.3.0; python_version >= '3.14'",
 ]
 readme = "README.md"
 requires-python = ">= 3.9"

src/textual_serve/_compat.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+__all__ = ["cached_property"]
+
+import sys
+from typing import Any, Callable, Generic, Self, TypeVar, overload
+
+if sys.version_info >= (3, 12):
+    from functools import cached_property
+else:
+    # based on the code from Python 3.14:
+    # https://github.com/python/cpython/blob/
+    # 5507eff19c757a908a2ff29dfe423e35595fda00/Lib/functools.py#L1089-L1138
+    # Copyright (C) 2006 Python Software Foundation.
+    # vendored under the PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2 because
+    # prior to Python 3.12 cached_property used a threading.Lock, which makes
+    # it very slow.
+    _T_co = TypeVar("_T_co", covariant=True)
+    _NOT_FOUND = object()
+
+    class cached_property(Generic[_T_co]):
+        def __init__(self, func: Callable[[Any, _T_co]]) -> None:
+            self.func = func
+            self.attrname = None
+            self.__doc__ = func.__doc__
+            self.__module__ = func.__module__
+
+        def __set_name__(self, owner: type[Any], name: str) -> None:
+            if self.attrname is None:
+                self.attrname = name
+            elif name != self.attrname:
+                raise TypeError(
+                    "Cannot assign the same cached_property to two different names "
+                    f"({self.attrname!r} and {name!r})."
+                )
+
+        @overload
+        def __get__(self, instance: None, owner: type[Any] | None = None) -> Self: ...
+
+        @overload
+        def __get__(
+            self, instance: object, owner: type[Any] | None = None
+        ) -> _T_co: ...
+
+        def __get__(
+            self, instance: object, owner: type[Any] | None = None
+        ) -> _T_co | Self:
+            if instance is None:
+                return self
+            if self.attrname is None:
+                raise TypeError(
+                    "Cannot use cached_property instance without calling __set_name__ on it."
+                )
+            try:
+                cache = instance.__dict__
+            except (
+                AttributeError
+            ):  # not all objects have __dict__ (e.g. class defines slots)
+                msg = (
+                    f"No '__dict__' attribute on {type(instance).__name__!r} "
+                    f"instance to cache {self.attrname!r} property."
+                )
+                raise TypeError(msg) from None
+            val = cache.get(self.attrname, _NOT_FOUND)
+            if val is _NOT_FOUND:
+                val = self.func(instance)
+                try:
+                    cache[self.attrname] = val
+                except TypeError:
+                    msg = (
+                        f"The '__dict__' attribute on {type(instance).__name__!r} instance "
+                        f"does not support item assignment for caching {self.attrname!r} property."
+                    )
+                    raise TypeError(msg) from None
+            return val

src/textual_serve/app_service.py

@@ -1,22 +1,26 @@
 from __future__ import annotations
-from pathlib import Path
 
 import asyncio
 import io
 import json
+import logging
 import os
-from typing import Awaitable, Callable, Literal
+import sys
+import uuid
 from asyncio.subprocess import Process
-import logging
-
 from importlib.metadata import version
-import uuid
+from pathlib import Path
+from typing import Awaitable, Callable, Literal
 
-from textual_serve.download_manager import DownloadManager
 from textual_serve._binary_encode import load as binary_load
+from textual_serve._compat import cached_property
+from textual_serve.download_manager import DownloadManager
 
 log = logging.getLogger("textual-serve")
 
+# `asyncio.get_event_loop()` is deprecated since Python 3.10:
+_ASYNCIO_GET_EVENT_LOOP_IS_DEPRECATED = sys.version_info >= (3, 10, 0)
+
 
 class AppService:
     """Creates and manages a single Textual app subprocess.
@@ -51,9 +55,13 @@ def __init__(
         self._process: Process | None = None
         self._task: asyncio.Task[None] | None = None
         self._stdin: asyncio.StreamWriter | None = None
-        self._exit_event = asyncio.Event()
+
         self._download_manager = download_manager
 
+    @cached_property
+    def _exit_event(self) -> asyncio.Event:
+        return asyncio.Event()
+
     @property
     def stdin(self) -> asyncio.StreamWriter:
         """The processes standard input stream."""

src/textual_serve/server.py

@@ -1,26 +1,21 @@
 from __future__ import annotations
 
 import asyncio
-
 import logging
 import os
-from pathlib import Path
 import signal
 import sys
-
+from importlib.metadata import version
+from pathlib import Path
 from typing import Any
 
 import aiohttp_jinja2
-from aiohttp import web
-from aiohttp import WSMsgType
-from aiohttp.web_runner import GracefulExit
 import jinja2
-
-from importlib.metadata import version
-
+from aiohttp import WSMsgType, web
+from aiohttp.web_runner import GracefulExit
 from rich.console import Console
-from rich.logging import RichHandler
 from rich.highlighter import RegexHighlighter
+from rich.logging import RichHandler
 
 from textual_serve.download_manager import DownloadManager
 
@@ -204,6 +199,34 @@ async def on_startup(self, app: web.Application) -> None:
         self.console.print(f"Serving {self.command!r} on {self.public_url}")
         self.console.print("\n[cyan]Press Ctrl+C to quit")
 
+    async def _make_server(self) -> Any:
+        """Make the aiohttp web server.
+
+        Returns:
+            New aiohttp web server.
+        """
+        app = await self._make_app()
+        runner = web.AppRunner(app, handle_signals=False)
+        await runner.setup()
+        try:
+            site = web.TCPSite(runner, self.host, self.port)
+            await site.start()
+            try:
+                loop = asyncio.get_running_loop()
+                loop.add_signal_handler(signal.SIGINT, self.request_exit)
+                loop.add_signal_handler(signal.SIGTERM, self.request_exit)
+            except NotImplementedError:
+                pass
+            # Keep the server running indefinitely
+            try:
+                await asyncio.Event().wait()
+            except GracefulExit:
+                # Gracefully handle exit signal
+                pass
+        finally:
+            # Cleanup
+            await runner.cleanup()
+
     def serve(self, debug: bool = False) -> None:
         """Serve the Textual application.
 
@@ -213,27 +236,7 @@ def serve(self, debug: bool = False) -> None:
         self.debug = debug
         self.initialize_logging()
 
-        try:
-            loop = asyncio.get_event_loop()
-        except Exception:
-            loop = asyncio.new_event_loop()
-        try:
-            loop.add_signal_handler(signal.SIGINT, self.request_exit)
-            loop.add_signal_handler(signal.SIGTERM, self.request_exit)
-        except NotImplementedError:
-            pass
-
-        if self.debug:
-            log.info("Running in debug mode. You may use textual dev tools.")
-
-        web.run_app(
-            self._make_app(),
-            host=self.host,
-            port=self.port,
-            handle_signals=False,
-            loop=loop,
-            print=lambda *args: None,
-        )
+        asyncio.run(self._make_server())
 
     @aiohttp_jinja2.template("app_index.html")
     async def handle_index(self, request: web.Request) -> dict[str, Any]: