Merge pull request #2827 from litestar-org/develop

chore(release): v2.5.0

.gitignore

@@ -8,6 +8,7 @@
 .scannerwork/
 .unasyncd_cache/
 .venv/
+.venv*
 .vscode/
 __pycache__/
 build/

docs/conf.py

@@ -57,6 +57,7 @@
     "opentelemetry": ("https://opentelemetry-python.readthedocs.io/en/latest/", None),
     "advanced-alchemy": ("https://docs.advanced-alchemy.jolt.rs/latest/", None),
     "jinja2": ("https://jinja.palletsprojects.com/en/latest/", None),
+    "trio": ("https://trio.readthedocs.io/en/stable/", None),
 }
 
 napoleon_google_docstring = True
@@ -226,6 +227,9 @@
     ),
     re.compile(r"litestar\.dto.*"): re.compile(".*T|.*FieldDefinition|Empty"),
     re.compile(r"litestar\.template\.(config|TemplateConfig).*"): re.compile(".*EngineType"),
+    "litestar.concurrency.set_asyncio_executor": {"ThreadPoolExecutor"},
+    "litestar.concurrency.get_asyncio_executor": {"ThreadPoolExecutor"},
+    re.compile(r"litestar\.channels\.backends\.asyncpg.*"): {"asyncpg.connection.Connection"},
 }
 
 # Do not warn about broken links to the following:

docs/reference/channels/backends/asyncpg.rst

@@ -0,0 +1,5 @@
+asyncpg
+=======
+
+.. automodule:: litestar.channels.backends.asyncpg
+    :members:

docs/reference/channels/backends/index.rst

@@ -6,3 +6,5 @@ backends
     base
     memory
     redis
+    psycopg
+    asyncpg

docs/reference/channels/backends/psycopg.rst

@@ -0,0 +1,5 @@
+psycopg
+=======
+
+.. automodule:: litestar.channels.backends.psycopg
+    :members:

docs/reference/concurrency.rst

@@ -0,0 +1,5 @@
+cli
+===
+
+.. automodule:: litestar.concurrency
+    :members:

docs/reference/index.rst

@@ -14,6 +14,7 @@ API reference
     connection
     contrib/index
     controller
+    concurrency
     data_extractors
     datastructures
     di

docs/release-notes/changelog.rst

@@ -3,6 +3,160 @@
 2.x Changelog
 =============
 
