最新消息:雨落星辰是一个专注网站SEO优化、网站SEO诊断、搜索引擎研究、网络营销推广、网站策划运营及站长类的自媒体原创博客

python - Subquery in hybrid property returns primary key instead of ORM object - Stack Overflow

programmeradmin0浏览0评论

I have a Student who takes Exams of a certain Subject. A subject may have multiple exams, but only the latest exams are considered. A student is considered passing if they have passed the latest exam of all their subjects.

I'm trying to write a hybrid property that returns the latest exams of all subjects the student takes.

The SQLAlchemy entities are written like so.

from sqlalchemy import select, func, ForeignKey, create_engine, Table, Column
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import Mapped, mapped_column, relationship, sessionmaker, DeclarativeBase, registry

reg = registry()

student_subjects = Table(
    "student_subjects",
    reg.metadata,
    Column('student_idx', ForeignKey('student.idx'), primary_key=True),
    Column('subject_idx', ForeignKey('subject.idx'), primary_key=True)
           )

@reg.mapped_as_dataclass
class Student:
    __tablename__ = "student"
    idx: Mapped[int] = mapped_column(primary_key=True, init=False, autoincrement=True)
    name: Mapped[str] = mapped_column()
    exams: Mapped[list["Exam"]] = relationship(back_populates="student", init=False, default_factory=list)
    subjects: Mapped[list["Subject"]] = relationship(back_populates="students", init=False, default_factory=list, secondary=student_subjects)

    @hybrid_property
    def latest_exams(self):
        ret = []

        for subject in self.subjects:
            exams = [exam for exam in self.exams if exam.subject == subject]
            exams.sort(key=lambda x: xpleted_at, reverse=True)

            if len(exams) > 0:
                ret.append(exams[0])

        return ret


@reg.mapped_as_dataclass
class Subject:
    __tablename__ = "subject"
    idx: Mapped[int] = mapped_column(primary_key=True, init=False, autoincrement=True)
    name: Mapped[str] = mapped_column()
    exams: Mapped[list["Exam"]] = relationship(back_populates="subject", init=False)
    students: Mapped[list["Student"]] = relationship(back_populates="subjects", init=False, secondary=student_subjects)


@reg.mapped_as_dataclass
class Exam:
    __tablename__ = "Exam"

    idx: Mapped[int] = mapped_column( primary_key=True, init=False, autoincrement=True)
    passed: Mapped[bool] = mapped_column()

    subject: Mapped["Subject"] = relationship(back_populates="exams")
    subject_idx: Mapped[int] = mapped_column(ForeignKey("subject.idx"), init=False)

    student: Mapped["Student"] = relationship(back_populates="exams")
    student_idx: Mapped[int] = mapped_column(ForeignKey("student.idx"), init=False)

    completed_at: Mapped[datetime] = mapped_column(default_factory=datetime.now)

I have an SQL query that successfully gets the latest exams that a student has taken. You can filter for a specific student name with a WHERE clause.

SELECT student_name,
       subject_name,
       exam_passed,
FROM (
    SELECT student.name student_name,
           subject.name subject_name,
           exam.passed exam_passed,
           max(exampleted_at) completed_at,
    FROM exam
    JOIN student ON exam.student_idx = student.idx
    JOIN subject ON exam.subject_idx = subject.idx
    GROUP BY exam.student_idx, exam.subject_idx
)

This is implemented in SQLAlchemy like so. I can verify that the SQL rendered by SQL alchemy is correct.

    @classmethod
    @latest_exams.inplace.expression
    def latest_exams(cls):
        stmt = (
            select(Exam, func.max(Exampleted_at))
            .join(Student)
            .group_by(Exam.student_idx, Exam.subject_idx)
        )
        subq = stmt.subquery()
        return subq

I can query using the following:

stmt = select(Student.latest_exams).join(Student).filter_by(name='Joe Bloggs')

with _Session() as s:
    results = s.execute(stmt)

for result in results.scalars():
    print(result)

But the executed statement only returns the primary key. From what I can tell, this may be due to using scalars, but in other cases scalars returns the ORM object. Why is it not doing so here?

I have a Student who takes Exams of a certain Subject. A subject may have multiple exams, but only the latest exams are considered. A student is considered passing if they have passed the latest exam of all their subjects.

