feat: refactor Litestar framework integration and update middleware imports

Author: pythonhubdev

dev-requirements.txt

@@ -1,26 +1,26 @@
-black==24.8.0
-Django==4.2.6
-django-cors-headers==4.4.0
-django-stubs==4.2.7
-django-stubs-ext==4.2.7
-fastapi==0.115.5
-Flask==3.0.3
-flask-cors==5.0.0
+black==25.9.0
+Django==5.2.7
+django-cors-headers==4.9.0
+django-stubs==5.2.7
+django-stubs-ext==5.2.7
+fastapi==0.120.1
+Flask==3.1.2
+flask-cors==6.0.1
 nest-asyncio==1.6.0
-pdoc3==0.11.0
-pre-commit==3.5.0
-pyfakefs==5.7.4
-pylint==3.2.7
-pyright==1.1.393
-python-dotenv==1.0.1
-pytest==8.3.3
-pytest-asyncio==0.24.0
-pytest-mock==3.14.0
-pytest-rerunfailures==14.0
-pyyaml==6.0.2
+pdoc3==0.11.6
+pre-commit==4.3.0
+pyfakefs==5.10.1
+pylint==4.0.2
+pyright==1.1.407
+python-dotenv==1.2.1
+pytest==8.4.2
+pytest-asyncio==1.2.0
+pytest-mock==3.15.1
+pytest-rerunfailures==16.1
+pyyaml==6.0.3
 requests-mock==1.12.1
 respx>=0.13.0, <1.0.0
-uvicorn==0.32.0
-wasmtime==25.0.0
-litestar==2.16.0
+uvicorn==0.38.0
+wasmtime==38.0.0
+litestar==2.18.0
 -e .

pyproject.toml

@@ -13,6 +13,9 @@ indent-style = "space" # Default
 typeCheckingMode = "strict"
 reportImportCycles = false
 include = ["supertokens_python/", "tests/", "examples/"]
+exclude = [".venv"]
+venvPath = "."
+venv = ".venv"
 
 [tool.pytest.ini_options]
 addopts = " -v -p no:warnings"

supertokens_python/framework/litestar/framework.py

@@ -11,13 +11,11 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
+from litestar import Request
 from supertokens_python.framework.types import Framework
 
-if TYPE_CHECKING:
-    from litestar import Request
-
 
 class LitestarFramework(Framework):
     def wrap_request(self, unwrapped: Request[Any, Any, Any]):

supertokens_python/framework/litestar/middleware.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Optional, cast
 
 from litestar import Request, Response
 from litestar.datastructures import MutableScopeHeaders