+.. changelog:: 2.5.0
+    :date: 2024/01/06
+
+    .. change:: Fix serialization of custom types in exception responses
+        :type: bugfix
+        :issue: 2867
+        :pr: 2941
+
+        Fix a bug that would lead to a :exc:`SerializationException` when custom types
+        were present in an exception response handled by the built-in exception
+        handlers.
+
+        .. code-block:: python
+
+            class Foo:
+                pass
+
+
+            @get()
+            def handler() -> None:
+                raise ValidationException(extra={"foo": Foo("bar")})
+
+
+            app = Litestar(route_handlers=[handler], type_encoders={Foo: lambda foo: "foo"})
+
+        The cause was that, in examples like the one shown above, ``type_encoders``
+        were not resolved properly from all layers by the exception handling middleware,
+        causing the serializer to throw an exception for an unknown type.
+
+    .. change:: Fix SSE reverting to default ``event_type`` after 1st message
+        :type: bugfix
+        :pr: 2888
+        :issue: 2877
+
+        The ``event_type`` set within an SSE returned from a handler would revert back
+        to a default after the first message sent:
+
+        .. code-block:: python
+
+            @get("/stream")
+            async def stream(self) -> ServerSentEvent:
+                async def gen() -> AsyncGenerator[str, None]:
+                    c = 0
+                    while True:
+                        yield f"<div>{c}</div>\n"
+                        c += 1
+
+                return ServerSentEvent(gen(), event_type="my_event")
+
+        In this example, the event type would only be ``my_event`` for the first
+        message, and fall back to a default afterwards. The implementation has been
+        fixed and will now continue sending the set event type for all messages.
+
+    .. change:: Correctly handle single file upload validation when multiple files are specified
+        :type: bugfix
+        :pr: 2950
+        :issue: 2939
+
+        Uploading a single file when the validation target allowed multiple would cause
+        a :exc:`ValidationException`:
+
+        .. code-block:: python
+
+            class FileUpload(Struct):
+                files: list[UploadFile]
+
+
+            @post(path="/")
+            async def upload_files_object(
+                data: Annotated[FileUpload, Body(media_type=RequestEncodingType.MULTI_PART)]
+            ) -> list[str]:
+                pass
+
+
+        This could would only allow for 2 or more files to be sent, and otherwise throw
+        an exception.
+
+    .. change:: Fix trailing messages after unsubscribe in channels
+        :type: bugfix
+        :pr: 2894
+
+        Fix a bug that would allow some channels backend to receive messages from a
+        channel it just unsubscribed from, for a short period of time, due to how the
+        different brokers handle unsubscribes.
+
+        .. code-block:: python
+
+            await backend.subscribe(["foo", "bar"])  # subscribe to two channels
+            await backend.publish(
+                b"something", ["foo"]
+            )  # publish a message to a channel we're subscribed to
+
+            # start the stream after publishing. Depending on the backend
+            # the previously published message might be in the stream
+            event_generator = backend.stream_events()
+
+            # unsubscribe from the channel we previously published to
+            await backend.unsubscribe(["foo"])
+
+            # this should block, as we expect messages from channels
+            # we unsubscribed from to not appear in the stream anymore
+            print(anext(event_generator))
+
+        Backends affected by this were in-memory, Redis PubSub and asyncpg. The Redis
+        stream and psycopg backends were not affected.
+
+    .. change:: Postgres channels backends
+        :type: feature
+        :pr: 2803
+
+        Two new channel backends were added to bring Postgres support:
+
+        :class:`~litestar.channels.backends.asyncpg.AsyncPgChannelsBackend`, using the
+        `asyncpg <https://magicstack.github.io/asyncpg/current/>`_ driver and
+        :class:`~litestar.channels.backends.psycopg.PsycoPgChannelsBackend` using the
+        `psycopg3 <https://www.psycopg.org/psycopg3/docs/>`_ async driver.
+
+        .. seealso::
+            :doc:`/usage/channels`
+
+
+    .. change:: Add ``--schema`` and ``--exclude`` option to ``litestar route`` CLI command
+        :type: feature
+        :pr: 2886
+
+        Two new options were added to the ``litestar route`` CLI command:
+
+        - ``--schema``, to include the routes serving OpenAPI schema and docs
+        - ``--exclude`` to exclude routes matching a specified pattern
+
+        .. seealso::
+            :ref:`usage/cli:routes`
+
+    .. change:: Improve performance of threaded synchronous execution
+        :type: misc
+        :pr: 2937
+
+        Performance of threaded synchronous code was improved by using the async
+        library's native threading helpers instead of anyio. On asyncio,
+        :meth:`asyncio.loop.run_in_executor` is now used and on trio
+        :func:`trio.to_thread.run_sync`.
+
+        Beneficiaries of these performance improvements are:
+
+        - Synchronous route handlers making use of ``sync_to_thread=True``
+        - Synchronous dependency providers making use of ``sync_to_thread=True``
+        - Synchronous SSE generators
+        - :class:`~litestar.stores.file.FileStore`
+        - Large file uploads where the ``max_spool_size`` is exceeded and the spooled
+          temporary file has been rolled to disk
+        - :class:`~litestar.response.file.File` and
+          :class:`~litestar.response.file.ASGIFileResponse`
+
+
 .. changelog:: 2.4.5
     :date: 2023/12/23
 

docs/usage/channels.rst

@@ -413,6 +413,17 @@ implemented are:
     when history is needed
 
 
+:class:`AsyncPgChannelsBackend <.asyncpg.AsyncPgChannelsBackend>`
+    A postgres backend using the
+    `asyncpg <https://magicstack.github.io/asyncpg/current/>`_ driver
+
+
+:class:`PsycoPgChannelsBackend <.psycopg.PsycoPgChannelsBackend>`
+    A postgres backend using the `psycopg3 <https://www.psycopg.org/psycopg3/docs/>`_
+    async driver
+
+
+
 
 Integrating with websocket handlers
 -----------------------------------

docs/usage/cli.rst

@@ -185,6 +185,19 @@ The ``routes`` command displays a tree view of the routing table.
 
    litestar routes
 
+Options
+~~~~~~~
+
++-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+
+| Flag            | Description                                                                                                                                               |
++=================+===========================================================================================================================================================+
+| ``--schema``    | Include default auto generated openAPI schema routes                                                                                                      |
++-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+
+| ``--exclude``   | Exclude endpoints from query with given regex patterns. Multiple excludes allowed. e.g., ``litestar routes  --schema --exclude=routes/.* --exclude=[]``   |
++-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+
+
+
+
 
 .. image:: /images/cli/litestar_routes.png
    :alt: litestar info