I'm trying to write a hybrid property that returns the latest exams of all subjects the student takes.

The SQLAlchemy entities are written like so.

from sqlalchemy import select, func, ForeignKey, create_engine, Table, Column
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import Mapped, mapped_column, relationship, sessionmaker, DeclarativeBase, registry

reg = registry()

student_subjects = Table(
    "student_subjects",
    reg.metadata,
    Column('student_idx', ForeignKey('student.idx'), primary_key=True),
    Column('subject_idx', ForeignKey('subject.idx'), primary_key=True)
           )

@reg.mapped_as_dataclass
class Student:
    __tablename__ = "student"
    idx: Mapped[int] = mapped_column(primary_key=True, init=False, autoincrement=True)
    name: Mapped[str] = mapped_column()
    exams: Mapped[list["Exam"]] = relationship(back_populates="student", init=False, default_factory=list)
    subjects: Mapped[list["Subject"]] = relationship(back_populates="students", init=False, default_factory=list, secondary=student_subjects)

    @hybrid_property
    def latest_exams(self):
        ret = []

        for subject in self.subjects:
            exams = [exam for exam in self.exams if exam.subject == subject]
            exams.sort(key=lambda x: x.completed_at, reverse=True)

            if len(exams) > 0:
                ret.append(exams[0])

        return ret


@reg.mapped_as_dataclass
class Subject:
    __tablename__ = "subject"
    idx: Mapped[int] = mapped_column(primary_key=True, init=False, autoincrement=True)
    name: Mapped[str] = mapped_column()
    exams: Mapped[list["Exam"]] = relationship(back_populates="subject", init=False)
    students: Mapped[list["Student"]] = relationship(back_populates="subjects", init=False, secondary=student_subjects)


@reg.mapped_as_dataclass
class Exam:
    __tablename__ = "Exam"

    idx: Mapped[int] = mapped_column( primary_key=True, init=False, autoincrement=True)
    passed: Mapped[bool] = mapped_column()

    subject: Mapped["Subject"] = relationship(back_populates="exams")
    subject_idx: Mapped[int] = mapped_column(ForeignKey("subject.idx"), init=False)

    student: Mapped["Student"] = relationship(back_populates="exams")
    student_idx: Mapped[int] = mapped_column(ForeignKey("student.idx"), init=False)

    completed_at: Mapped[datetime] = mapped_column(default_factory=datetime.now)

I have an SQL query that successfully gets the latest exams that a student has taken. You can filter for a specific student name with a WHERE clause.

SELECT student_name,
       subject_name,
       exam_passed,
FROM (
    SELECT student.name student_name,
           subject.name subject_name,
           exam.passed exam_passed,
           max(exam.completed_at) completed_at,
    FROM exam
    JOIN student ON exam.student_idx = student.idx
    JOIN subject ON exam.subject_idx = subject.idx
    GROUP BY exam.student_idx, exam.subject_idx
)

This is implemented in SQLAlchemy like so. I can verify that the SQL rendered by SQL alchemy is correct.

    @classmethod
    @latest_exams.inplace.expression
    def latest_exams(cls):
        stmt = (
            select(Exam, func.max(Exam.completed_at))
            .join(Student)
            .group_by(Exam.student_idx, Exam.subject_idx)
        )
        subq = stmt.subquery()
        return subq

I can query using the following:

stmt = select(Student.latest_exams).join(Student).filter_by(name='Joe Bloggs')

with _Session() as s:
    results = s.execute(stmt)

for result in results.scalars():
    print(result)

But the executed statement only returns the primary key. From what I can tell, this may be due to using scalars, but in other cases scalars returns the ORM object. Why is it not doing so here?

Share Improve this question edited 6 hours ago Barmar 781k56 gold badges545 silver badges659 bronze badges asked 16 hours ago frimannfrimann 635 bronze badges
Add a comment  | 

1 Answer 1

Reset to default 0

The solution I found was to use an alias & subquery.

with _Session() as s:
    stmt = select(Student.latest_exams).join(Student).where(Student.name=='Joe Bloggs')
    exam_alias = aliased(Exam, stmt.subquery())
    results = s.execute(select(exam_alias))
发布评论

评论列表(0)

  1. 暂无评论