@@ -32,7 +32,10 @@ async def handle(
 
         try:
             response = LitestarResponse(Response[Any](content=None))
-            result = await st.middleware(custom_request, response, user_context)
+            result = cast(
+                Optional[LitestarResponse],
+                await st.middleware(custom_request, response, user_context),
+            )
 
             if result is not None:
                 # SuperTokens handled the request
@@ -50,8 +53,8 @@ async def handle(
                 async def modified_send(message: Message):
                     if message["type"] == "http.response.start":
                         mutable_headers = MutableScopeHeaders(message)
-                        for cookie in result.response.cookies:
-                            cookie_value = cookie.to_header().split(": ", 1)[1]
+                        for cookie in result.response.cookies:  # type: ignore
+                            cookie_value = cookie.to_header().split(": ", 1)[1]  # type: ignore
                             mutable_headers.add("set-cookie", cookie_value)
                     await send(message)
 
@@ -64,16 +67,16 @@ async def send_wrapper(message: Message):
                             request.state.supertokens, SessionContainer
                         ):
                             temp_response = Response[Any](content=None)
-                            temp_response.headers = MutableScopeHeaders(message)
+                            temp_response.headers = MutableScopeHeaders(message)  # type: ignore
                             litestar_response = LitestarResponse(temp_response)
                             manage_session_post_response(
                                 request.state.supertokens,
                                 litestar_response,
                                 user_context,
                             )
                             mutable_headers = MutableScopeHeaders(message)
-                            for cookie in litestar_response.response.cookies:
-                                cookie_value = cookie.to_header().split(": ", 1)[1]
+                            for cookie in litestar_response.response.cookies:  # type: ignore
+                                cookie_value = cookie.to_header().split(": ", 1)[1]  # type: ignore
                                 mutable_headers.add("set-cookie", cookie_value)
                     await send(message)
 
@@ -82,8 +85,11 @@ async def send_wrapper(message: Message):
         except SuperTokensError as e:
             # Handle SuperTokens errors
             response = LitestarResponse(Response[Any](content=None))
-            result = await st.handle_supertokens_error(
-                custom_request, e, response, user_context
+            result = cast(
+                Optional[LitestarResponse],
+                await st.handle_supertokens_error(
+                    custom_request, e, response, user_context
+                ),
             )
             if isinstance(result, LitestarResponse):
                 asgi_response = await result.response.to_asgi_response(
@@ -93,8 +99,8 @@ async def send_wrapper(message: Message):
                 async def modified_send(message: Message):
                     if message["type"] == "http.response.start":
                         mutable_headers = MutableScopeHeaders(message)
-                        for cookie in result.response.cookies:
-                            cookie_value = cookie.to_header().split(": ", 1)[1]
+                        for cookie in result.response.cookies:  # type: ignore
+                            cookie_value = cookie.to_header().split(": ", 1)[1]  # type: ignore
                             mutable_headers.add("set-cookie", cookie_value)
                     await send(message)
 

supertokens_python/framework/litestar/request.py

@@ -12,7 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 import json
-from typing import Any, Dict, Union
+from typing import Any, Union
 from urllib.parse import parse_qsl
 
 from litestar import Request
@@ -33,10 +33,10 @@ def get_query_param(
     ) -> Union[str, None]:
         return self.request.query_params.get(key, default)
 
-    def get_query_params(self) -> Dict[str, Any]:
+    def get_query_params(self) -> dict[str, Any]:
         return dict(self.request.query_params.items())  # type: ignore
 
-    async def json(self) -> dict:
+    async def json(self) -> dict[str, Any]:
         """
         Read the entire ASGI stream and JSON-decode it,
         sidestepping Litestar’s internal max-body-size logic.

tests/litestar/test_litestar.py

@@ -12,7 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 import json
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, Union
 from unittest import skip
 
 from litestar import Litestar, MediaType, Request, Response, get, post
@@ -61,7 +61,7 @@
 
 def override_dashboard_functions(original_implementation: RecipeInterface):
     async def should_allow_access(
-        request: BaseRequest, __: DashboardConfig, ___: Dict[str, Any]
+        request: BaseRequest, config: DashboardConfig, user_context: Dict[str, Any]
     ):
         auth_header = request.get_header("authorization")
         return auth_header == "Bearer testapikey"
@@ -121,7 +121,7 @@ async def handle_get(
         },
     )
     async def handle_get_optional(
-        session: SessionContainer,
+        session: Optional[SessionContainer],
     ) -> Dict[str, Any]:
         if session is None:
             return {"s": "empty session"}
@@ -739,9 +739,9 @@ async def test_search_with_email_t(driver_config_client: TestClient[Litestar]):
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(emailpassword=True)
     query = {"limit": "10", "email": "t"}
     res = driver_config_client.get(
@@ -786,9 +786,9 @@ async def test_search_with_email_multiple_email_entry(
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(emailpassword=True)
     query = {"limit": "10", "email": "iresh;john"}
     res = driver_config_client.get(
@@ -831,9 +831,9 @@ async def test_search_with_email_iresh(driver_config_client: TestClient[Litestar
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(emailpassword=True)
     query = {"limit": "10", "email": "iresh"}
     res = driver_config_client.get(
@@ -879,9 +879,9 @@ async def test_search_with_phone_plus_one(driver_config_client: TestClient[Lites
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(passwordless=True)
     query = {"limit": "10", "phone": "+1"}
     res = driver_config_client.get(
@@ -929,9 +929,9 @@ async def test_search_with_phone_one_bracket(
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(passwordless=True)
     query = {"limit": "10", "phone": "1("}
     res = driver_config_client.get(
@@ -1016,9 +1016,9 @@ async def test_search_with_provider_google(driver_config_client: TestClient[Lite
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(thirdparty=True)
     query = {"limit": "10", "provider": "google"}
     res = driver_config_client.get(
@@ -1109,9 +1109,9 @@ async def test_search_with_provider_google_and_phone_1(
     querier = Querier.get_instance(DashboardRecipe.recipe_id)
     cdi_version = await querier.get_api_version()
     if not cdi_version:
-        skip()
+        skip("CDI version not available")
     if not is_version_gte(cdi_version, "2.20"):
-        skip()
+        skip("CDI version too old")
     await create_users(thirdparty=True, passwordless=True)
     query = {"limit": "10", "provider": "google", "phone": "1"}
     res = driver_config_client.get(