Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SQLAlchemy QUERY] operator does not exist #82

Closed
Rustam95Ur opened this issue Dec 16, 2024 · 1 comment
Closed

[SQLAlchemy QUERY] operator does not exist #82

Rustam95Ur opened this issue Dec 16, 2024 · 1 comment

Comments

@Rustam95Ur
Copy link

Rustam95Ur commented Dec 16, 2024

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

class Investment(BaseId):
    __tablename__ = "investment"

    fund_id: Mapped[int] = mapped_column(ForeignKey("fund.id"))
    fund: Mapped["Fund"] = relationship(back_populates="wallets")

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
@vsdudakov
Copy link
Owner

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

2 participants