Programmiererforum

Code: Select all

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>
Ich habe eine SQL -Abfrage, die erfolgreich die neuesten Prüfungen erhält, die ein Schüler abgelegt hat. Sie können einen bestimmten Schülernamen mit einer Where 

Code: Select all

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
)
< /code>
Dies wird in SQLalchemy mögen. Ich kann überprüfen, ob die von SQL-Alchemie gerenderte SQL korrekt ist.    @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
< /code>
Ich kann mit Folgendem abfragen: < /p>
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)
< /code>
, aber die ausgeführte Anweisung gibt nur den Primärschlüssel zurück. Nach allem, was ich beurteilen kann, kann dies auf die Verwendung von Scalars