Merge pull request #21 from SELab-2/dao

Voeg DAO interfaces + implementaties toe

backend/db/errors/database_errors.py

@@ -2,3 +2,8 @@ class ItemNotFoundError(Exception):
 
     def __init__(self, message: str) -> None:
         super().__init__(message)
+
+
+class UniqueConstraintError(Exception):
+    def __init__(self, message: str) -> None:
+        super().__init__(message)

backend/db/implementation/SqlAdminDAO.py

@@ -0,0 +1,27 @@
+from sqlalchemy import select
+
+from db.errors.database_errors import ItemNotFoundError
+from db.extensions import db
+from db.interface.AdminDAO import AdminDAO
+from db.models.models import Admin
+from domain.models.AdminDataclass import AdminDataclass
+
+
+class SqlAdminDAO(AdminDAO):
+    def get_admin(self, ident: int) -> AdminDataclass:
+        admin: Admin | None = db.session.get(Admin, ident=ident)
+        if not admin:
+            msg = f"Admin with id {ident} not found"
+            raise ItemNotFoundError(msg)
+        return admin.to_domain_model()
+
+    def get_all_admins(self) -> list[AdminDataclass]:
+        admins: list[Admin] = list(db.session.scalars(select(Admin)).all())
+        return [admin.to_domain_model() for admin in admins]
+
+    def create_admin(self, name: str, email: str) -> AdminDataclass:
+        new_admin: Admin = Admin(name=name, email=email)
+        db.session.add(new_admin)
+        db.session.commit()
+        return new_admin.to_domain_model()
+

backend/db/implementation/SqlGroupDAO.py

@@ -0,0 +1,65 @@
+from db.errors.database_errors import ItemNotFoundError, UniqueConstraintError
+from db.extensions import db
+from db.interface.GroupDAO import GroupDAO
+from db.models.models import Group, Project, Student
+from domain.models.GroupDataclass import GroupDataclass
+from domain.models.StudentDataclass import StudentDataclass
+
+
+class SqlGroupDAO(GroupDAO):
+    def create_group(self, project_id: int) -> GroupDataclass:
+        project: Project | None = db.session.get(Project, ident=project_id)
+        if not project:
+            msg = f"Project with id {project} not found"
+            raise ItemNotFoundError(msg)
+        new_group: Group = Group(project_id=project_id)
+        db.session.add(new_group)
+        db.session.commit()
+        return new_group.to_domain_model()
+
+    def get_group(self, group_id: int) -> GroupDataclass:
+        group: Group | None = db.session.get(Group, ident=group_id)
+        if not group:
+            msg = f"Group with id {group_id} not found"
+            raise ItemNotFoundError(msg)
+        return group.to_domain_model()
+
+    def get_groups_of_project(self, project_id: int) -> list[GroupDataclass]:
+        project: Project | None = db.session.get(Project, ident=project_id)
+        if not project:
+            msg = f"Project with id {project} not found"
+            raise ItemNotFoundError(msg)
+        groups: list[Group] = project.groups
+        return [group.to_domain_model() for group in groups]
+
+    def get_groups_of_student(self, student_id: int) -> list[GroupDataclass]:
+        student: Student | None = db.session.get(Student, ident=student_id)
+        if not student:
+            msg = f"Student with id {student_id} not found"
+            raise ItemNotFoundError(msg)
+        groups: list[Group] = student.groups
+        return [group.to_domain_model() for group in groups]
+
+    def add_student_to_group(self, student_id: int, group_id: int) -> None:
+        student: Student | None = db.session.get(Student, ident=student_id)
+        group: Group | None = db.session.get(Group, ident=group_id)
+        if not student:
+            msg = f"Student with id {student_id} not found"
+            raise ItemNotFoundError(msg)
+        if not group:
+            msg = f"Group with id {group_id} not found"
+            raise ItemNotFoundError(msg)
+        if student in group.students:
+            msg = f"Student with id {student_id} already in group with id {group_id}"
+            raise UniqueConstraintError(msg)
+
+        group.students.append(student)
+        db.session.commit()
+
+    def get_students_of_group(self, group_id: int) -> list[StudentDataclass]:
+        group: Group | None = db.session.get(Group, ident=group_id)
+        if not group:
+            msg = f"Group with id {group_id} not found"
+            raise ItemNotFoundError(msg)
+        students: list[Student] = group.students
+        return [student.to_domain_model() for student in students]

backend/db/implementation/SqlProjectDAO.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+
+from db.errors.database_errors import ItemNotFoundError
+from db.extensions import db
+from db.interface.ProjectDAO import ProjectDAO
+from db.models.models import Project, Subject
+from domain.models.ProjectDataclass import ProjectDataclass
+
+
+class SqlProjectDAO(ProjectDAO):
+    def create_project(self, subject_id: int, name: str, deadline: datetime, archived: bool, requirements: str,
+                       visible: bool, max_students: int) -> ProjectDataclass:
+        subject: Subject | None = db.session.get(Subject, subject_id)
+        if not subject:
+            msg = f"Subject with id {subject_id} not found"
+            raise ItemNotFoundError(msg)
+
+        new_project: Project = Project(subject_id=subject_id, name=name, deadline=deadline,
+                                       archived=archived, requirements=requirements, visible=visible,
+                                       max_students=max_students)
+
+        db.session.add(new_project)
+        db.session.commit()
+        return new_project.to_domain_model()
+
+
+    def get_project(self, project_id: int) -> ProjectDataclass:
+        project: Project | None = db.session.get(Project, ident=project_id)
+        if not project:
+            msg = f"Project with id {project_id} not found"
+            raise ItemNotFoundError(msg)
+        return project.to_domain_model()
+
+    def get_projects_of_subject(self, subject_id: int) -> list[ProjectDataclass]:
+        subject: Subject | None = db.session.get(Subject, ident=subject_id)
+        if not subject:
+            msg = f"Subject with id {subject_id} not found"
+            raise ItemNotFoundError(msg)
+        projects: list[Project] = subject.projects
+        return [project.to_domain_model() for project in projects]

backend/db/implementation/SqlStudentDAO.py

@@ -0,0 +1,26 @@
+from sqlalchemy import select
+
+from db.errors.database_errors import ItemNotFoundError
+from db.extensions import db
+from db.interface.StudentDAO import StudentDAO
+from db.models.models import Student
+from domain.models.StudentDataclass import StudentDataclass
+
+
+class SqlStudentDAO(StudentDAO):
+    def get_student(self, ident: int) -> StudentDataclass:
+        student: Student | None = db.session.get(Student, ident=ident)
+        if not student:
+            msg = f"Student with id {ident} not found"
+            raise ItemNotFoundError(msg)
+        return student.to_domain_model()
+
+    def get_all_students(self) -> list[StudentDataclass]:
+        students: list[Student] = list(db.session.scalars(select(Student)).all())
+        return [student.to_domain_model() for student in students]
+
+    def create_student(self, name: str, email: str) -> StudentDataclass:
+        new_student: Student = Student(name=name, email=email)
+        db.session.add(new_student)
+        db.session.commit()
+        return new_student.to_domain_model()

backend/db/implementation/SqlSubjectDAO.py

@@ -0,0 +1,70 @@
+from db.errors.database_errors import ItemNotFoundError, UniqueConstraintError
+from db.extensions import db
+from db.interface.SubjectDAO import SubjectDAO
+from db.models.models import Student, Subject, Teacher
+from domain.models.SubjectDataclass import SubjectDataclass
+
+
+class SqlSubjectDAO(SubjectDAO):
+    def create_subject(self, name: str) -> SubjectDataclass:
+        new_subject = Subject(name=name)
+        db.session.add(new_subject)
+        db.session.commit()
+        return new_subject.to_domain_model()
+
+    def get_subject(self, subject_id: int) -> SubjectDataclass:
+        subject: Subject | None = db.session.get(Subject, ident=subject_id)
+        if not subject:
+            msg = f"Subject with id {subject_id} not found"
+            raise ItemNotFoundError(msg)
+        return subject.to_domain_model()
+
+    def get_subjects_of_teacher(self, teacher_id: int) -> list[SubjectDataclass]:
+        teacher: Teacher | None = db.session.get(Teacher, ident=teacher_id)
+        if not teacher:
+            msg = f"Teacher with id {teacher_id} not found"
+            raise ItemNotFoundError(msg)
+        subjects: list[Subject] = teacher.subjects
+        return [vak.to_domain_model() for vak in subjects]
+
+    def get_subjects_of_student(self, student_id: int) -> list[SubjectDataclass]:
+        student: Student | None = db.session.get(Student, ident=student_id)
+        if not student:
+            msg = f"Student with id {student_id} not found"
+            raise ItemNotFoundError(msg)
+        subjects: list[Subject] = student.subjects
+        return [vak.to_domain_model() for vak in subjects]
+
+    def add_student_to_subject(self, student_id: int, subject_id: int) -> None:
+        student: Student | None = db.session.get(Student, ident=student_id)
+        subject: Subject | None = db.session.get(Subject, ident=subject_id)
+
+        if not student:
+            msg = f"Student with id {student_id} not found"
+            raise ItemNotFoundError(msg)
+        if not subject:
+            msg = f"Subject with id {subject_id} not found"
+            raise ItemNotFoundError(msg)
+        if subject in student.subjects:
+            msg = f"Student with id {student_id} already has subject with id {subject_id}"
+            raise UniqueConstraintError(msg)
+
+        student.subjects.append(subject)
+        db.session.commit()
+
+    def add_teacher_to_subject(self, teacher_id: int, subject_id: int) -> None:
+        teacher: Teacher | None = db.session.get(Teacher, ident=teacher_id)
+        subject: Subject | None = db.session.get(Subject, ident=subject_id)
+
+        if not teacher:
+            msg = f"Teacher with id {teacher_id} not found"
+            raise ItemNotFoundError(msg)
+        if not subject:
+            msg = f"Subject with id {subject_id} not found"
+            raise ItemNotFoundError(msg)
+        if subject in teacher.subjects:
+            msg = f"Teacher with id {teacher_id} already has subject with id {subject_id}"
+            raise UniqueConstraintError(msg)
+
+        teacher.subjects.append(subject)
+        db.session.commit()

