bijna heel fastAPI geintegreerd + DAOs zouden bijna af moeten zijn

backend/db/extensions.py

@@ -1,7 +1,7 @@
 import os
 
-from fastapi_sqlalchemy import SQLAlchemy
 from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase
 
 db_host = os.getenv("DB_HOST", "localhost")
 db_port = os.getenv("DB_PORT", "5432")
@@ -12,4 +12,6 @@
 DB_URI = f"postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_database}"
 engine = create_engine(DB_URI)
 
-db = SQLAlchemy(custom_engine=engine)
+
+class Base(DeclarativeBase):
+    pass

backend/db/implementation/SqlAbstractDAO.py

@@ -0,0 +1,35 @@
+from sqlalchemy import select
+from abc import ABC
+
+from sqlalchemy.orm import Session
+
+from db.errors.database_errors import ItemNotFoundError
+from db.extensions import engine
+from typing import Generic, TypeVar, Type
+
+from db.models.models import AbstractModel
+
+T = TypeVar("T", bound=AbstractModel)
+D = TypeVar("D")
+
+
+class SqlAbstractDAO(Generic[T, D], ABC):
+
+    @staticmethod
+    def get_object(ident: int) -> D:
+        with Session(engine) as session:
+            generic_object: T | None = session.get(T, ident)
+
+            if not generic_object:
+                msg = f"object with id {ident} not found"
+                raise ItemNotFoundError(msg)
+
+            return generic_object.to_domain_model()
+
+    @staticmethod
+    def get_all() -> list[D]:
+        with Session(engine) as session:
+            generic_objects: list[T] = list(session.scalars(select(T)).all())
+            return [generic_object.to_domain_model() for generic_object in generic_objects]
+
+

backend/db/implementation/SqlAdminDAO.py

@@ -1,27 +1,17 @@
-from sqlalchemy import select
+from sqlalchemy.orm import Session
 
-from db.errors.database_errors import ItemNotFoundError
-from db.extensions import db
+from db.extensions import engine
+from db.implementation.SqlAbstractDAO import SqlAbstractDAO
 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]
+class SqlAdminDAO(AdminDAO, SqlAbstractDAO[Admin, AdminDataclass]):
 
     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()
-
+        with Session(engine) as session:
+            new_admin: Admin = Admin(name=name, email=email)
+            session.add(new_admin)
+            session.commit()
+            return new_admin.to_domain_model()

backend/db/implementation/SqlGroupDAO.py

@@ -1,65 +1,70 @@
 from db.errors.database_errors import ItemNotFoundError, UniqueConstraintError
-from db.extensions import db
+from db.extensions import engine
+from sqlalchemy.orm import Session
+
+from db.implementation.SqlAbstractDAO import SqlAbstractDAO
 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()
+class SqlGroupDAO(GroupDAO, SqlAbstractDAO[Group, GroupDataclass]):
+    @staticmethod
+    def create_group(project_id: int) -> GroupDataclass:
+        with Session(engine) as session:
+            project: Project | None = 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)
+            session.add(new_group)
+            return new_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]
+    @staticmethod
+    def get_groups_of_project(project_id: int) -> list[GroupDataclass]:
+        with Session(engine) as session:
+            project: Project | None = 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]
+    @staticmethod
+    def get_groups_of_student(student_id: int) -> list[GroupDataclass]:
+        with Session(engine) as session:
+            student: Student | None = 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)
+    @staticmethod
+    def add_student_to_group(student_id: int, group_id: int) -> None:
+        with Session(engine) as session:
+            student: Student | None = session.get(Student, ident=student_id)
+            group: Group | None = 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()
+            group.students.append(student)
+            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]
+    @staticmethod
+    def get_students_of_group(group_id: int) -> list[StudentDataclass]:
+        with Session(engine) as session:
+            group: Group | None = 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

@@ -1,40 +1,49 @@
 from datetime import datetime
 
 from db.errors.database_errors import ItemNotFoundError
-from db.extensions import db
+from sqlalchemy.orm import Session
+
+from db.extensions import engine
+from db.implementation.SqlAbstractDAO import SqlAbstractDAO
+from db.interface.AbstractDAO import D
 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,
+class SqlProjectDAO(ProjectDAO, SqlAbstractDAO[Project, ProjectDataclass]):
+
+    @staticmethod
+    def get_all() -> list[ProjectDataclass]:
+        return SqlAbstractDAO.get_all()
+
+    @staticmethod
+    def get_object(ident: int) -> ProjectDataclass:
+        return SqlAbstractDAO.get_object(ident)
+
+    @staticmethod
+    def create_project(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]
+        with Session(engine) as session:
+            subject: Subject | None = 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)
+
+            session.add(new_project)
+            session.commit()
+            return new_project.to_domain_model()
+
+    @staticmethod
+    def get_projects_of_subject(subject_id: int) -> list[ProjectDataclass]:
+        with Session(engine) as session:
+            subject: Subject | None = 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

@@ -1,26 +1,30 @@
 from sqlalchemy import select
 
 from db.errors.database_errors import ItemNotFoundError
-from db.extensions import db
+from db.extensions import engine
+from sqlalchemy.orm import Session
+
+from db.implementation.SqlAbstractDAO import SqlAbstractDAO
+from db.interface.AbstractDAO import D
 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()
+class SqlStudentDAO(StudentDAO, SqlAbstractDAO[Student, StudentDataclass]):
+
+    @staticmethod
+    def get_all() -> list[StudentDataclass]:
+        return SqlAbstractDAO.get_all()
 
-    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]
+    @staticmethod
+    def get_object(ident: int) -> StudentDataclass:
+        return SqlAbstractDAO.get_object(ident)
 
-    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()
+    @staticmethod
+    def create_student(name: str, email: str) -> StudentDataclass:
+        with Session(engine) as session:
+            new_student: Student = Student(name=name, email=email)
+            session.add(new_student)
+            session.commit()
+            return new_student.to_domain_model()