litestar/channels/backends/asyncpg.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import asyncio
+from contextlib import AsyncExitStack
+from functools import partial
+from typing import AsyncGenerator, Awaitable, Callable, Iterable, overload
+
+import asyncpg
+
+from litestar.channels import ChannelsBackend
+from litestar.exceptions import ImproperlyConfiguredException
+
+
+class AsyncPgChannelsBackend(ChannelsBackend):
+    _listener_conn: asyncpg.Connection
+
+    @overload
+    def __init__(self, dsn: str) -> None:
+        ...
+
+    @overload
+    def __init__(
+        self,
+        *,
+        make_connection: Callable[[], Awaitable[asyncpg.Connection]],
+    ) -> None:
+        ...
+
+    def __init__(
+        self,
+        dsn: str | None = None,
+        *,
+        make_connection: Callable[[], Awaitable[asyncpg.Connection]] | None = None,
+    ) -> None:
+        if not (dsn or make_connection):
+            raise ImproperlyConfiguredException("Need to specify dsn or make_connection")
+
+        self._subscribed_channels: set[str] = set()
+        self._exit_stack = AsyncExitStack()
+        self._connect = make_connection or partial(asyncpg.connect, dsn=dsn)
+        self._queue: asyncio.Queue[tuple[str, bytes]] | None = None
+
+    async def on_startup(self) -> None:
+        self._queue = asyncio.Queue()
+        self._listener_conn = await self._connect()
+
+    async def on_shutdown(self) -> None:
+        await self._listener_conn.close()
+        self._queue = None
+
+    async def publish(self, data: bytes, channels: Iterable[str]) -> None:
+        if self._queue is None:
+            raise RuntimeError("Backend not yet initialized. Did you forget to call on_startup?")
+
+        dec_data = data.decode("utf-8")
+
+        conn = await self._connect()
+        try:
+            for channel in channels:
+                await conn.execute("SELECT pg_notify($1, $2);", channel, dec_data)
+        finally:
+            await conn.close()
+
+    async def subscribe(self, channels: Iterable[str]) -> None:
+        for channel in set(channels) - self._subscribed_channels:
+            await self._listener_conn.add_listener(channel, self._listener)  # type: ignore[arg-type]
+            self._subscribed_channels.add(channel)
+
+    async def unsubscribe(self, channels: Iterable[str]) -> None:
+        for channel in channels:
+            await self._listener_conn.remove_listener(channel, self._listener)  # type: ignore[arg-type]
+        self._subscribed_channels = self._subscribed_channels - set(channels)
+
+    async def stream_events(self) -> AsyncGenerator[tuple[str, bytes], None]:
+        if self._queue is None:
+            raise RuntimeError("Backend not yet initialized. Did you forget to call on_startup?")
+
+        while True:
+            channel, message = await self._queue.get()
+            self._queue.task_done()
+            # an UNLISTEN may be in transit while we're getting here, so we double-check
+            # that we are actually supposed to deliver this message
+            if channel in self._subscribed_channels:
+                yield channel, message
+
+    async def get_history(self, channel: str, limit: int | None = None) -> list[bytes]:
+        raise NotImplementedError()
+
+    def _listener(self, /, connection: asyncpg.Connection, pid: int, channel: str, payload: object) -> None:
+        if not isinstance(payload, str):
+            raise RuntimeError("Invalid data received")
+        self._queue.put_nowait((channel, payload.encode("utf-8")))  # type: ignore[union-attr]

litestar/channels/backends/memory.py

@@ -63,10 +63,19 @@ async def unsubscribe(self, channels: Iterable[str]) -> None:
 
     async def stream_events(self) -> AsyncGenerator[tuple[str, Any], None]:
         """Return a generator, iterating over events of subscribed channels as they become available"""
-        while self._queue:
-            yield await self._queue.get()
+        if self._queue is None:
+            raise RuntimeError("Backend not yet initialized. Did you forget to call on_startup?")
+
+        while True:
+            channel, message = await self._queue.get()
             self._queue.task_done()
 
+            # if a message is published to a channel and the channel is then
+            # unsubscribed before retrieving that message from the stream, it can still
+            # end up here, so we double-check if we still are interested in this message
+            if channel in self._channels:
+                yield channel, message
+
     async def get_history(self, channel: str, limit: int | None = None) -> list[bytes]:
         """Return the event history of ``channel``, at most ``limit`` entries"""
         history = list(self._history[channel])