add querysetproxy methods and tests
This commit is contained in:
@ -276,6 +276,59 @@ class QuerysetProxy(Generic[T]):
|
|||||||
)
|
)
|
||||||
return await queryset.delete(**kwargs) # type: ignore
|
return await queryset.delete(**kwargs) # type: ignore
|
||||||
|
|
||||||
|
async def values(
|
||||||
|
self, fields: Union[List, str, Set, Dict] = None, exclude_through: bool = False
|
||||||
|
) -> List:
|
||||||
|
"""
|
||||||
|
Return a list of dictionaries with column values in order of the fields
|
||||||
|
passed or all fields from queried models.
|
||||||
|
|
||||||
|
To filter for given row use filter/exclude methods before values,
|
||||||
|
to limit number of rows use limit/offset or paginate before values.
|
||||||
|
|
||||||
|
Note that it always return a list even for one row from database.
|
||||||
|
|
||||||
|
:param exclude_through: flag if through models should be excluded
|
||||||
|
:type exclude_through: bool
|
||||||
|
:param fields: field name or list of field names to extract from db
|
||||||
|
:type fields: Union[List, str, Set, Dict]
|
||||||
|
"""
|
||||||
|
return await self.queryset.values(
|
||||||
|
fields=fields, exclude_through=exclude_through,
|
||||||
|
)
|
||||||
|
|
||||||
|
async def values_list(
|
||||||
|
self,
|
||||||
|
fields: Union[List, str, Set, Dict] = None,
|
||||||
|
flatten: bool = False,
|
||||||
|
exclude_through: bool = False,
|
||||||
|
) -> List:
|
||||||
|
"""
|
||||||
|
Return a list of tuples with column values in order of the fields passed or
|
||||||
|
all fields from queried models.
|
||||||
|
|
||||||
|
When one field is passed you can flatten the list of tuples into list of values
|
||||||
|
of that single field.
|
||||||
|
|
||||||
|
To filter for given row use filter/exclude methods before values,
|
||||||
|
to limit number of rows use limit/offset or paginate before values.
|
||||||
|
|
||||||
|
Note that it always return a list even for one row from database.
|
||||||
|
|
||||||
|
:param exclude_through: flag if through models should be excluded
|
||||||
|
:type exclude_through: bool
|
||||||
|
:param fields: field name or list of field names to extract from db
|
||||||
|
:type fields: Union[str, List[str]]
|
||||||
|
:param flatten: when one field is passed you can flatten the list of tuples
|
||||||
|
:type flatten: bool
|
||||||
|
"""
|
||||||
|
return await self.queryset.values(
|
||||||
|
fields=fields,
|
||||||
|
exclude_through=exclude_through,
|
||||||
|
_as_dict=False,
|
||||||
|
_flatten=flatten,
|
||||||
|
)
|
||||||
|
|
||||||
async def first(self, *args: Any, **kwargs: Any) -> "T":
|
async def first(self, *args: Any, **kwargs: Any) -> "T":
|
||||||
"""
|
"""
|
||||||
Gets the first row from the db ordered by primary key column ascending.
|
Gets the first row from the db ordered by primary key column ascending.
|
||||||
|
|||||||
@ -355,3 +355,87 @@ async def test_queryset_values_multiple_select_related():
|
|||||||
"posts__name": "Check this out, ormar now for free",
|
"posts__name": "Check this out, ormar now for free",
|
||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_querysetproxy_values():
|
||||||
|
async with database:
|
||||||
|
role = (
|
||||||
|
await Role.objects.select_related("users__categories")
|
||||||
|
.filter(name="admin")
|
||||||
|
.get()
|
||||||
|
)
|
||||||
|
user = await role.users.values()
|
||||||
|
assert user == [
|
||||||
|
{
|
||||||
|
"id": 1,
|
||||||
|
"name": "Anonymous",
|
||||||
|
"roles__id": 1,
|
||||||
|
"roles__name": "admin",
|
||||||
|
"roleuser__id": 1,
|
||||||
|
"roleuser__role": 1,
|
||||||
|
"roleuser__user": 1,
|
||||||
|
}
|
||||||
|
]
|
||||||
|
|
||||||
|
user = (
|
||||||
|
await role.users.filter(name="Anonymous")
|
||||||
|
.select_related("categories")
|
||||||
|
.fields({"name": ..., "categories": {"name"}})
|
||||||
|
.values(exclude_through=True)
|
||||||
|
)
|
||||||
|
assert user == [
|
||||||
|
{
|
||||||
|
"name": "Anonymous",
|
||||||
|
"roles__id": 1,
|
||||||
|
"roles__name": "admin",
|
||||||
|
"categories__name": "News",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
|
||||||
|
user = (
|
||||||
|
await role.users.filter(name="Anonymous")
|
||||||
|
.select_related("categories")
|
||||||
|
.fields({"name": ..., "categories": {"name"}})
|
||||||
|
.exclude_fields("roles")
|
||||||
|
.values(exclude_through=True)
|
||||||
|
)
|
||||||
|
assert user == [{"name": "Anonymous", "categories__name": "News"}]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_querysetproxy_values_list():
|
||||||
|
async with database:
|
||||||
|
role = (
|
||||||
|
await Role.objects.select_related("users__categories")
|
||||||
|
.filter(name="admin")
|
||||||
|
.get()
|
||||||
|
)
|
||||||
|
user = await role.users.values_list()
|
||||||
|
assert user == [(1, "Anonymous", 1, 1, 1, 1, "admin")]
|
||||||
|
|
||||||
|
user = (
|
||||||
|
await role.users.filter(name="Anonymous")
|
||||||
|
.select_related("categories")
|
||||||
|
.fields({"name": ..., "categories": {"name"}})
|
||||||
|
.values_list(exclude_through=True)
|
||||||
|
)
|
||||||
|
assert user == [("Anonymous", "News", 1, "admin")]
|
||||||
|
|
||||||
|
user = (
|
||||||
|
await role.users.filter(name="Anonymous")
|
||||||
|
.select_related("categories")
|
||||||
|
.fields({"name": ..., "categories": {"name"}})
|
||||||
|
.exclude_fields("roles")
|
||||||
|
.values_list(exclude_through=True)
|
||||||
|
)
|
||||||
|
assert user == [("Anonymous", "News")]
|
||||||
|
|
||||||
|
user = (
|
||||||
|
await role.users.filter(name="Anonymous")
|
||||||
|
.select_related("categories")
|
||||||
|
.fields({"name"})
|
||||||
|
.exclude_fields("roles")
|
||||||
|
.values_list(exclude_through=True, flatten=True)
|
||||||
|
)
|
||||||
|
assert user == ["Anonymous"]
|
||||||
|
|||||||
Reference in New Issue
Block a user