backend/db/implementation/SqlSubmissionDAO.py

@@ -0,0 +1,49 @@
+from datetime import datetime
+
+from db.errors.database_errors import ItemNotFoundError
+from db.extensions import db
+from db.interface.SubmissionDAO import SubmissionDAO
+from db.models.models import Group, Student, Submission
+from domain.models.SubmissionDataclass import SubmissionDataclass, SubmissionState
+
+
+class SqlSubmissionDAO(SubmissionDAO):
+    def create_submission(self, student_id: int, group_id: int, message: str,state: SubmissionState,
+                          date_time: datetime) -> SubmissionDataclass:
+
+        student: Student | None = db.session.get(Student, ident=student_id)
+        group: Group | None = db.session.get(Group, ident=group_id)
+        if not student:
+            msg = f"Student with id {student_id} not found"
+            raise ItemNotFoundError(msg)
+        if not group:
+            msg = f"Group with id {group_id} not found"
+            raise ItemNotFoundError(msg)
+        new_submission: Submission = Submission(student_id=student_id,
+                                                group_id=group_id, message=message, state=state, date_time=date_time)
+        db.session.add(new_submission)
+        db.session.commit()
+        return new_submission.to_domain_model()
+
+    def get_submission(self, submission_id: int) -> SubmissionDataclass:
+        submission: Submission | None = db.session.get(Submission, ident=submission_id)
+        if not submission:
+            msg = f"Submission with id {submission_id} not found"
+            raise ItemNotFoundError(msg)
+        return submission.to_domain_model()
+
+    def get_submissions_of_student(self, student_id: int) -> list[SubmissionDataclass]:
+        student: Student | None = db.session.get(Student, ident=student_id)
+        if not student:
+            msg = f"Student with id {student_id} not found"
+            raise ItemNotFoundError(msg)
+        submissions: list[Submission] = student.submissions
+        return [submission.to_domain_model() for submission in submissions]
+
+    def get_submissions_of_group(self, group_id: int) -> list[SubmissionDataclass]:
+        group: Group | None = db.session.get(Group, ident=group_id)
+        if not group:
+            msg = f"Group with id {group_id} not found"
+            raise ItemNotFoundError(msg)
+        submissions: list[Submission] = group.submissions
+        return [submission.to_domain_model() for submission in submissions]

backend/db/implementation/SqlLesgeverDAO.py → backend/db/implementation/SqlTeacherDAO.py

@@ -1,3 +1,5 @@
+from sqlalchemy import select
+
 from db.errors.database_errors import ItemNotFoundError
 from db.extensions import db
 from db.interface.TeacherDAO import TeacherDAO
@@ -7,22 +9,20 @@
 
 class SqlTeacherDAO(TeacherDAO):
     def get_teacher(self, ident: int) -> TeacherDataclass:
-        teacher: Teacher | None = Teacher.query.get(ident=ident)
+        teacher: Teacher | None = db.session.get(Teacher, ident)
 
         if not teacher:
-            msg = f"Teacher with id {ident}  not found"
+            msg = f"Teacher with id {ident} not found"
             raise ItemNotFoundError(msg)
 
         return teacher.to_domain_model()
 
     def get_all_teachers(self) -> list[TeacherDataclass]:
-        teachers: list[Teacher] = Teacher.query.all()
-        return [lesgever.to_domain_model() for lesgever in teachers]
-
-    def create_teacher(self, teacher: TeacherDataclass) -> None:
-        new_teacher = Teacher(name=teacher.name, email=teacher.email)
+        teachers: list[Teacher] = list(db.session.scalars(select(Teacher)).all())
+        return [teacher.to_domain_model() for teacher in teachers]
 
+    def create_teacher(self, name: str, email: str) -> TeacherDataclass:
+        new_teacher = Teacher(name=name, email=email)
         db.session.add(new_teacher)
         db.session.commit()
-
-        teacher.id = new_teacher.id
+        return new_teacher.to_domain_model()

backend/db/implementation/SqlUserDAO.py

@@ -0,0 +1,21 @@
+from sqlalchemy import select
+
+from db.errors.database_errors import ItemNotFoundError
+from db.extensions import db
+from db.interface.UserDAO import UserDAO
+from db.models.models import User
+from domain.models.UserDataclass import UserDataclass
+
+
+class SqlUserDAO(UserDAO):
+    def get_user(self, ident: int) -> UserDataclass:
+        user: User | None = db.session.get(User, ident=ident)
+        if not user:
+            msg = f"User with id {ident} not found"
+            raise ItemNotFoundError(msg)
+        return user.to_domain_model()
+
+    def get_all_users(self) -> list[UserDataclass]:
+        users: list[User] = list(db.session.scalars(select(User)).all())
+        return [user.to_domain_model() for user in users]
+