[pull] master from aio-libs:master

CHANGES/13299.misc.rst

@@ -0,0 +1,2 @@
+Avoided formatting an unused fallback ``Date`` header value when the response
+already has one -- by :user:`marcus-campbell`.

CHANGES/13346.bugfix.rst

@@ -0,0 +1 @@
+Fixed event loop state possibly being corrupted on Python 3.12+ -- by :user:`Dreamsorcerer`.

CONTRIBUTORS.txt

@@ -262,6 +262,7 @@ Manuel Miranda
 Marat Sharafutdinov
 Marc Mueller
 Marco Paolini
+Marcus Campbell
 Marcus Stojcevich
 Mariano Anaya
 Mariusz Masztalerczuk

aiohttp/_websocket/writer.py

@@ -3,6 +3,7 @@
 import asyncio
 import random
 import sys
+from asyncio.base_events import BaseEventLoop
 from functools import partial
 from typing import Final
 
@@ -96,13 +97,20 @@ async def send_frame(
             # Create a task to shield from cancellation
             # The lock is acquired inside the shielded task so the entire
             # operation (lock + compress + send) completes atomically.
-            # Use eager_start on Python 3.12+ to avoid scheduling overhead
-            loop = asyncio.get_running_loop()
+            # Use eager_start to avoid scheduling overhead
             coro = self._send_compressed_frame_async_locked(message, opcode, compress)
-            if sys.version_info >= (3, 12):
-                send_task = asyncio.Task(coro, loop=loop, eager_start=True)
+            if sys.version_info >= (3, 14):
+                loop = asyncio.get_running_loop()
+                if isinstance(loop, BaseEventLoop):
+                    send_task = asyncio.create_task(coro, eager_start=True)
+                else:
+                    send_task = asyncio.Task(coro, loop=loop, eager_start=True)
+            elif sys.version_info >= (3, 12):
+                send_task = asyncio.Task(
+                    coro, loop=asyncio.get_running_loop(), eager_start=True
+                )
             else:
-                send_task = loop.create_task(coro)
+                send_task = asyncio.create_task(coro)
             # Keep a strong reference to prevent garbage collection
             self._background_tasks.add(send_task)
             send_task.add_done_callback(self._background_tasks.discard)

aiohttp/client_reqrep.py

@@ -7,6 +7,7 @@
 import sys
 import traceback
 import warnings
+from asyncio.base_events import BaseEventLoop
 from collections.abc import Callable, Iterable, Sequence
 from hashlib import md5, sha1, sha256
 from http.cookies import BaseCookie, SimpleCookie
@@ -909,7 +910,7 @@ def _update_headers(self, headers: CIMultiDict[str]) -> None:
         # host_port_subcomponent is None when the URL is a relative URL.
         # but we know we do not have a relative URL here.
         assert host is not None
-        self.headers[hdrs.HOST] = headers.pop(hdrs.HOST, host)
+        self.headers[hdrs.HOST] = headers.popall(hdrs.HOST, (host,))[0]
         self.headers.extend(headers)
 
     def _create_response(
@@ -983,13 +984,20 @@ async def _send(self, conn: "Connection") -> ClientResponse:
         task: asyncio.Task[None] | None
         if self._should_write(protocol):
             coro = self._write_bytes(writer, conn, self._get_content_length())
-            if sys.version_info >= (3, 12):
-                # Optimization for Python 3.12, try to write
-                # bytes immediately to avoid having to schedule
+            if sys.version_info >= (3, 14):
+                # Try to write bytes immediately to avoid having to schedule
                 # the task on the event loop.
-                task = asyncio.Task(coro, loop=self.loop, eager_start=True)
+                loop = asyncio.get_running_loop()
+                if isinstance(loop, BaseEventLoop):
+                    task = asyncio.create_task(coro, eager_start=True)
+                else:
+                    task = asyncio.Task(coro, loop=loop, eager_start=True)
+            elif sys.version_info >= (3, 12):
+                task = asyncio.Task(
+                    coro, loop=asyncio.get_running_loop(), eager_start=True
+                )
             else:
-                task = self.loop.create_task(coro)
+                task = asyncio.create_task(coro)
             if task.done():
                 task = None
             else:

aiohttp/client_ws.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import sys
+from asyncio.base_events import BaseEventLoop
 from collections.abc import Callable
 from types import TracebackType
 from typing import Any, Final, Generic, Literal, overload
@@ -185,13 +186,17 @@ def _send_heartbeat(self) -> None:
         self._pong_response_cb = loop.call_at(when, self._pong_not_received)
 
         coro = self._writer.send_frame(b"", WSMsgType.PING)
-        if sys.version_info >= (3, 12):
-            # Optimization for Python 3.12, try to send the ping
-            # immediately to avoid having to schedule
+        if sys.version_info >= (3, 14):
+            # Try to send the ping immediately to avoid having to schedule
             # the task on the event loop.
+            if isinstance(loop, BaseEventLoop):
+                ping_task = asyncio.create_task(coro, eager_start=True)
+            else:
+                ping_task = asyncio.Task(coro, loop=loop, eager_start=True)
+        elif sys.version_info >= (3, 12):
             ping_task = asyncio.Task(coro, loop=loop, eager_start=True)
         else:
-            ping_task = loop.create_task(coro)
+            ping_task = asyncio.create_task(coro)
 
         if not ping_task.done():
             self._ping_task = ping_task

aiohttp/connector.py

@@ -5,6 +5,7 @@
 import sys
 import traceback
 import warnings
+from asyncio.base_events import BaseEventLoop
 from collections import OrderedDict, defaultdict, deque
 from collections.abc import Awaitable, Callable, Iterator, Sequence
 from contextlib import suppress
@@ -1195,12 +1196,19 @@ async def _resolve_host(
         # all the waiters across all connections.
         #
         coro = self._resolve_host_with_throttle(key, host, port, futures, traces)
-        loop = asyncio.get_running_loop()
-        if sys.version_info >= (3, 12):
-            # Optimization for Python 3.12, try to send immediately
-            resolved_host_task = asyncio.Task(coro, loop=loop, eager_start=True)
+        if sys.version_info >= (3, 14):
+            # Try to send immediately to avoid having to schedule the task.
+            loop = asyncio.get_running_loop()
+            if isinstance(loop, BaseEventLoop):
+                resolved_host_task = asyncio.create_task(coro, eager_start=True)
+            else:
+                resolved_host_task = asyncio.Task(coro, loop=loop, eager_start=True)
+        elif sys.version_info >= (3, 12):
+            resolved_host_task = asyncio.Task(
+                coro, loop=asyncio.get_running_loop(), eager_start=True
+            )
         else:
-            resolved_host_task = loop.create_task(coro)
+            resolved_host_task = asyncio.create_task(coro)
 
         if not resolved_host_task.done():
             self._resolve_host_tasks.add(resolved_host_task)

aiohttp/web_protocol.py

@@ -2,6 +2,7 @@
 import asyncio.streams
 import sys
 import traceback
+from asyncio.base_events import BaseEventLoop
 from collections import deque
 from collections.abc import Awaitable, Callable, Sequence
 from contextlib import suppress
@@ -404,7 +405,12 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self._manager.connection_made(self, real_transport)
 
         loop = self._loop
-        if sys.version_info >= (3, 12):
+        if sys.version_info >= (3, 14):
+            if isinstance(loop, BaseEventLoop):
+                task = asyncio.create_task(self.start(), eager_start=True)
+            else:
+                task = asyncio.Task(self.start(), loop=loop, eager_start=True)
+        elif sys.version_info >= (3, 12):
             task = asyncio.Task(self.start(), loop=loop, eager_start=True)
         else:
             task = loop.create_task(self.start())
@@ -726,7 +732,12 @@ async def start(self) -> None:
             try:
                 # a new task is used for copy context vars (#3406)
                 coro = self._handle_request(request, start, self._request_handler)
-                if sys.version_info >= (3, 12):
+                if sys.version_info >= (3, 14):
+                    if isinstance(loop, BaseEventLoop):
+                        task = asyncio.create_task(coro, eager_start=True)
+                    else:
+                        task = asyncio.Task(coro, loop=loop, eager_start=True)
+                elif sys.version_info >= (3, 12):
                     task = asyncio.Task(coro, loop=loop, eager_start=True)
                 else:
                     task = loop.create_task(coro)

aiohttp/web_response.py

@@ -421,7 +421,8 @@ async def _prepare_headers(self) -> None:
         elif (writer.length if self._length_check else self.content_length) != 0:
             # https://www.rfc-editor.org/rfc/rfc9110#section-8.3-5
             headers.setdefault(hdrs.CONTENT_TYPE, "application/octet-stream")
-        headers.setdefault(hdrs.DATE, rfc822_formatted_time())
+        if hdrs.DATE not in headers:
+            headers[hdrs.DATE] = rfc822_formatted_time()
         headers.setdefault(hdrs.SERVER, SERVER_SOFTWARE)
 
         # connection header

aiohttp/web_ws.py

@@ -4,6 +4,7 @@
 import hashlib
 import json
 import sys
+from asyncio.base_events import BaseEventLoop
 from collections.abc import Callable, Iterable
 from typing import Any, Final, Generic, Literal, Union, overload
 
@@ -209,13 +210,17 @@ def _send_heartbeat(self) -> None:
         self._pong_response_cb = loop.call_at(when, self._pong_not_received)
 
         coro = self._writer.send_frame(b"", WSMsgType.PING)
-        if sys.version_info >= (3, 12):
-            # Optimization for Python 3.12, try to send the ping
-            # immediately to avoid having to schedule
+        if sys.version_info >= (3, 14):
+            # Try to send the ping immediately to avoid having to schedule
             # the task on the event loop.
+            if isinstance(loop, BaseEventLoop):
+                ping_task = asyncio.create_task(coro, eager_start=True)
+            else:
+                ping_task = asyncio.Task(coro, loop=loop, eager_start=True)
+        elif sys.version_info >= (3, 12):
             ping_task = asyncio.Task(coro, loop=loop, eager_start=True)
         else:
-            ping_task = loop.create_task(coro)
+            ping_task = asyncio.create_task(coro)
 
         if not ping_task.done():
             self._ping_task = ping_task

tests/test_benchmarks_web_response.py

@@ -1,10 +1,12 @@
 """codspeed benchmarks for the web responses."""
 
+import asyncio
 from typing import TYPE_CHECKING
 
 import pytest
 
 from aiohttp import web
+from aiohttp.test_utils import make_mocked_request
 
 if TYPE_CHECKING:
     from pytest_codspeed import BenchmarkFixture
@@ -38,6 +40,39 @@ def _run() -> None:
             web.Response(headers=headers)
 
 
+@pytest.mark.parametrize(
+    "date_header",
+    (None, "Sun, 01 Aug 2021 12:00:00 GMT"),
+    ids=("generated-date", "explicit-date"),
+)
+def test_prepare_web_response_headers(
+    benchmark: BenchmarkFixture,
+    event_loop: asyncio.AbstractEventLoop,
+    date_header: str | None,
+) -> None:
+    """Benchmark preparing 100 response headers with and without a Date header."""
+    response_count = 100
+    headers = {
+        "Content-Length": "0",
+        "Content-Type": "text/plain",
+        "Server": "aiohttp",
+    }
+    if date_header is not None:
+        headers["Date"] = date_header
+    request = make_mocked_request("GET", "/")
+
+    async def prepare_responses() -> None:
+        for _ in range(response_count):
+            response = web.Response(headers=headers)
+            response._req = request
+            response._payload_writer = request._payload_writer
+            await response._prepare_headers()
+
+    @benchmark
+    def _run() -> None:
+        event_loop.run_until_complete(prepare_responses())
+
+
 def test_web_response_with_bytes_body(
     benchmark: BenchmarkFixture,
 ) -> None:

tests/test_client_request.py

@@ -313,6 +313,17 @@ async def test_host_header_explicit_host_with_port(
     assert req.headers["HOST"] == "example.com:99"
 
 
+async def test_host_header_duplicate_dropped(
+    make_client_request: _RequestMaker,
+) -> None:
+    req = make_client_request(
+        "get",
+        URL("http://python.org/"),
+        headers=CIMultiDict([("host", "example.com"), ("host", "evil.example")]),
+    )
+    assert req.headers.getall("HOST") == ["example.com"]
+
+
 async def test_host_header_ipv4(make_client_request: _RequestMaker) -> None:
     req = make_client_request("get", URL("http://127.0.0.2"))
     assert req.headers["HOST"] == "127.0.0.2"