You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
(sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.UndefinedFunctionError'>: operator does not exist: bigint = character varying\nHINT: No operator matches the given name and argument types. You might need to add explicit type casts.\n\n(Background on this error at: https://sqlalche.me/e/20/f405)"
this was the request: ?fund_id=17&limit=10&offset=0
Here is my solution, maybe it will be released someday
async def orm_get_list(
self,
offset: int | None = None,
limit: int | None = None,
search: str | None = None,
sort_by: str | None = None,
filters: dict | None = None,
) -> tuple[list[Any], int]:
"""This method is used to get list of orm/db model objects.
:params offset: an offset for pagination.
:params limit: a limit for pagination.
:params search: a search query.
:params sort_by: a sort by field name.
:params filters: a dict of filters.
:return: A tuple of list of objects and total count.
"""
def convert_sort_by(sort_by: str) -> str:
if sort_by.startswith("-"):
return sort_by[1:] + " desc"
return sort_by
session_maker = self.get_sessionmaker()
async with session_maker() as session:
qs = select(self.model_cls)
if filters:
q = []
for field_with_condition, value in filters.items():
field = field_with_condition[0]
condition = field_with_condition[1]
model_field = getattr(self.model_cls, field)
if isinstance(model_field.expression.type, (BIGINT, Integer)):
try:
value = int(value)
except ValueError:
raise ValueError(
f"Invalid value for field '{field}': expected an integer."
)
match condition:
case "lte":
q.append(model_field >= value)
case "gte":
q.append(model_field <= value)
case "lt":
q.append(model_field > value)
case "gt":
q.append(model_field < value)
case "exact":
q.append(model_field == value)
case "contains":
q.append(model_field.like(f"%{value}%"))
case "icontains":
q.append(model_field.ilike(f"%{value}%"))
qs = qs.filter(and_(*q))
if search and self.search_fields:
q = []
for field in self.search_fields:
q.append(getattr(self.model_cls, field).ilike(f"%{search}%"))
qs = qs.filter(or_(*q))
if sort_by:
qs = qs.order_by(text(convert_sort_by(sort_by)))
elif self.ordering:
sort_by_text = ", ".join([convert_sort_by(f) for f in self.ordering])
qs = qs.order_by(text(sort_by_text))
objs = await session.execute(select(func.count()).select_from(qs))
total = objs.scalar()
if self.list_select_related:
for field in self.list_select_related:
qs = qs.options(selectinload(getattr(self.model_cls, field)))
if offset is not None and limit is not None:
qs = qs.offset(offset)
qs = qs.limit(limit)
return await session.scalars(qs), total
The text was updated successfully, but these errors were encountered:
Error getting list of related data by foreign key
(sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.UndefinedFunctionError'>: operator does not exist: bigint = character varying\nHINT: No operator matches the given name and argument types. You might need to add explicit type casts.\n\n(Background on this error at: https://sqlalche.me/e/20/f405)"
this was the request:
?fund_id=17&limit=10&offset=0
and my model
Here is my solution, maybe it will be released someday
The text was updated successfully, but these errors were encountered: