test on 3.14

.github/workflows/test-suite.yml

@@ -15,13 +15,14 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
         os: [windows-latest, ubuntu-latest, macos-latest]
     steps:
       - uses: "actions/checkout@v4"
       - uses: "actions/setup-python@v5"
         with:
           python-version: "${{ matrix.python-version }}"
+          allow-prereleases: true
       - name: "Install dependencies"
         run: "scripts/install"
         shell: bash

pyproject.toml

@@ -25,6 +25,7 @@ classifiers = [
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
     "Programming Language :: Python :: Implementation :: CPython",
     "Programming Language :: Python :: Implementation :: PyPy",
     "Topic :: Internet :: WWW/HTTP",
@@ -97,7 +98,7 @@ filterwarnings = [
 [tool.coverage.run]
 source_pkgs = ["uvicorn", "tests"]
 plugins = ["coverage_conditional_plugin"]
-omit = ["uvicorn/workers.py", "uvicorn/__main__.py"]
+omit = ["uvicorn/workers.py", "uvicorn/__main__.py", "uvicorn/_compat.py"]
 
 [tool.coverage.report]
 precision = 2
@@ -120,6 +121,7 @@ exclude_lines = [
     "tests/supervisors/test_multiprocess.py",
 ]
 "sys_platform != 'win32'" = ["uvicorn/loops/asyncio.py"]
+"sys_version_info >= (3, 14)" = ["uvicorn/loops/uvloop.py"]
 
 [tool.coverage.coverage_conditional_plugin.rules]
 py-win32 = "sys_platform == 'win32'"
@@ -132,3 +134,5 @@ py-gte-310 = "sys_version_info >= (3, 10)"
 py-lt-310 = "sys_version_info < (3, 10)"
 py-gte-311 = "sys_version_info >= (3, 11)"
 py-lt-311 = "sys_version_info < (3, 11)"
+py-gte-314 = "sys_version_info >= (3, 14)"
+py-lt-314 = "sys_version_info < (3, 14)"

scripts/install

@@ -16,4 +16,4 @@ else
 fi
 
 ${PIP} install -U pip
-${PIP} install -r "$REQUIREMENTS"
+PYO3_USE_ABI3_FORWARD_COMPATIBILITY=1 ${PIP} install -r "$REQUIREMENTS"

tests/custom_loop_utils.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+import asyncio
+
+
+class CustomLoop(asyncio.SelectorEventLoop):
+    pass

tests/test_auto_detection.py

@@ -1,19 +1,26 @@
 import asyncio
+import contextlib
 import importlib
+import sys
 
 import pytest
 
 from uvicorn.config import Config
-from uvicorn.loops.auto import auto_loop_setup
+from uvicorn.loops.auto import auto_loop_factory
 from uvicorn.main import ServerState
 from uvicorn.protocols.http.auto import AutoHTTPProtocol
 from uvicorn.protocols.websockets.auto import AutoWebSocketsProtocol
 
 try:
     importlib.import_module("uvloop")
-    expected_loop = "uvloop"  # pragma: py-win32
 except ImportError:  # pragma: py-not-win32
     expected_loop = "asyncio"
+except AttributeError:  # pragma: py-lt-314  # pragma: py-win32
+    if sys.version_info < (3, 14):  # pragma: no cover
+        raise
+    expected_loop = "asyncio"
+else:  # pragma: py-win32  # pragma: py-gte-314
+    expected_loop = "uvloop"
 
 try:
     importlib.import_module("httptools")
@@ -33,10 +40,10 @@ async def app(scope, receive, send):
 
 
 def test_loop_auto():
-    auto_loop_setup()
-    policy = asyncio.get_event_loop_policy()
-    assert isinstance(policy, asyncio.events.BaseDefaultEventLoopPolicy)
-    assert type(policy).__module__.startswith(expected_loop)
+    loop_factory = auto_loop_factory(use_subprocess=True)
+    with contextlib.closing(loop_factory()) as loop:
+        assert isinstance(loop, asyncio.AbstractEventLoop)
+        assert type(loop).__module__.startswith(expected_loop)
 
 
 @pytest.mark.anyio

tests/test_compat.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from asyncio import AbstractEventLoop
+
+import pytest
+
+from tests.custom_loop_utils import CustomLoop
+from tests.utils import get_asyncio_default_loop_per_os
+from uvicorn._compat import asyncio_run
+
+
+async def assert_event_loop(expected_loop_class: type[AbstractEventLoop]):
+    assert isinstance(asyncio.get_event_loop(), expected_loop_class)
+
+
+def test_asyncio_run__default_loop_factory() -> None:
+    asyncio_run(assert_event_loop(get_asyncio_default_loop_per_os()), loop_factory=None)
+
+
+def test_asyncio_run__custom_loop_factory() -> None:
+    asyncio_run(assert_event_loop(CustomLoop), loop_factory=CustomLoop)
+
+
+def test_asyncio_run__passing_a_non_awaitable_callback_should_throw_error() -> None:
+    with pytest.raises(TypeError if sys.version_info >= (3, 14) else ValueError):
+        asyncio_run(
+            lambda: None,  # type: ignore
+            loop_factory=CustomLoop,
+        )

tests/test_config.py

@@ -8,6 +8,7 @@
 import socket
 import sys
 import typing
+from contextlib import closing
 from pathlib import Path
 from typing import Any, Literal
 from unittest.mock import MagicMock
@@ -16,7 +17,8 @@
 import yaml
 from pytest_mock import MockerFixture
 
-from tests.utils import as_cwd
+from tests.custom_loop_utils import CustomLoop
+from tests.utils import as_cwd, get_asyncio_default_loop_per_os
 from uvicorn._types import (
     ASGIApplication,
     ASGIReceiveCallable,
@@ -25,7 +27,7 @@
     Scope,
     StartResponse,
 )
-from uvicorn.config import Config
+from uvicorn.config import Config, LoopFactoryType
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 from uvicorn.middleware.wsgi import WSGIMiddleware
 from uvicorn.protocols.http.h11_impl import H11Protocol
@@ -545,3 +547,48 @@ def test_warn_when_using_reload_and_workers(caplog: pytest.LogCaptureFixture) ->
     Config(app=asgi_app, reload=True, workers=2)
     assert len(caplog.records) == 1
     assert '"workers" flag is ignored when reloading is enabled.' in caplog.records[0].message
+
+
+@pytest.mark.parametrize(
+    ("loop_type", "expected_loop_factory"),
+    [
+        ("none", None),
+        ("asyncio", get_asyncio_default_loop_per_os()),
+    ],
+)
+def test_get_loop_factory(loop_type: LoopFactoryType, expected_loop_factory: Any):
+    config = Config(app=asgi_app, loop=loop_type)
+    loop_factory = config.get_loop_factory()
+    if loop_factory is None:
+        assert expected_loop_factory is loop_factory
+    else:
+        loop = loop_factory()
+        with closing(loop):
+            assert loop is not None
+            assert isinstance(loop, expected_loop_factory)
+
+
+def test_custom_loop__importable_custom_loop_setup_function() -> None:
+    config = Config(app=asgi_app, loop="tests.custom_loop_utils:CustomLoop")
+    config.load()
+    loop_factory = config.get_loop_factory()
+    assert loop_factory, "Loop factory should be set"
+    event_loop = loop_factory()
+    with closing(event_loop):
+        assert event_loop is not None
+        assert isinstance(event_loop, CustomLoop)
+
+
+@pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
+def test_custom_loop__not_importable_custom_loop_setup_function(caplog: pytest.LogCaptureFixture) -> None:
+    config = Config(app=asgi_app, loop="tests.test_config:non_existing_setup_function")
+    config.load()
+    with pytest.raises(SystemExit):
+        config.get_loop_factory()
+    error_messages = [
+        record.message for record in caplog.records if record.name == "uvicorn.error" and record.levelname == "ERROR"
+    ]
+    assert (
+        'Error loading custom loop setup function. Attribute "non_existing_setup_function" not found in module "tests.test_config".'  # noqa: E501
+        == error_messages.pop(0)
+    )

tests/utils.py

@@ -3,6 +3,7 @@
 import asyncio
 import os
 import signal
+import sys
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager, contextmanager
 from pathlib import Path
@@ -44,3 +45,11 @@ def as_cwd(path: Path):
         yield
     finally:
         os.chdir(prev_cwd)
+
+
+def get_asyncio_default_loop_per_os() -> type[asyncio.AbstractEventLoop]:
+    """Get the default asyncio loop per OS."""
+    if sys.platform == "win32":
+        return asyncio.ProactorEventLoop  # type: ignore  # pragma: nocover
+    else:
+        return asyncio.SelectorEventLoop  # pragma: nocover

uvicorn/_compat.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from collections.abc import Callable, Coroutine
+from typing import Any, TypeVar
+
+__all__ = ["asyncio_run", "iscoroutinefunction"]
+
+if sys.version_info >= (3, 14):
+    from inspect import iscoroutinefunction
+else:
+    from asyncio import iscoroutinefunction
+
+_T = TypeVar("_T")
+
+if sys.version_info >= (3, 12):
+    asyncio_run = asyncio.run
+elif sys.version_info >= (3, 11):
+
+    def asyncio_run(
+        main: Coroutine[Any, Any, _T],
+        *,
+        debug: bool = False,
+        loop_factory: Callable[[], asyncio.AbstractEventLoop] | None = None,
+    ) -> _T:
+        # asyncio.run from Python 3.12
+        # https://docs.python.org/3/license.html#psf-license
+        with asyncio.Runner(debug=debug, loop_factory=loop_factory) as runner:
+            return runner.run(main)
+
+else:
+    # modified version of asyncio.run from Python 3.10 to add loop_factory kwarg
+    # https://docs.python.org/3/license.html#psf-license
+    def asyncio_run(
+        main: Coroutine[Any, Any, _T],
+        *,
+        debug: bool = False,
+        loop_factory: Callable[[], asyncio.AbstractEventLoop] | None = None,
+    ) -> _T:
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            pass
+        else:
+            raise RuntimeError("asyncio.run() cannot be called from a running event loop")
+
+        if not asyncio.iscoroutine(main):
+            raise ValueError(f"a coroutine was expected, got {main!r}")
+
+        if loop_factory is None:
+            loop = asyncio.new_event_loop()
+        else:
+            loop = loop_factory()
+        try:
+            if loop_factory is None:
+                asyncio.set_event_loop(loop)
+            if debug is not None:
+                loop.set_debug(debug)
+            return loop.run_until_complete(main)
+        finally:
+            try:
+                _cancel_all_tasks(loop)
+                loop.run_until_complete(loop.shutdown_asyncgens())
+                loop.run_until_complete(loop.shutdown_default_executor())
+            finally:
+                if loop_factory is None:
+                    asyncio.set_event_loop(None)
+                loop.close()
+
+    def _cancel_all_tasks(loop: asyncio.AbstractEventLoop) -> None:
+        to_cancel = asyncio.all_tasks(loop)
+        if not to_cancel:
+            return
+
+        for task in to_cancel:
+            task.cancel()
+
+        loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True))
+
+        for task in to_cancel:
+            if task.cancelled():
+                continue
+            if task.exception() is not None:
+                loop.call_exception_handler(
+                    {
+                        "message": "unhandled exception during asyncio.run() shutdown",
+                        "exception": task.exception(),
+                        "task": task,
+                    }
+                )