diff --git a/CHANGELOG.md b/CHANGELOG.md index d4e7fe9..fa92499 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/lightapi/cache_helper.py b/lightapi/cache_helper.py index f40b942..0feb162 100644 --- a/lightapi/cache_helper.py +++ b/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: diff --git a/lightapi/handler_factory.py b/lightapi/handler_factory.py index bc0bf29..11de341 100644 --- a/lightapi/handler_factory.py +++ b/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,7 +149,9 @@ 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): @@ -137,7 +159,9 @@ async def handler(request: Request) -> Response: 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 diff --git a/lightapi/pagination.py b/lightapi/pagination.py index 7e2a69b..dc50445 100644 --- a/lightapi/pagination.py +++ b/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, diff --git a/lightapi/rest.py b/lightapi/rest.py index dfe5c06..562aa6c 100644 --- a/lightapi/rest.py +++ b/lightapi/rest.py @@ -10,6 +10,7 @@ from uuid import UUID if TYPE_CHECKING: + from sqlalchemy.orm import Session from starlette.background import BackgroundTasks from sqlalchemy import ( @@ -43,6 +44,7 @@ normalise_serializer, resolve_fields, ) +from lightapi.session import get_async_session, get_sync_session _AUTO_FIELDS = AUTO_FIELDS @@ -338,70 +340,55 @@ def _serialize_row(self, row: Any, method: str) -> dict[str, Any]: result = {k: v for k, v in result.items() if k in fields} return result - def list(self, request: Request) -> Response: - """Handle GET /{path} — return collection.""" - from sqlalchemy.orm import Session + # ── CRUD core: one implementation, always given an open sync Session ───── - engine = self._get_engine() + def _list(self, session: Session, request: Request, qs: Any) -> Response: pagination_cfg = self._meta.get("pagination") + qs = self._run_filter_backends(request, qs) - with Session(engine) as session: - qs = self._get_queryset(request) - qs = self._run_filter_backends(request, qs) - - if pagination_cfg: - from lightapi.pagination import CursorPaginator, PageNumberPaginator + if pagination_cfg: + from lightapi.pagination import CursorPaginator, PageNumberPaginator - if pagination_cfg.style == "cursor": - pager = CursorPaginator() - rows, next_cursor = pager.paginate( - request, qs, session, pagination_cfg.page_size - ) - results = [self._serialize_row(r, "GET") for r in rows] - return JSONResponse(pager.wrap(results, next_cursor, None)) - else: - pager = PageNumberPaginator() - page = int(request.query_params.get("page", 1)) - rows, total = pager.paginate( - request, qs, session, pagination_cfg.page_size - ) - results = [self._serialize_row(r, "GET") for r in rows] - return JSONResponse( - pager.wrap( - request, results, total, page, pagination_cfg.page_size - ) - ) - - instances = session.execute(qs).scalars().all() - results = [self._serialize_row(inst, "GET") for inst in instances] - return JSONResponse({RESPONSE_KEY_RESULTS: results}) + if pagination_cfg.style == "cursor": + pager = CursorPaginator() + rows, next_cursor = pager.paginate( + request, qs, session, pagination_cfg.page_size + ) + results = [self._serialize_row(r, "GET") for r in rows] + return JSONResponse(pager.wrap(results, next_cursor, None)) + else: + pager = PageNumberPaginator() + page = int(request.query_params.get("page", 1)) + rows, total = pager.paginate( + request, qs, session, pagination_cfg.page_size + ) + results = [self._serialize_row(r, "GET") for r in rows] + return JSONResponse( + pager.wrap(request, results, total, page, pagination_cfg.page_size) + ) - def retrieve(self, request: Request, pk: int) -> Response: - """Handle GET /{path}/{id}.""" - from sqlalchemy.orm import Session + instances = session.execute(qs).scalars().all() + results = [self._serialize_row(inst, "GET") for inst in instances] + return JSONResponse({RESPONSE_KEY_RESULTS: results}) - engine = self._get_engine() + def _retrieve(self, session: Session, pk: int) -> Response: cls = type(self) - with Session(engine) as session: - instance = ( - session.execute( - sa_select(cls._model_class).where(cls._model_class.id == pk) - ) - .scalars() - .first() + instance = ( + session.execute( + sa_select(cls._model_class).where(cls._model_class.id == pk) ) - if instance is None: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND - ) - return JSONResponse(self._serialize_row(instance, "GET")) + .scalars() + .first() + ) + if instance is None: + return JSONResponse( + {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND + ) + return JSONResponse(self._serialize_row(instance, "GET")) - def create(self, data: dict[str, Any]) -> Response: - """Handle POST /{path} — validate input and insert row.""" + def _create(self, session: Session, data: dict[str, Any]) -> Response: from pydantic import ValidationError - from sqlalchemy.orm import Session - engine = self._get_engine() cls = type(self) try: validated = cls.__schema_create__.model_validate(data) @@ -411,27 +398,25 @@ def create(self, data: dict[str, Any]) -> Response: status_code=HTTPStatus.UNPROCESSABLE_ENTITY, ) - with Session(engine) as session: - now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - instance = cls._model_class( - **validated.model_dump(), - created_at=now, - updated_at=now, - version=1, - ) - session.add(instance) - session.flush() # executes INSERT, populates auto-increment id - session.refresh( - instance - ) # re-loads all columns (including DB-generated ones) - response_data = self._serialize_row(instance, "POST") - session.commit() - return JSONResponse(response_data, status_code=HTTPStatus.CREATED) + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + instance = cls._model_class( + **validated.model_dump(), + created_at=now, + updated_at=now, + version=1, + ) + session.add(instance) + session.flush() # executes INSERT, populates auto-increment id + session.refresh(instance) # re-loads DB-generated columns + return JSONResponse( + self._serialize_row(instance, "POST"), status_code=HTTPStatus.CREATED + ) - def update(self, data: dict[str, Any], pk: int, partial: bool = False) -> Response: - """Handle PUT/PATCH /{path}/{id} with optimistic locking.""" + def _update( + self, session: Session, data: dict[str, Any], pk: int, partial: bool + ) -> Response: + """PUT/PATCH with optimistic locking on the ``version`` column.""" from pydantic import ValidationError - from sqlalchemy.orm import Session client_version = data.get("version") if client_version is None: @@ -444,7 +429,6 @@ def update(self, data: dict[str, Any], pk: int, partial: bool = False) -> Respon status_code=HTTPStatus.UNPROCESSABLE_ENTITY, ) - engine = self._get_engine() cls = type(self) try: @@ -495,68 +479,85 @@ def update(self, data: dict[str, Any], pk: int, partial: bool = False) -> Respon update_data.pop("version", None) - with Session(engine) as session: - result = session.execute( - update(cls._model_class) - .where( - cls._model_class.id == pk, - cls._model_class.version == client_version, - ) - .values( - **update_data, - version=client_version + 1, - updated_at=datetime.datetime.now(datetime.timezone.utc).replace( - tzinfo=None - ), - ) + result = session.execute( + update(cls._model_class) + .where( + cls._model_class.id == pk, + cls._model_class.version == client_version, ) - if result.rowcount == 0: - exists = session.execute( - sa_select(cls._model_class.id).where(cls._model_class.id == pk) - ).first() - session.rollback() - if not exists: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, - status_code=HTTPStatus.NOT_FOUND, - ) + .values( + **update_data, + version=client_version + 1, + updated_at=datetime.datetime.now(datetime.timezone.utc).replace( + tzinfo=None + ), + ) + ) + if result.rowcount == 0: + exists = session.execute( + sa_select(cls._model_class.id).where(cls._model_class.id == pk) + ).first() + session.rollback() + if not exists: return JSONResponse( - {RESPONSE_KEY_DETAIL: "version conflict"}, - status_code=HTTPStatus.CONFLICT, - ) - # Re-fetch so all columns (including updated_at/version) are current - instance = ( - session.execute( - sa_select(cls._model_class).where(cls._model_class.id == pk) + {RESPONSE_KEY_DETAIL: "not found"}, + status_code=HTTPStatus.NOT_FOUND, ) - .scalars() - .first() + return JSONResponse( + {RESPONSE_KEY_DETAIL: "version conflict"}, + status_code=HTTPStatus.CONFLICT, ) - response_data = self._serialize_row(instance, "PUT") - session.commit() - return JSONResponse(response_data) - - def destroy(self, request: Request, pk: int) -> Response: - """Handle DELETE /{path}/{id}.""" - from sqlalchemy.orm import Session + # Re-fetch so all columns (including updated_at/version) are current + instance = ( + session.execute( + sa_select(cls._model_class).where(cls._model_class.id == pk) + ) + .scalars() + .first() + ) + return JSONResponse(self._serialize_row(instance, "PUT")) - engine = self._get_engine() + def _destroy(self, session: Session, pk: int) -> Response: cls = type(self) - with Session(engine) as session: - stmt = ( - delete(cls._model_class) - .where(cls._model_class.id == pk) - .returning(cls._model_class.id) + stmt = ( + delete(cls._model_class) + .where(cls._model_class.id == pk) + .returning(cls._model_class.id) + ) + if session.execute(stmt).first() is None: + return JSONResponse( + {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND ) - result = session.execute(stmt).first() - if result is None: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND - ) - session.commit() - return Response(status_code=HTTPStatus.NO_CONTENT) + return Response(status_code=HTTPStatus.NO_CONTENT) + + # ── Sync CRUD: the core inside a sync Session ───────────────────────────── + + def list(self, request: Request) -> Response: + """Handle GET /{path} — return collection.""" + with get_sync_session(self._get_engine()) as session: + return self._list(session, request, self._get_queryset(request)) - # ── Async queryset resolver ─────────────────────────────────────────────── + def retrieve(self, request: Request, pk: int) -> Response: + """Handle GET /{path}/{id}.""" + with get_sync_session(self._get_engine()) as session: + return self._retrieve(session, pk) + + def create(self, data: dict[str, Any]) -> Response: + """Handle POST /{path} — validate input and insert row.""" + with get_sync_session(self._get_engine()) as session: + return self._create(session, data) + + def update(self, data: dict[str, Any], pk: int, partial: bool = False) -> Response: + """Handle PUT/PATCH /{path}/{id} with optimistic locking.""" + with get_sync_session(self._get_engine()) as session: + return self._update(session, data, pk, partial) + + def destroy(self, request: Request, pk: int) -> Response: + """Handle DELETE /{path}/{id}.""" + with get_sync_session(self._get_engine()) as session: + return self._destroy(session, pk) + + # ── Async CRUD: the same core, run through AsyncSession.run_sync ────────── async def _get_queryset_async(self, request: Request) -> Any: """Resolve queryset; await if it is a coroutine function.""" @@ -586,222 +587,27 @@ def _get_async_engine(self) -> Any: return session_manager.engine - # ── Async CRUD ──────────────────────────────────────────────────────────── + async def _in_async_session(self, work: Callable[[Session], Response]) -> Response: + async with get_async_session(self._get_async_engine()) as session: + return await session.run_sync(work) async def _list_async(self, request: Request) -> Response: - """Async mirror of list(); uses AsyncSession.""" - from lightapi.session import get_async_session - - engine = self._get_async_engine() - pagination_cfg = self._meta.get("pagination") - - async with get_async_session(engine) as session: - qs = await self._get_queryset_async(request) - qs = self._run_filter_backends(request, qs) - - if pagination_cfg: - from lightapi.pagination import CursorPaginator, PageNumberPaginator - - if pagination_cfg.style == "cursor": - pager = CursorPaginator() - rows, next_cursor = await pager.paginate_async( - request, qs, session, pagination_cfg.page_size - ) - results = [self._serialize_row(r, "GET") for r in rows] - return JSONResponse(pager.wrap(results, next_cursor, None)) - else: - pager = PageNumberPaginator() - page = int(request.query_params.get("page", 1)) - rows, total = await pager.paginate_async( - request, qs, session, pagination_cfg.page_size - ) - results = [self._serialize_row(r, "GET") for r in rows] - return JSONResponse( - pager.wrap( - request, results, total, page, pagination_cfg.page_size - ) - ) - - instances = (await session.execute(qs)).scalars().all() - results = [self._serialize_row(inst, "GET") for inst in instances] - return JSONResponse({"results": results}) + """Async list(); an ``async def queryset`` is awaited first.""" + qs = await self._get_queryset_async(request) + return await self._in_async_session(lambda s: self._list(s, request, qs)) async def _retrieve_async(self, request: Request, pk: int) -> Response: - """Async mirror of retrieve(); uses AsyncSession.""" - from lightapi.session import get_async_session - - engine = self._get_async_engine() - cls = type(self) - async with get_async_session(engine) as session: - instance = ( - ( - await session.execute( - sa_select(cls._model_class).where(cls._model_class.id == pk) - ) - ) - .scalars() - .first() - ) - if instance is None: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND - ) - return JSONResponse(self._serialize_row(instance, "GET")) + return await self._in_async_session(lambda s: self._retrieve(s, pk)) async def _create_async(self, data: dict[str, Any]) -> Response: - """Async mirror of create(); ORM-style insert with flush/refresh.""" - from pydantic import ValidationError - - from lightapi.session import get_async_session - - engine = self._get_async_engine() - cls = type(self) - try: - validated = cls.__schema_create__.model_validate(data) - except ValidationError as exc: - return JSONResponse( - {RESPONSE_KEY_DETAIL: exc.errors()}, - status_code=HTTPStatus.UNPROCESSABLE_ENTITY, - ) - - async with get_async_session(engine) as session: - now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - instance = cls._model_class( - **validated.model_dump(), - created_at=now, - updated_at=now, - version=1, - ) - session.add(instance) - await session.flush() - await session.refresh(instance) - response_data = self._serialize_row(instance, "POST") - return JSONResponse(response_data, status_code=HTTPStatus.CREATED) + return await self._in_async_session(lambda s: self._create(s, data)) async def _update_async( self, data: dict[str, Any], pk: int, partial: bool = False ) -> Response: - """Async mirror of update() with optimistic locking.""" - from pydantic import ValidationError - - from lightapi.session import get_async_session - - client_version = data.get("version") - if client_version is None: - return JSONResponse( - { - RESPONSE_KEY_DETAIL: [ - {"loc": ["version"], "msg": "Field required", "type": "missing"} - ] - }, - status_code=HTTPStatus.UNPROCESSABLE_ENTITY, - ) - - engine = self._get_async_engine() - cls = type(self) - - try: - if partial: - from typing import Optional as _Opt - - from pydantic import ConfigDict as _CD - from pydantic import create_model as _cm - - patch_fields: dict[str, Any] = {} - for fname, finfo in cls.__schema_create__.model_fields.items(): - ann = finfo.annotation - patch_fields[fname] = (_Opt[ann], None) # type: ignore[valid-type] - PatchSchema = _cm( - f"{cls.__name__}PatchSchema", - __config__=_CD(from_attributes=True), - **patch_fields, - ) - validated = PatchSchema.model_validate(data) - from sqlalchemy import inspect as _sa_inspect - - nullable_cols: set[str] = { - attr.key - for attr in _sa_inspect(cls._model_class).mapper.column_attrs - if any(c.nullable for c in attr.columns) - } - update_data = { - k: v - for k, v in validated.model_dump(exclude_unset=True).items() - if k not in _AUTO_FIELDS and (v is not None or k in nullable_cols) - } - else: - validated = cls.__schema_create__.model_validate(data) - update_data = { - k: v - for k, v in validated.model_dump().items() - if k not in _AUTO_FIELDS - } - except ValidationError as exc: - return JSONResponse( - {RESPONSE_KEY_DETAIL: exc.errors()}, - status_code=HTTPStatus.UNPROCESSABLE_ENTITY, - ) - - update_data.pop("version", None) - - async with get_async_session(engine) as session: - result = await session.execute( - update(cls._model_class) - .where( - cls._model_class.id == pk, - cls._model_class.version == client_version, - ) - .values( - **update_data, - version=client_version + 1, - updated_at=datetime.datetime.now(datetime.timezone.utc).replace( - tzinfo=None - ), - ) - ) - if result.rowcount == 0: - exists = ( - await session.execute( - sa_select(cls._model_class.id).where(cls._model_class.id == pk) - ) - ).first() - await session.rollback() - if not exists: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, - status_code=HTTPStatus.NOT_FOUND, - ) - return JSONResponse( - {RESPONSE_KEY_DETAIL: "version conflict"}, - status_code=HTTPStatus.CONFLICT, - ) - instance = ( - ( - await session.execute( - sa_select(cls._model_class).where(cls._model_class.id == pk) - ) - ) - .scalars() - .first() - ) - response_data = self._serialize_row(instance, "PUT") - return JSONResponse(response_data) + return await self._in_async_session( + lambda s: self._update(s, data, pk, partial) + ) async def _destroy_async(self, request: Request, pk: int) -> Response: - """Async mirror of destroy().""" - from lightapi.session import get_async_session - - engine = self._get_async_engine() - cls = type(self) - async with get_async_session(engine) as session: - stmt = ( - delete(cls._model_class) - .where(cls._model_class.id == pk) - .returning(cls._model_class.id) - ) - result = (await session.execute(stmt)).first() - if result is None: - return JSONResponse( - {RESPONSE_KEY_DETAIL: "not found"}, status_code=HTTPStatus.NOT_FOUND - ) - return Response(status_code=HTTPStatus.NO_CONTENT) + return await self._in_async_session(lambda s: self._destroy(s, pk)) diff --git a/lightapi/session.py b/lightapi/session.py index de57ba8..1a3c8f1 100644 --- a/lightapi/session.py +++ b/lightapi/session.py @@ -3,14 +3,23 @@ from __future__ import annotations from contextlib import asynccontextmanager, contextmanager -from typing import TYPE_CHECKING, AsyncGenerator, Generator +from typing import TYPE_CHECKING, Any, AsyncGenerator, Callable, Generator, TypeVar from sqlalchemy.orm import Session +from sqlalchemy.pool import SingletonThreadPool, StaticPool +from starlette.concurrency import run_in_threadpool if TYPE_CHECKING: from sqlalchemy import Engine from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession +T = TypeVar("T") + +# In-memory SQLite runs on one of these. SingletonThreadPool gives every thread +# its own empty database, and StaticPool shares a single connection that is not +# safe to use from two threads at once. +_SINGLE_CONNECTION_POOLS = (SingletonThreadPool, StaticPool) + @contextmanager def get_sync_session(engine: Engine) -> Generator[Session, None, None]: @@ -36,3 +45,13 @@ async def get_async_session(engine: AsyncEngine) -> AsyncGenerator[AsyncSession, except Exception: await session.rollback() raise + + +async def run_blocking(engine: Engine, fn: Callable[..., T], *args: Any) -> T: + """Run a blocking database call in a worker thread so the event loop stays free. + + Engines on a single-connection pool run the call in place instead. + """ + if isinstance(engine.pool, _SINGLE_CONNECTION_POOLS): + return fn(*args) + return await run_in_threadpool(fn, *args) diff --git a/tests/test_cache_async.py b/tests/test_cache_async.py new file mode 100644 index 0000000..81e47d5 --- /dev/null +++ b/tests/test_cache_async.py @@ -0,0 +1,72 @@ +"""Meta.cache must also work when the app runs on an AsyncEngine.""" + +from unittest.mock import patch + +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from sqlalchemy.ext.asyncio import create_async_engine + +from lightapi import Cache, LightApi, RestEndpoint +from lightapi.fields import Field as LField + +TTL = 60 + + +class AsyncCachedEndpoint(RestEndpoint): + name: str = LField(min_length=1) + + class Meta: + cache = Cache(ttl=TTL) + + +@pytest_asyncio.fixture +async def client(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + app = LightApi(engine=engine, mode="async") + app.register({"/asynccached": AsyncCachedEndpoint}) + async with AsyncClient( + transport=ASGITransport(app=app.build_app()), base_url="http://test" + ) as c: + yield c + await engine.dispose() + + +async def test_async_get_list_cache_hit_returns_cached_body(client): + cached_body = {"results": [{"id": 1, "name": "from-cache"}]} + + with patch("lightapi.cache_helper.get_cached", return_value=cached_body): + with patch("lightapi.cache_helper.set_cached") as mock_set: + response = await client.get("/asynccached") + + assert response.status_code == 200 + assert response.json() == cached_body + mock_set.assert_not_called() + + +async def test_async_get_list_cache_miss_stores_response_with_ttl(client): + with patch("lightapi.cache_helper.get_cached", return_value=None): + with patch("lightapi.cache_helper.set_cached") as mock_set: + response = await client.get("/asynccached") + + assert response.status_code == 200 + key, body, ttl = mock_set.call_args.args + assert "AsyncCachedEndpoint" in key + assert body == {"results": []} + assert ttl == TTL + + +async def test_async_post_invalidates_the_endpoint_prefix(client): + with patch("lightapi.cache_helper.invalidate_cache_prefix") as mock_invalidate: + response = await client.post("/asynccached", json={"name": "fresh"}) + + assert response.status_code == 201 + mock_invalidate.assert_called_once_with("lightapi:AsyncCachedEndpoint:") + + +async def test_async_get_does_not_invalidate(client): + with patch("lightapi.cache_helper.get_cached", return_value=None): + with patch("lightapi.cache_helper.set_cached"): + with patch("lightapi.cache_helper.invalidate_cache_prefix") as mock_inv: + await client.get("/asynccached") + + mock_inv.assert_not_called() diff --git a/tests/test_paginate_async.py b/tests/test_paginate_async.py new file mode 100644 index 0000000..d762570 --- /dev/null +++ b/tests/test_paginate_async.py @@ -0,0 +1,71 @@ +"""paginate_async() is public API that the framework itself no longer calls.""" + +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from pydantic import Field as PydanticField +from sqlalchemy import select as sa_select +from sqlalchemy.ext.asyncio import create_async_engine +from starlette.requests import Request + +from lightapi import LightApi, RestEndpoint, get_async_session +from lightapi.pagination import CursorPaginator, PageNumberPaginator, decode_cursor + +ROWS = 5 +PAGE_SIZE = 2 + + +class PagedGadget(RestEndpoint): + name: str = PydanticField(min_length=1) + + +def _get_request(query: bytes) -> Request: + return Request( + { + "type": "http", + "method": "GET", + "path": "/pagedgadgets", + "query_string": query, + "headers": [], + } + ) + + +@pytest_asyncio.fixture +async def seeded_engine(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + app = LightApi(engine=engine, mode="async") + app.register({"/pagedgadgets": PagedGadget}) + async with AsyncClient( + transport=ASGITransport(app=app.build_app()), base_url="http://test" + ) as client: + for n in range(ROWS): + created = await client.post("/pagedgadgets", json={"name": f"gadget-{n}"}) + assert created.status_code == 201 + yield engine + await engine.dispose() + + +async def test_page_number_paginate_async_second_page_returns_rows_and_total( + seeded_engine, +): + qs = sa_select(PagedGadget).order_by(PagedGadget.id) + + async with get_async_session(seeded_engine) as session: + rows, total = await PageNumberPaginator().paginate_async( + _get_request(b"page=2"), qs, session, PAGE_SIZE + ) + + assert total == ROWS + assert [row.name for row in rows] == ["gadget-2", "gadget-3"] + + +async def test_cursor_paginate_async_full_page_returns_next_cursor(seeded_engine): + qs = sa_select(PagedGadget) + + async with get_async_session(seeded_engine) as session: + rows, next_cursor = await CursorPaginator().paginate_async( + _get_request(b""), qs, session, PAGE_SIZE + ) + + assert [row.name for row in rows] == ["gadget-0", "gadget-1"] + assert decode_cursor(next_cursor) == rows[-1].id diff --git a/tests/test_sync_offload.py b/tests/test_sync_offload.py new file mode 100644 index 0000000..ac1f156 --- /dev/null +++ b/tests/test_sync_offload.py @@ -0,0 +1,77 @@ +"""Sync CRUD must not run on the event loop thread, except on single-connection pools.""" + +import threading + +import pytest +from sqlalchemy import create_engine +from sqlalchemy import select as sa_select +from sqlalchemy.pool import StaticPool +from starlette.requests import Request +from starlette.responses import Response +from starlette.testclient import TestClient + +from lightapi import LightApi, RestEndpoint +from lightapi.core import Middleware +from lightapi.fields import Field as LField + +threads: dict[str, int] = {} + + +class RecordLoopThread(Middleware): + async def process( + self, request: Request, response: Response | None + ) -> Response | None: + if response is None: + threads["loop"] = threading.get_ident() + return response + + +class PooledProbe(RestEndpoint): + label: str = LField(min_length=1) + + def queryset(self, request: Request): + threads["crud"] = threading.get_ident() + return sa_select(type(self)._model_class) + + +class StaticProbe(RestEndpoint): + label: str = LField(min_length=1) + + def queryset(self, request: Request): + threads["crud"] = threading.get_ident() + return sa_select(type(self)._model_class) + + +@pytest.fixture(autouse=True) +def reset_threads(): + threads.clear() + + +def _client(engine, route: str, endpoint: type) -> TestClient: + app = LightApi(engine=engine, middlewares=[RecordLoopThread]) + app.register({route: endpoint}) + return TestClient(app.build_app()) + + +def test_sync_list_on_pooled_engine_runs_off_the_event_loop_thread(tmp_path): + engine = create_engine(f"sqlite:///{tmp_path / 'offload.db'}") + client = _client(engine, "/pooledprobes", PooledProbe) + + response = client.get("/pooledprobes") + + assert response.status_code == 200 + assert threads["crud"] != threads["loop"] + + +def test_sync_list_on_static_pool_stays_on_the_event_loop_thread(): + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + client = _client(engine, "/staticprobes", StaticProbe) + + response = client.get("/staticprobes") + + assert response.status_code == 200 + assert threads["crud"] == threads["loop"]