from __future__ import annotations
from typing import ClassVar
from eventsourcing.application import AggregateNotFoundError, Application
from eventsourcing.utils import get_topic
from examples.aggregate9.msgpack import MessagePackMapper
from examples.dcb_enrolment.domainmodel import Course, Student
from examples.dcb_enrolment.interface import (
CourseID,
CourseNotFoundError,
EnrolmentInterface,
StudentID,
StudentNotFoundError,
)
[docs]
class EnrolmentWithAggregates(Application[str], EnrolmentInterface):
env: ClassVar[dict[str, str]] = {
"MAPPER_TOPIC": get_topic(MessagePackMapper),
"ORIGINATOR_ID_TYPE": "text",
}
[docs]
def register_student(self, name: str, max_courses: int) -> StudentID:
student = Student(name, max_courses=max_courses)
self.save(student)
return student.id
[docs]
def register_course(self, name: str, places: int) -> CourseID:
course = Course(name, places=places)
self.save(course)
return course.id
[docs]
def join_course(self, student_id: StudentID, course_id: CourseID) -> None:
course = self.get_course(course_id)
student = self.get_student(student_id)
course.accept_student(student_id)
student.join_course(course_id)
self.save(course, student)
[docs]
def list_students_for_course(self, course_id: CourseID) -> list[str]:
course = self.get_course(course_id)
return [self.get_student(s).name for s in course.student_ids]
[docs]
def list_courses_for_student(self, student_id: StudentID) -> list[str]:
student = self.get_student(student_id)
return [self.get_course(s).name for s in student.course_ids]
[docs]
def get_student(self, student_id: StudentID) -> Student:
try:
return self.repository.get(student_id)
except AggregateNotFoundError:
raise StudentNotFoundError from None
[docs]
def get_course(self, course_id: CourseID) -> Course:
try:
return self.repository.get(course_id)
except AggregateNotFoundError:
raise CourseNotFoundError from None