django-ninja
django-ninja copied to clipboard
[BUG] TypeError: 'coroutine' object is not subscriptable [async pagination]
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
i think this is a duplicate of #548
-> wrong return type for a @pagination
decorated function
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
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.
@chrismaille i happend the same problem as you. can you solve it?
@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.
@chrismaille thanks a lot .
@chrismaille thanks a lot !!!
@vitalik Thanks for the great work! Is there any support for async paginate in the v1 expected?
@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?
Hi @jamesrkiger
Would you be open to a PR for async pagination?
sure - would be nice
@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.
@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
@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
@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,
)]