0

The crux of the problem is that I only need to retrieve 3 entries from each QuizTopic. I've been looking for a solution for a long time, but I can't find one..

models.py

class Base(DeclarativeBase):
    pass

class QuizTopic(Base):
    __tablename__ = "quiz_topic"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(60))

    quizzes_info: Mapped[list["Quiz"]] = relationship(
        back_populates="topic_info",
        cascade="all, delete",
        passive_deletes=True,
    )


class Quiz(Base):
    __tablename__ = "quiz"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    topic_id: Mapped[int] = mapped_column(Integer, ForeignKey(
        "quiz_topic.id", ondelete="CASCADE", onupdate="CASCADE"
    ))
    logo_url: Mapped[str] = mapped_column(String, nullable=True)
    title: Mapped[str] = mapped_column(String(100))
    meta: Mapped[str] = mapped_column(String(80))
    description: Mapped[str] = mapped_column(String(255))

    topic_info: Mapped[QuizTopic] = relationship(
        back_populates="quizzes_info"
    )

Right now I am getting a complete list of records. Here's the code:

orm.py

async def get_content_topics(session: AsyncSession, limit: int, continue_after: int):
    smt = (
        select(QuizTopic)
        .order_by(QuizTopic.id)
        .offset(continue_after).limit(limit)
        .options(
            load_only(QuizTopic.id, QuizTopic.title),
            joinedload(QuizTopic.quizzes_info).load_only(
                Quiz.id, Quiz.logo_url, Quiz.title, Quiz.meta
            )
        )
    )
    result = await session.scalars(smt)
    return result.unique().all()

Example output:

enter image description here

TheJecksMan
  • 15
  • 6

1 Answers1

0

looks like I did it (Probably not optimally). Here's my solution:

async def get_topics(session: AsyncSession, limit: int, continue_after: int):
    subquery = (
        select(
            QuizTopic.id,
            Quiz.id.label("quiz_id"),
            func.row_number().over(partition_by=QuizTopic.id, order_by=Quiz.id).label("row_num")
        )
        .join(Quiz, Quiz.topic_id == QuizTopic.id)
        .subquery()
    )

    smt = (
        select(QuizTopic)
        .join(QuizTopic.quizzes_info)
        .join(subquery, subquery.c.quiz_id == Quiz.id)
        .where(subquery.c.row_num <= 3)
        .order_by(QuizTopic.id)
        .offset(continue_after).limit(limit)
        .options(
            load_only(QuizTopic.id, QuizTopic.title),
            contains_eager(
                QuizTopic.quizzes_info
            ).load_only(Quiz.id, Quiz.logo_url, Quiz.title, Quiz.meta)
        )
    )
    result = await session.scalars(smt)
    return result.unique().all()
TheJecksMan
  • 15
  • 6