[BUG] TypeError: 'coroutine' object is not subscriptable [async pagination]

[ddzyx]

async def search(request, keywords: str):
    item = await sync_to_async(list)(Movie.objects.all())

    if item:
        return item

Error code after running

Traceback (most recent call last):
  File "D:\Program Files (x86)\Anaconda3\envs\django\lib\site-packages\ninja\operation.py", line 99, in run
    result = self.view_func(request, **values)
  File "D:\Program Files (x86)\Anaconda3\envs\django\lib\site-packages\ninja\pagination.py", line 145, in view_with_pagination
    result = paginator.paginate_queryset(
  File "D:\Program Files (x86)\Anaconda3\envs\django\lib\site-packages\ninja\pagination.py", line 69, in paginate_queryset
    "items": queryset[offset : offset + limit],
TypeError: 'coroutine' object is not subscriptable

[hiaselhans]

i think this is a duplicate of #548

-> wrong return type for a @pagination decorated function

[chrismaille]

In fact, the queryset parameter are receiving the decorated function, which was a coroutine, instead receiving the queryset which returns from this function/coroutine. Not sure why, but I believe this can be a bug indeed.

from django.contrib.auth.models import Permission
from asgiref.sync import sync_to_async

# Pytest Asyncio
async def test_paginate(async_client):
    
    # Arrange
    class PermissionOut(ModelSchema):
        codename: str
        class Config:
            model = Permission
            model_fields = ["codename"]

    @api.get("/test", response=list[PermissionOut], auth=None)
    @paginate
    async def long_response(request):
        return await sync_to_async(list)(Permission.objects.all())

    # Act
    response = await async_client.get("/test")

    # Assert
    assert response.status_code == 200

Traceback

self = <ninja.pagination.LimitOffsetPagination object at 0x7f3ac6112c20>
queryset = <coroutine object test_paginate.<locals>.long_response at 0x7f3ac6140e40>
pagination = Input(limit=100, offset=0), params = {}, offset = 0, limit = 100

    def paginate_queryset(
        self,
        queryset: QuerySet,
        pagination: Input,
        **params: DictStrAny,
    ) -> Any:
        offset = pagination.offset
        limit: int = pagination.limit
        return {
>           "items": queryset[offset : offset + limit],
            "count": self._items_count(queryset),
        }  # noqa: E203
E       TypeError: 'coroutine' object is not subscriptable

[chrismaille]

I think the issue is here:

https://github.com/vitalik/django-ninja/blob/3e9c10d5c8915a2ef76b08de8cdff98b45ca55f8/ninja/pagination.py#L143

Because view_with_pagination run the function syncronously, items is the coroutine.

[sunboy123]

@chrismaille i happend the same problem as you. can you solve it?

[chrismaille]

@sunboy123 sure, I've created a async version for the decorator. Here is the code:

import inspect
from functools import partial, wraps
from typing import Any, Callable, Tuple, Type

from asgiref.sync import sync_to_async
from django.db.models import QuerySet
from ninja.constants import NOT_SET
from ninja.pagination import LimitOffsetPagination, PaginationBase, make_response_paginated
from ninja.types import DictStrAny


class AsyncLimitOffsetPagination(LimitOffsetPagination):
    async def paginate_queryset(
        self,
        queryset: QuerySet,
        pagination: LimitOffsetPagination.Input,
        **params: DictStrAny,
    ):
        offset = pagination.offset
        limit: int = pagination.limit

        @sync_to_async
        def process_query_set():
            return {
                "items": queryset[offset : offset + limit] if queryset else [],
                "count": self._items_count(queryset) if queryset else 0,
            }

        return await process_query_set()


def apaginate(func_or_pgn_class: Any = NOT_SET, **paginator_params: DictStrAny) -> Callable:

    isfunction = inspect.isfunction(func_or_pgn_class)
    isnotset = func_or_pgn_class == NOT_SET

    pagination_class: Type[PaginationBase] = AsyncLimitOffsetPagination

    if isfunction:
        return _inject_pagination(func_or_pgn_class, pagination_class)

    if not isnotset:
        pagination_class = func_or_pgn_class

    async def wrapper(func: Callable) -> Any:
        return await _inject_pagination(func, pagination_class, **paginator_params)

    return wrapper


def _inject_pagination(
    func: Callable,
    paginator_class: Type[PaginationBase],
    **paginator_params: Any,
) -> Callable:
    paginator: PaginationBase = paginator_class(**paginator_params)

    @wraps(func)
    async def view_with_pagination(*args: Tuple[Any], **kwargs: DictStrAny) -> Any:
        pagination_params = kwargs.pop("ninja_pagination")
        if paginator.pass_parameter:
            kwargs[paginator.pass_parameter] = pagination_params

        items = await func(*args, **kwargs)

        result = await paginator.paginate_queryset(items, pagination=pagination_params, **kwargs)
        if paginator.Output:
            result[paginator.items_attribute] = list(result[paginator.items_attribute])
            # ^ forcing queryset evaluation #TODO: check why pydantic did not do it here
        return result

    view_with_pagination._ninja_contribute_args = [  # type: ignore
        (
            "ninja_pagination",
            paginator.Input,
            paginator.InputSource,
        ),
    ]

    if paginator.Output:
        view_with_pagination._ninja_contribute_to_operation = partial(  # type: ignore
            make_response_paginated,
            paginator,
        )

    return view_with_pagination

An example:

from django.http import HttpRequest
from django.db import models, QuerySet
from ninja import NinjaAPI, ModelSchema

class MyModel(models.Model):
   foo = models.CharField(max_length=255)
   is_active = models.BooleanField(default=True)

class MyModelOut(ModelSchema):
  class Meta:
     model = MyModel
     model_fields = ["foo"]


api= NinjaAPI()

@api.get("/my-model", response=list[MyModelOut])
@apaginate
async def list_model(request: HttpRequest) -> QuerySet[MyModel]:
    return MyModel.objects.filter(is_active=True)

Please note, with this solution, if you need to use another Pagination class (like AsyncLimitOffsetPagination) you need to make that async compatible too.

[sunboy123]

@chrismaille thanks a lot .

[joaoflaviosantosmoniari]

@chrismaille thanks a lot !!!

[babusqb]

@vitalik Thanks for the great work! Is there any support for async paginate in the v1 expected?

[jamesrkiger]

@vitalik I made some modifications to the code suggested above for a project I am working on and got async pagination working. We'd prefer to have these changes implemented upstream rather than copying in modified code and I would be happy to contribute to help make that happen. Would you be open to a PR for async pagination?

[vitalik]

Hi @jamesrkiger

Would you be open to a PR for async pagination?

sure - would be nice

[jamesrkiger]

@vitalik I have been working on a solution to the async pagination issue as I mentioned. If you're curious, you can see my WIP here. It's not done yet since I still want to find a graceful way to abstract view_with_pagination and avoid repetition. But in the meantime I was hoping to get your input on testing. Currently I am trying to keep in line with the existing testing for pagination, which only tests on lists rather than actual querysets. Do you think I should add db tests for this, or are the lists sufficient? Also, should I add the testing in a new test_pagination_async file or keep it in the existing pagination files?

I'd also be happy to hear what you think about my general approach to the issue, which is to check for whether the view function is async in _inject_pagination and then, if so, call a dedicated async pagination class method named apaginate_queryset instead of the current standard paginate_queryset sync method. Currently I am checking for this method when an async paginated view function is created and throwing a config error if it is not found. I made a separate base class for async pagination, which would enable ninja users to continue creating sync pagination without worrying about adding async support if they don't need it.

[dtasev]

@jamesrkiger I don't think the repetition of a few lines should stop you from making the PR - especially when it's covered in tests. The merge of a sync and async coroutine will probably require more lines of code than are duplicated, and if someone knows how to write it neatly then that would be a more visible place to do that

[karambaq]

@sunboy123 sure, I've created a async version for the decorator. Here is the code:

import inspect
from functools import partial, wraps
from typing import Any, Callable, Tuple, Type

from asgiref.sync import sync_to_async
from django.db.models import QuerySet
from ninja.constants import NOT_SET
from ninja.pagination import LimitOffsetPagination, PaginationBase, make_response_paginated
from ninja.types import DictStrAny


class AsyncLimitOffsetPagination(LimitOffsetPagination):
    async def paginate_queryset(
        self,
        queryset: QuerySet,
        pagination: LimitOffsetPagination.Input,
        **params: DictStrAny,
    ):
        offset = pagination.offset
        limit: int = pagination.limit

        @sync_to_async
        def process_query_set():
            return {
                "items": queryset[offset : offset + limit] if queryset else [],
                "count": self._items_count(queryset) if queryset else 0,
            }

        return await process_query_set()


def apaginate(func_or_pgn_class: Any = NOT_SET, **paginator_params: DictStrAny) -> Callable:

    isfunction = inspect.isfunction(func_or_pgn_class)
    isnotset = func_or_pgn_class == NOT_SET

    pagination_class: Type[PaginationBase] = AsyncLimitOffsetPagination

    if isfunction:
        return _inject_pagination(func_or_pgn_class, pagination_class)

    if not isnotset:
        pagination_class = func_or_pgn_class

    async def wrapper(func: Callable) -> Any:
        return await _inject_pagination(func, pagination_class, **paginator_params)

    return wrapper


def _inject_pagination(
    func: Callable,
    paginator_class: Type[PaginationBase],
    **paginator_params: Any,
) -> Callable:
    paginator: PaginationBase = paginator_class(**paginator_params)

    @wraps(func)
    async def view_with_pagination(*args: Tuple[Any], **kwargs: DictStrAny) -> Any:
        pagination_params = kwargs.pop("ninja_pagination")
        if paginator.pass_parameter:
            kwargs[paginator.pass_parameter] = pagination_params

        items = await func(*args, **kwargs)

        result = await paginator.paginate_queryset(items, pagination=pagination_params, **kwargs)
        if paginator.Output:
            result[paginator.items_attribute] = list(result[paginator.items_attribute])
            # ^ forcing queryset evaluation #TODO: check why pydantic did not do it here
        return result

    view_with_pagination._ninja_contribute_args = [  # type: ignore
        (
            "ninja_pagination",
            paginator.Input,
            paginator.InputSource,
        ),
    ]

    if paginator.Output:
        view_with_pagination._ninja_contribute_to_operation = partial(  # type: ignore
            make_response_paginated,
            paginator,
        )

    return view_with_pagination

An example:

from django.http import HttpRequest
from django.db import models, QuerySet
from ninja import NinjaAPI, ModelSchema

class MyModel(models.Model):
   foo = models.CharField(max_length=255)
   is_active = models.BooleanField(default=True)

class MyModelOut(ModelSchema):
  class Meta:
     model = MyModel
     model_fields = ["foo"]


api= NinjaAPI()

@api.get("/my-model", response=list[MyModelOut])
@apaginate
async def list_model(request: HttpRequest) -> QuerySet[MyModel]:
    return MyModel.objects.filter(is_active=True)

Please note, with this solution, if you need to use another Pagination class (like AsyncLimitOffsetPagination) you need to make that async compatible too.

Thanks for this. Which version of django-ninja you were using here?

I'm facing this problem with this code:

   for callback in callbacks:
TypeError: 'functools.partial' object is not iterable

[kajiczech]

@karambaq

I'm facing this problem with this code:

   for callback in callbacks:
TypeError: 'functools.partial' object is not iterable

I fixed it by adding a list of partial to _ninja_contribute_to_operation in view_with_pagination()

view_with_pagination._ninja_contribute_to_operation = [partial(  # type: ignore
            make_response_paginated,
            paginator,
        )]