Код: Выделить всё
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)
< /code>
Я хочу написать гибридный атрибут, который позволяет мне запросить, прошел ли студент их тесты. ">stmt = select(Student).where(Student.passed)
< /code>
Я написал следующее выражение SQL. Внутренняя подразделение находит последние тесты всех предметов, которые проходит студент. Затем тесты проходят группы по студенческому, и не сбои, и подсчитываются неудачные тесты. Если есть ровно 0 неудачных тестов, студент помечен как пройден.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".completed_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
< /code>
Я могу написать запрос в sqlalchemy, как SO, но я не знаю, как реализовать это в выражение гибридного свойства. < /p>
latest_exams = (
select(
Student,
Student.idx.label("student_idx"),
Exam.idx.label("exam_idx"),
func.max(Exam.completed_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)
)
Подробнее здесь: https://stackoverflow.com/questions/794 ... m-subquery
