Add Request contextvars

sanic/http.py

@@ -265,6 +265,7 @@ async def http1_request_header(self):  # no cov
             transport=self.protocol.transport,
             app=self.protocol.app,
         )
+        self.protocol.request_class._current.set(request)
         await self.dispatch(
             "http.lifecycle.request",
             inline=True,

sanic/request.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from contextvars import ContextVar
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -35,7 +36,7 @@
 
 from sanic.compat import CancelledErrors, Header
 from sanic.constants import DEFAULT_HTTP_CONTENT_TYPE
-from sanic.exceptions import BadRequest, BadURL, ServerError
+from sanic.exceptions import BadRequest, BadURL, SanicException, ServerError
 from sanic.headers import (
     AcceptContainer,
     Options,
@@ -82,6 +83,8 @@ class Request:
     Properties of an HTTP request such as URL, headers, etc.
     """
 
+    _current: ContextVar[Request] = ContextVar("request")
+
     __slots__ = (
         "__weakref__",
         "_cookies",
@@ -174,6 +177,13 @@ def __repr__(self):
         class_name = self.__class__.__name__
         return f"<{class_name}: {self.method} {self.path}>"
 
+    @classmethod
+    def get_current(cls) -> Request:
+        request = cls._current.get(None)
+        if not request:
+            raise SanicException("No current request")
+        return request
+
     @classmethod
     def generate_id(*_):
         return uuid.uuid4()

tests/test_request.py

@@ -4,7 +4,7 @@
 import pytest
 
 from sanic import Sanic, response
-from sanic.exceptions import BadURL
+from sanic.exceptions import BadURL, SanicException
 from sanic.request import Request, uuid
 from sanic.server import HttpProtocol
 
@@ -217,3 +217,17 @@ async def get(request):
     assert request.scope is not None
     assert request.scope["method"].lower() == "get"
     assert request.scope["path"].lower() == "/"
+
+
+def test_cannot_get_request_outside_of_cycle():
+    with pytest.raises(SanicException, match="No current request"):
+        Request.get_current()
+
+
+def test_get_current_request(app):
+    @app.get("/")
+    async def get(request):
+        return response.json({"same": request is Request.get_current()})
+
+    _, resp = app.test_client.get("/")
+    assert resp.json["same"]