From 5f4969c82dc8476afd7803e73fe2d8b865933aee Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 10:14:01 -0300 Subject: [PATCH 1/3] refactor: one CRUD implementation for sync and async engines RestEndpoint carried every CRUD method twice: list, retrieve, create, update and destroy, and again as _list_async ... _destroy_async, copied line by line with `await` added. update alone was 107 lines, twice. Any fix had to be made in both copies. The logic now lives once in _list, _retrieve, _create, _update and _destroy, which take an open sync Session. The public methods are thin: - list/retrieve/create/update/destroy open a sync session (get_sync_session) and call the core. - the _*_async helpers call the same core through AsyncSession.run_sync, which is SQLAlchemy's supported way to run sync ORM code on an async driver. Names and signatures do not change. The _*_async helpers and paginate_async stay because the README, docs/advanced/async.md and three examples tell users to call them. paginate_async now delegates to paginate the same way; it had no test, so it gets one. Checked on aiosqlite: the five test_async_* files and test_background_tasks pass through the new path (insert with refresh, update with rowcount, delete with RETURNING, both paginators, async queryset, reflection). Not checked on asyncpg: no Postgres here. Tests: 329 passed (327 + 2 new), 1 xfailed, 2 xpassed. --- lightapi/pagination.py | 40 +-- lightapi/rest.py | 474 +++++++++++------------------------ tests/test_paginate_async.py | 71 ++++++ 3 files changed, 217 insertions(+), 368 deletions(-) create mode 100644 tests/test_paginate_async.py 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/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 From 432ec497d315ab819e4438cf0e0b65ba60b42002 Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 10:17:33 -0300 Subject: [PATCH 2/3] fix: run sync CRUD off the event loop thread With a sync Engine the route handlers are `async def` but called endpoint.list(), create(), update() and the rest directly, so every database round trip ran on the event loop thread. One slow query stalled all other requests on that worker. The Redis cache calls had the same problem. The sync branches now go through session.run_blocking, which uses Starlette's run_in_threadpool. Cache invalidation goes the same way and is skipped for GET, which never invalidates. Engines on StaticPool or SingletonThreadPool keep running in place. Those are the in-memory SQLite pools: SingletonThreadPool gives every thread its own empty database, and StaticPool shares one connection that is not safe across threads. The docs install snippet and most examples use them, and they would have broken under a plain thread pool. tests/test_sync_offload.py pins both cases by comparing the thread that runs the queryset with the event loop thread. It fails without this change. Tests: 331 passed, 1 xfailed, 2 xpassed. --- CHANGELOG.md | 10 +++++ lightapi/handler_factory.py | 41 +++++++++++++++----- lightapi/session.py | 21 +++++++++- tests/test_sync_offload.py | 77 +++++++++++++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 11 deletions(-) create mode 100644 tests/test_sync_offload.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d4e7fe9..dcc4604 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ 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. + ## [Unreleased] — 0.1.24 ### Fixed diff --git a/lightapi/handler_factory.py b/lightapi/handler_factory.py index bc0bf29..9b6f09e 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( @@ -45,7 +46,13 @@ async def handler(request: Request) -> Response: 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 +61,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 +76,12 @@ async def handler(request: Request) -> Response: response = wrap_dict_response(result) - if not is_async: + if not is_async and request.method != "GET": from lightapi.cache_helper import maybe_invalidate_cache - maybe_invalidate_cache(cls, request) + await run_blocking( + endpoint._get_engine(), maybe_invalidate_cache, cls, request + ) if endpoint._background.tasks: response.background = endpoint._background @@ -117,8 +128,12 @@ async def handler(request: Request) -> Response: 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 +144,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 +154,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 +171,12 @@ async def handler(request: Request) -> Response: response = wrap_dict_response(result) - if not is_async: + if not is_async and request.method != "GET": from lightapi.cache_helper import maybe_invalidate_cache - maybe_invalidate_cache(cls, request) + await run_blocking( + endpoint._get_engine(), maybe_invalidate_cache, cls, request + ) if endpoint._background.tasks: response.background = endpoint._background 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_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"] From 49e2a4204c898224acb877e5e27d230421c4baa7 Mon Sep 17 00:00:00 2001 From: iklo Date: Fri, 18 Sep 2026 10:21:23 -0300 Subject: [PATCH 3/3] fix: honour Meta.cache on async engines maybe_cached and the cache invalidation lived only in the sync branches of the route handlers. An app on an AsyncEngine with `cache = Cache(...)` never read, wrote or invalidated anything, with no warning. cache_helper now splits the work into _cached_response and _store_response. maybe_cached (sync) and the new maybe_cached_async share them, so the rules exist once. The Redis client is blocking, so the async path sends those calls to a worker thread and skips the hop when the endpoint has no cache configured. invalidate_cache_after_write replaces maybe_invalidate_cache and is called for both engine kinds, which removes one `if not is_async` from each handler. tests/test_cache_async.py covers hit, miss with the stored key, body and ttl, invalidation on POST, and no invalidation on GET. Three of the four fail without this change. Tests: 335 passed, 1 xfailed, 2 xpassed. --- CHANGELOG.md | 5 +++ lightapi/cache_helper.py | 69 ++++++++++++++++++++++------------- lightapi/handler_factory.py | 26 +++++++------- tests/test_cache_async.py | 72 +++++++++++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 37 deletions(-) create mode 100644 tests/test_cache_async.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dcc4604..fa92499 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,11 @@ Versions align with [Semantic Versioning](https://semver.org/spec/v2.0.0.html). `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 9b6f09e..11de341 100644 --- a/lightapi/handler_factory.py +++ b/lightapi/handler_factory.py @@ -42,7 +42,11 @@ 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 @@ -76,12 +80,9 @@ async def handler(request: Request) -> Response: response = wrap_dict_response(result) - if not is_async and request.method != "GET": - from lightapi.cache_helper import maybe_invalidate_cache + from lightapi.cache_helper import invalidate_cache_after_write - await run_blocking( - endpoint._get_engine(), maybe_invalidate_cache, cls, request - ) + await invalidate_cache_after_write(cls, request) if endpoint._background.tasks: response.background = endpoint._background @@ -124,7 +125,11 @@ 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 @@ -171,12 +176,9 @@ async def handler(request: Request) -> Response: response = wrap_dict_response(result) - if not is_async and request.method != "GET": - from lightapi.cache_helper import maybe_invalidate_cache + from lightapi.cache_helper import invalidate_cache_after_write - await run_blocking( - endpoint._get_engine(), maybe_invalidate_cache, cls, request - ) + await invalidate_cache_after_write(cls, request) if endpoint._background.tasks: response.background = endpoint._background 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()