python - SQL Alchemy hybrid property expression - select from subquery - Stack Overflow

admin管理员组

文章数量:1295845

Suppose I have the following model. We have Students 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.

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 want to write a hybrid attribute that allows me to query if a student has passed their tests.

stmt = select(Student).where(Student.passed)

I have written the following SQL expression. The inner subquery finds the latest tests of all subjects the student takes. Tests are then groups by student, & failed tests are counted. If there are exactly 0 failed tests, the student is marked as passed.

SELECT 
    anon_1.student_idx, 
    count(CASE WHEN (anon_1.exam_passed IS 0) THEN 1 END) = 0 AS student_passed 
FROM (
    SELECT 
        student.idx AS student_idx, 
        "Exam".idx AS exam_idx, 
        max("Exam"pleted_at) AS max_1, 
        "Exam".passed AS exam_passed 
    FROM student 
    JOIN "Exam" ON student.idx = "Exam".student_idx 
    GROUP BY "Exam".student_idx, "Exam".subject_idx
) AS anon_1 
GROUP BY anon_1.student_idx

I can write the query in SQLAlchemy like so, but I don't know how to implement this into a hybrid property's expression.

latest_exams = (
    select(
        Student,
        Student.idx.label("student_idx"),
        Exam.idx.label("exam_idx"),
        func.max(Exampleted_at),
    )
    .join(Student.exams)
    .group_by(Exam.student_idx, Exam.subject_idx)
)

subq = latest_exams.subquery()
stmt = (
    select(label("student_passed", func.count(case((Exam.passed == 0, 1)))))
    .select_from(subq)
    .join(Exam, Exam.idx == subq.c.exam_idx)
    .group_by(subq.c.student_idx)
)

How do I translate this query into a hybrid property?

Suppose I have the following model. We have Students 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.

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 want to write a hybrid attribute that allows me to query if a student has passed their tests.

stmt = select(Student).where(Student.passed)

I have written the following SQL expression. The inner subquery finds the latest tests of all subjects the student takes. Tests are then groups by student, & failed tests are counted. If there are exactly 0 failed tests, the student is marked as passed.

SELECT 
    anon_1.student_idx, 
    count(CASE WHEN (anon_1.exam_passed IS 0) THEN 1 END) = 0 AS student_passed 
FROM (
    SELECT 
        student.idx AS student_idx, 
        "Exam".idx AS exam_idx, 
        max("Exam"pleted_at) AS max_1, 
        "Exam".passed AS exam_passed 
    FROM student 
    JOIN "Exam" ON student.idx = "Exam".student_idx 
    GROUP BY "Exam".student_idx, "Exam".subject_idx
) AS anon_1 
GROUP BY anon_1.student_idx

I can write the query in SQLAlchemy like so, but I don't know how to implement this into a hybrid property's expression.

latest_exams = (
    select(
        Student,
        Student.idx.label("student_idx"),
        Exam.idx.label("exam_idx"),
        func.max(Exampleted_at),
    )
    .join(Student.exams)
    .group_by(Exam.student_idx, Exam.subject_idx)
)

subq = latest_exams.subquery()
stmt = (
    select(label("student_passed", func.count(case((Exam.passed == 0, 1)))))
    .select_from(subq)
    .join(Exam, Exam.idx == subq.c.exam_idx)
    .group_by(subq.c.student_idx)
)

How do I translate this query into a hybrid property?

• python
• mysql
• sqlalchemy

Share Improve this question Follow asked Feb 12 at 0:59 frimannfrimann 7155 bronze badges

Add a comment  | 

2 Answers 2

Sorted by: Reset to default 0

The following code stmt = select(Student).where(Student.passed), this means that Student.passed needs to return the WHERE clause that selects the correct students.

I rewrote my SQL query like so:

SELECT student.idx
FROM student
WHERE EXISTS (
    SELECT
        anon_1.student_idx
    FROM (
        SELECT
            student.idx AS student_idx,
            "Exam".idx AS exam_idx,
            max("Exam"pleted_at) AS max_1,
            "Exam".passed AS exam_passed
        FROM student
        JOIN "Exam" ON student.idx = "Exam".student_idx
        GROUP BY "Exam".student_idx, "Exam".subject_idx
    ) AS anon_1
    WHERE anon_1.student_idx == student.idx
    GROUP BY anon_1.student_idx
    HAVING count(CASE WHEN (anon_1.exam_passed IS 0) THEN 1 END) = 0
)

So everything under the EXISTS clauses needed to be written in Python.

@passed.inplace.expression
@classmethod
def passed(cls):
    latest_exams = (
        select(
            cls.idx.label("student_idx"),
            Exam.idx.label("exam_idx"),
            func.max(Exampleted_at),
            label('exam_passed', Exam.passed)
        )
        .join(cls.exams)
        .group_by(Exam.student_idx, Exam.subject_idx)
        .subquery()
    )

    students_passed = (
        select(func.count(case(
            (latest_exams.c.exam_passed.is_(0), 1)
        )) == 0)
        .where(Student.idx == latest_exams.c.student_idx)
        .group_by(latest_exams.c.student_idx)
        .having(func.count(case(
            (latest_exams.c.exam_passed.is_(0), 1)
        )) == 0)
     ).exists()

    return students_passed

How about like this?

from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.sql import func, select
from sqlalchemy.orm import relationship, Mapped, mapped_column
from sqlalchemy import ForeignKey, Boolean, case

class Student(Base):
    __tablename__ = "student"
    idx: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    exams: Mapped[list["Exam"]] = relationship(back_populates="student")

    @hybrid_property
    def passed(self):
        for subject in {exam.subject for exam in self.exams}:
            latest_exam = max(
                [exam for exam in self.exams if exam.subject == subject],
                key=lambda e: epleted_at,
                default=None,
            )
            if not latest_exam or not latest_exam.passed:
                return False  # If the latest exam in any subject is failed, return False
        return True  # Return True only if the latest exam in all subjects is passed

    @passed.expression
    def passed(cls):
        latest_exams_subq = (
            select(
                Exam.subject_idx,
                Exam.student_idx,
                func.max(Exampleted_at).label("latest_completed_at"),
            )
            .group_by(Exam.student_idx, Exam.subject_idx)
            .subquery()
        )

        passed_latest_exams = (
            select(func.count())
            .where(
                Exam.student_idx == cls.idx,
                Exam.subject_idx == latest_exams_subq.c.subject_idx,
                Exampleted_at == latest_exams_subq.c.latest_completed_at,
                Exam.passed.is_(True),
            )
            .scalar_subquery()
        )

        total_subjects = (
            select(func.count(func.distinct(Exam.subject_idx)))
            .where(Exam.student_idx == cls.idx)
            .scalar_subquery()
        )

        return case(
            (passed_latest_exams == total_subjects, True),
            else_=False,
        )

本文标签： pythonSQL Alchemy hybrid property expressionselect from subqueryStack Overflow