refactor: one CRUD path for sync and async engines

CHANGELOG.md

@@ -7,6 +7,21 @@ Versions align with [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ---
 
+## [Unreleased]
+
+### Changed
+- **Sync CRUD runs in a worker thread**: with a sync `Engine`, `list`, `retrieve`,
+  `create`, `update`, `destroy` and the response-cache calls ran on the event loop
+  thread, so one slow query held up every other request on that worker. They now
+  go through `starlette.concurrency.run_in_threadpool`. Engines on `StaticPool` or
+  `SingletonThreadPool` (in-memory SQLite) still run in place, because those pools
+  cannot be shared across threads.
+
+### Fixed
+- **`Meta.cache` was ignored on async engines**: cached GETs and cache invalidation
+  only ran on the sync path, so an app on an `AsyncEngine` never read or wrote the
+  cache. Both engine kinds now use it. The Redis calls run in a worker thread.
+
 ## [Unreleased] — 0.1.24
 
 ### Fixed

lightapi/cache_helper.py

@@ -1,8 +1,9 @@
 """Cache helper functions for GET caching and cache invalidation."""
 
 import json
-from typing import Callable
+from typing import Awaitable, Callable
 
+from starlette.concurrency import run_in_threadpool
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 
@@ -11,38 +12,56 @@
 
 def maybe_cached(cls: type, request: Request, fn: Callable[[], Response]) -> Response:
     """Serve from Redis cache (GET only) or call fn() and populate cache."""
-    cache_cfg = cls._meta.get("cache")
-    if cache_cfg is None:
+    if cls._meta.get("cache") is None:
         return fn()
-
-    key = _cache_key(cls, request)
-    try:
-        cached = get_cached(key)
-    except Exception:
-        cached = None
+    cached = _cached_response(cls, request)
     if cached is not None:
-        return JSONResponse(cached)
+        return cached
     response = fn()
-    if isinstance(response, JSONResponse) and response.status_code == 200:
-        try:
-            body = response.body
-            if hasattr(body, "decode"):
-                body = body.decode("utf-8")
-            set_cached(key, json.loads(body), cache_cfg.ttl)
-        except Exception:
-            pass
+    _store_response(cls, request, response)
     return response
 
 
-def maybe_invalidate_cache(cls: type, request: Request) -> None:
-    """Invalidate cache entries after mutating requests."""
-    if request.method == "GET":
-        return
-    cache_cfg = cls._meta.get("cache")
-    if cache_cfg is None:
+async def maybe_cached_async(
+    cls: type, request: Request, fn: Callable[[], Awaitable[Response]]
+) -> Response:
+    """maybe_cached() for a coroutine; Redis calls run in a worker thread."""
+    if cls._meta.get("cache") is None:
+        return await fn()
+    cached = await run_in_threadpool(_cached_response, cls, request)
+    if cached is not None:
+        return cached
+    response = await fn()
+    await run_in_threadpool(_store_response, cls, request, response)
+    return response
+
+
+async def invalidate_cache_after_write(cls: type, request: Request) -> None:
+    """Drop the endpoint's cached GETs after a request that may have changed data."""
+    if request.method == "GET" or cls._meta.get("cache") is None:
         return
+    await run_in_threadpool(invalidate_cache_prefix, _cache_key_prefix(cls))
 
-    invalidate_cache_prefix(_cache_key_prefix(cls))
+
+def _cached_response(cls: type, request: Request) -> Response | None:
+    """The cached response, or None on a miss or when Redis fails."""
+    try:
+        cached = get_cached(_cache_key(cls, request))
+    except Exception:
+        return None
+    return JSONResponse(cached) if cached is not None else None
+
+
+def _store_response(cls: type, request: Request, response: Response) -> None:
+    if not (isinstance(response, JSONResponse) and response.status_code == 200):
+        return
+    try:
+        body = response.body
+        if hasattr(body, "decode"):
+            body = body.decode("utf-8")
+        set_cached(_cache_key(cls, request), json.loads(body), cls._meta["cache"].ttl)
+    except Exception:
+        pass
 
 
 def _cache_key(cls: type, request: Request) -> str:

lightapi/handler_factory.py

@@ -9,6 +9,7 @@
 
 from lightapi.constants import RESPONSE_KEY_DETAIL, HTTPStatus
 from lightapi.rest import RestEndpoint
+from lightapi.session import run_blocking
 
 
 def make_collection_handler(
@@ -41,11 +42,21 @@ async def handler(request: Request) -> Response:
             if get_override and asyncio.iscoroutinefunction(get_override):
                 result = await get_override(endpoint, request)
             elif is_async:
-                result = await endpoint._list_async(request)
+                from lightapi.cache_helper import maybe_cached_async
+
+                result = await maybe_cached_async(
+                    cls, request, lambda: endpoint._list_async(request)
+                )
             else:
                 from lightapi.cache_helper import maybe_cached
 
-                result = maybe_cached(cls, request, lambda: endpoint.list(request))
+                result = await run_blocking(
+                    endpoint._get_engine(),
+                    maybe_cached,
+                    cls,
+                    request,
+                    lambda: endpoint.list(request),
+                )
         elif request.method == "POST":
             data = await read_body(request)
             post_override = getattr(cls, "post", None)
@@ -54,7 +65,9 @@ async def handler(request: Request) -> Response:
             elif is_async:
                 result = await endpoint._create_async(data)
             else:
-                result = endpoint.create(data)
+                result = await run_blocking(
+                    endpoint._get_engine(), endpoint.create, data
+                )
         else:
             allowed = ", ".join(sorted(cls._allowed_methods & {"GET", "POST"}))
             result = __import__(
@@ -67,10 +80,9 @@ async def handler(request: Request) -> Response:
 
         response = wrap_dict_response(result)
 
-        if not is_async:
-            from lightapi.cache_helper import maybe_invalidate_cache
+        from lightapi.cache_helper import invalidate_cache_after_write
 
-            maybe_invalidate_cache(cls, request)
+        await invalidate_cache_after_write(cls, request)
 
         if endpoint._background.tasks:
             response.background = endpoint._background
@@ -113,12 +125,20 @@ async def handler(request: Request) -> Response:
             if get_override and asyncio.iscoroutinefunction(get_override):
                 result = await get_override(endpoint, request)
             elif is_async:
-                result = await endpoint._retrieve_async(request, pk)
+                from lightapi.cache_helper import maybe_cached_async
+
+                result = await maybe_cached_async(
+                    cls, request, lambda: endpoint._retrieve_async(request, pk)
+                )
             else:
                 from lightapi.cache_helper import maybe_cached
 
-                result = maybe_cached(
-                    cls, request, lambda: endpoint.retrieve(request, pk)
+                result = await run_blocking(
+                    endpoint._get_engine(),
+                    maybe_cached,
+                    cls,
+                    request,
+                    lambda: endpoint.retrieve(request, pk),
                 )
         elif request.method in {"PUT", "PATCH"}:
             data = await read_body(request)
@@ -129,15 +149,19 @@ async def handler(request: Request) -> Response:
             elif is_async:
                 result = await endpoint._update_async(data, pk, partial=partial)
             else:
-                result = endpoint.update(data, pk, partial=partial)
+                result = await run_blocking(
+                    endpoint._get_engine(), endpoint.update, data, pk, partial
+                )
         elif request.method == "DELETE":
             delete_override = getattr(cls, "delete", None)
             if delete_override and asyncio.iscoroutinefunction(delete_override):
                 result = await delete_override(endpoint, request)
             elif is_async:
                 result = await endpoint._destroy_async(request, pk)
             else:
-                result = endpoint.destroy(request, pk)
+                result = await run_blocking(
+                    endpoint._get_engine(), endpoint.destroy, request, pk
+                )
         else:
             allowed = ", ".join(
                 sorted(cls._allowed_methods & {"GET", "PUT", "PATCH", "DELETE"})
@@ -152,10 +176,9 @@ async def handler(request: Request) -> Response:
 
         response = wrap_dict_response(result)
 
-        if not is_async:
-            from lightapi.cache_helper import maybe_invalidate_cache
+        from lightapi.cache_helper import invalidate_cache_after_write
 
-            maybe_invalidate_cache(cls, request)
+        await invalidate_cache_after_write(cls, request)
 
         if endpoint._background.tasks:
             response.background = endpoint._background

lightapi/pagination.py

@@ -55,15 +55,10 @@ async def paginate_async(
         session: AsyncSession,
         page_size: int,
     ) -> tuple[list[Any], int]:
-        """Async mirror of paginate(); uses await session.execute()."""
-        page = max(1, int(request.query_params.get(PAGE_PARAM, 1)))
-        offset = (page - 1) * page_size
-        count_stmt = select(func.count()).select_from(qs.subquery())
-        total: int = (await session.execute(count_stmt)).scalar_one()
-        rows = (
-            (await session.execute(qs.limit(page_size).offset(offset))).scalars().all()
+        """paginate() for an AsyncSession."""
+        return await session.run_sync(
+            lambda sync_session: self.paginate(request, qs, sync_session, page_size)
         )
-        return list(rows), total
 
     def wrap(
         self,
@@ -136,33 +131,10 @@ async def paginate_async(
         session: AsyncSession,
         page_size: int,
     ) -> tuple[list[Any], str | None]:
-        """Async mirror of paginate(); uses await session.execute()."""
-        cursor_str = request.query_params.get(CURSOR_PARAM)
-        if cursor_str:
-            try:
-                last_id = decode_cursor(cursor_str)
-                entity = (
-                    qs.columns_clause_froms[0]
-                    if hasattr(qs, "columns_clause_froms")
-                    else None
-                )
-                id_col = None
-                if entity is not None:
-                    id_col = entity.c.get("id")
-                if id_col is not None:
-                    qs = qs.where(id_col > last_id)
-            except Exception:
-                pass
-        rows = (
-            (await session.execute(qs.order_by("id").limit(page_size))).scalars().all()
+        """paginate() for an AsyncSession."""
+        return await session.run_sync(
+            lambda sync_session: self.paginate(request, qs, sync_session, page_size)
         )
-        next_cursor = None
-        if len(rows) == page_size:
-            last_obj = rows[-1]
-            last_row_id = getattr(last_obj, "id", None)
-            if last_row_id is not None:
-                next_cursor = encode_cursor(last_row_id)
-        return list(rows), next_cursor
 
     def wrap(
         self,