feat: support app=None for tortoise.contrib.fastapi.RegisterTortoise

examples/fastapi/_tests.py

@@ -1,11 +1,14 @@
 # mypy: no-disallow-untyped-decorators
 # pylint: disable=E0611,E0401
+import multiprocessing
 import os
+from concurrent.futures import ProcessPoolExecutor
 from contextlib import asynccontextmanager
 from datetime import datetime
 from pathlib import Path
 from typing import AsyncGenerator, Tuple
 
+import anyio
 import pytest
 import pytz
 from asgi_lifespan import LifespanManager
@@ -16,6 +19,7 @@
 
 os.environ["DB_URL"] = MEMORY_SQLITE
 try:
+    from config import register_orm
     from main import app
     from main_custom_timezone import app as app_east
     from models import Users
@@ -117,3 +121,18 @@ async def test_user_list(self, client_east: AsyncClient) -> None:  # nosec
         created_at = user_obj.created_at
         assert (created_at.hour - time.hour) in [self.delta_hours, self.delta_hours - 24]
         assert item.model_dump()["created_at"].hour == created_at.hour
+
+
+def query_without_app(pk: int) -> int:
+    async def runner() -> bool:
+        async with register_orm():
+            return await Users.filter(id__gt=pk).count()
+
+    return anyio.run(runner)
+
+
+def test_query_without_app():
+    multiprocessing.set_start_method("spawn")
+    with ProcessPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(query_without_app, 0)
+        assert future.result() >= 0

tortoise/contrib/fastapi/__init__.py

@@ -99,7 +99,7 @@ class RegisterTortoise(AbstractAsyncContextManager):
 
     def __init__(
         self,
-        app: FastAPI,
+        app: Optional[FastAPI] = None,
         config: Optional[dict] = None,
         config_file: Optional[str] = None,
         db_url: Optional[str] = None,
@@ -120,7 +120,7 @@ def __init__(
         self.timezone = timezone
         self._create_db = _create_db
 
-        if add_exception_handlers:
+        if add_exception_handlers and app is not None:
 
             @app.exception_handler(DoesNotExist)
             async def doesnotexist_exception_handler(request: "Request", exc: DoesNotExist):