Add loop_factory parameter for persistent event loops

docs/adapter.md

@@ -6,9 +6,11 @@ The heart of Mangum is the adapter class. It is a configurable wrapper that allo
 handler = Mangum(
     app,
     lifespan="auto",
-    api_gateway_base_path=None,
+    api_gateway_base_path="/",
     custom_handlers=None,
     text_mime_types=None,
+    exclude_headers=None,
+    loop_factory=None,
 )
 ```
 
@@ -21,7 +23,7 @@ All arguments are optional.
 
 ## Creating an AWS Lambda handler
 
-The adapter can be used to wrap any application without referencing the underlying methods. It defines a `__call__` method that allows the class instance to be used as an AWS Lambda event handler function. 
+The adapter can be used to wrap any application without referencing the underlying methods. It defines a `__call__` method that allows the class instance to be used as an AWS Lambda event handler function.
 
 ```python
 from mangum import Mangum
@@ -82,3 +84,31 @@ def hello(request: Request):
 
 handler = Mangum(app)
 ```
+
+## Persistent event loop
+
+For Lambda warm invocations where you want to reuse async resources across requests:
+
+```python
+import asyncio
+from mangum import Mangum
+
+handler = Mangum(app, lifespan="off", loop_factory=asyncio.new_event_loop)
+```
+
+When `loop_factory` is provided, Mangum creates the loop once at initialization and reuses it for all invocations. The loop is not closed after each request, allowing you to maintain persistent connections and background tasks across warm invocations.
+
+**Note:** Mangum does not close the factory-created loop. In Lambda, the loop persists until the execution environment is recycled, at which point cleanup happens automatically. If you need explicit cleanup (e.g., in tests), you're responsible for closing the loop.
+
+You can also use `uvloop` for better performance:
+
+```python
+import uvloop
+from mangum import Mangum
+
+handler = Mangum(app, lifespan="off", loop_factory=uvloop.new_event_loop)
+```
+
+Use with `lifespan="off"` if you're managing lifespan startup/shutdown externally.
+
+Note: Don't call the handler from within an already-running event loop.

mangum/adapter.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 from itertools import chain
 from typing import Any
@@ -8,7 +9,7 @@
 from mangum.exceptions import ConfigurationError
 from mangum.handlers import ALB, APIGateway, HTTPGateway, LambdaAtEdge
 from mangum.protocols import HTTPCycle, LifespanCycle
-from mangum.types import ASGI, LambdaConfig, LambdaContext, LambdaEvent, LambdaHandler, LifespanMode
+from mangum.types import ASGI, LambdaConfig, LambdaContext, LambdaEvent, LambdaHandler, LifespanMode, LoopFactory
 
 logger = logging.getLogger("mangum")
 
@@ -33,12 +34,17 @@ def __init__(
         custom_handlers: list[type[LambdaHandler]] | None = None,
         text_mime_types: list[str] | None = None,
         exclude_headers: list[str] | None = None,
+        loop_factory: LoopFactory | None = None,
     ) -> None:
         if lifespan not in ("auto", "on", "off"):
             raise ConfigurationError("Invalid argument supplied for `lifespan`. Choices are: auto|on|off")
 
         self.app = app
         self.lifespan = lifespan
+        self._factory_loop: asyncio.AbstractEventLoop | None = None
+        if loop_factory:
+            self._factory_loop = loop_factory()
+            asyncio.set_event_loop(self._factory_loop)
         self.custom_handlers = custom_handlers or []
         exclude_headers = exclude_headers or []
         self.config = LambdaConfig(
@@ -75,4 +81,7 @@ async def handle_request() -> dict[str, Any]:
                 http_response = await http_cycle(self.app)
                 return handler(http_response)
 
+        if self._factory_loop is not None:
+            return self._factory_loop.run_until_complete(handle_request())
+
         return asyncio_run(handle_request())

mangum/types.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Awaitable, MutableMapping, Sequence
 from typing import Any, Callable, Union
 
 from typing_extensions import Literal, Protocol, TypeAlias, TypedDict
 
+LoopFactory: TypeAlias = Callable[[], asyncio.AbstractEventLoop]
+
 LambdaEvent = dict[str, Any]
 QueryParams: TypeAlias = MutableMapping[str, Union[str, Sequence[str]]]
 

tests/test_adapter.py

@@ -1,3 +1,6 @@
+import asyncio
+from unittest.mock import patch
+
 import pytest
 
 from mangum import Mangum
@@ -31,3 +34,93 @@ def test_invalid_options(arguments, message):
         Mangum(app, **arguments)
 
     assert str(exc.value) == message
+
+
+@pytest.mark.parametrize("mock_aws_api_gateway_event", [["GET", None, None]], indirect=True)
+def test_loop_factory_reuses_loop(mock_aws_api_gateway_event):
+    """Loop is not closed between invocations when loop_factory is provided."""
+
+    async def app(scope, receive, send):
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"OK"})
+
+    handler = Mangum(app, lifespan="off", loop_factory=asyncio.new_event_loop)
+
+    try:
+        handler(mock_aws_api_gateway_event, {})
+        handler(mock_aws_api_gateway_event, {})
+        assert not handler._factory_loop.is_closed()
+    finally:
+        handler._factory_loop.close()
+        asyncio.set_event_loop(None)
+
+
+def test_loop_factory_sets_current_loop():
+    """loop_factory sets the loop as current during initialization."""
+
+    async def app(scope, receive, send):
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"OK"})
+
+    with patch("mangum.adapter.asyncio.set_event_loop") as mock_set:
+        handler = Mangum(app, lifespan="off", loop_factory=asyncio.new_event_loop)
+        mock_set.assert_called_once_with(handler._factory_loop)
+
+    handler._factory_loop.close()
+    asyncio.set_event_loop(None)
+
+
+@pytest.mark.parametrize("mock_aws_api_gateway_event", [["GET", None, None]], indirect=True)
+def test_loop_factory_bypasses_asyncio_run(mock_aws_api_gateway_event):
+    """asyncio_run is not called when loop_factory is provided."""
+
+    async def app(scope, receive, send):
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"OK"})
+
+    handler = Mangum(app, lifespan="off", loop_factory=asyncio.new_event_loop)
+
+    try:
+        with patch("mangum.adapter.asyncio_run") as mock:
+            handler(mock_aws_api_gateway_event, {})
+            mock.assert_not_called()
+    finally:
+        handler._factory_loop.close()
+        asyncio.set_event_loop(None)
+
+
+@pytest.mark.parametrize("mock_aws_api_gateway_event", [["GET", None, None]], indirect=True)
+def test_loop_factory_with_lifespan_auto(mock_aws_api_gateway_event):
+    """loop_factory works correctly with lifespan='auto'."""
+    startup_complete = False
+    shutdown_complete = False
+
+    async def app(scope, receive, send):
+        nonlocal startup_complete, shutdown_complete
+
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    await send({"type": "lifespan.startup.complete"})
+                    startup_complete = True
+                elif message["type"] == "lifespan.shutdown":
+                    await send({"type": "lifespan.shutdown.complete"})
+                    shutdown_complete = True
+                    return
+
+        if scope["type"] == "http":
+            await send({"type": "http.response.start", "status": 200, "headers": []})
+            await send({"type": "http.response.body", "body": b"OK"})
+
+    handler = Mangum(app, lifespan="auto", loop_factory=asyncio.new_event_loop)
+
+    try:
+        response = handler(mock_aws_api_gateway_event, {})
+        assert startup_complete
+        assert shutdown_complete
+        assert response["statusCode"] == 200
+        assert not handler._factory_loop.is_closed()
+    finally:
+        handler._factory_loop.close()
+        asyncio.set_event_loop(None)