Change to fast api

backend/app.py

@@ -1,24 +1,12 @@
-import os
+import uvicorn
+from fastapi import FastAPI
 
-from flask import Flask
+from routes.teachers import teachers_router
 
-from db.extensions import db
-from routes.teachers import teachers_blueprint
-
-app = Flask(__name__)
+app = FastAPI()
 
 # Koppel routes uit andere modules.
-app.register_blueprint(teachers_blueprint)
-
-db_host = os.getenv("DB_HOST", "localhost")
-db_port = os.getenv("DB_PORT", "5432")
-db_user = os.getenv("DB_USERNAME", "postgres")
-db_password = os.getenv("DB_PASSWORD", "postgres")
-db_database = os.getenv("DB_DATABASE", "delphi")
-
-# Koppel postgres uri en db aan app instantie
-app.config["SQLALCHEMY_DATABASE_URI"] = f"postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_database}"
-db.init_app(app)
+app.include_router(teachers_router)
 
 if __name__ == "__main__":
-    app.run(debug=True)
+    uvicorn.run("app:app")

backend/db/extensions.py

@@ -1,9 +1,17 @@
-from flask_sqlalchemy import SQLAlchemy
+import os
+
+from sqlalchemy import create_engine
 from sqlalchemy.orm import DeclarativeBase
 
+db_host = os.getenv("DB_HOST", "localhost")
+db_port = os.getenv("DB_PORT", "5432")
+db_user = os.getenv("DB_USERNAME", "postgres")
+db_password = os.getenv("DB_PASSWORD", "postgres")
+db_database = os.getenv("DB_DATABASE", "delphi")
 
-class Base(DeclarativeBase):
-    pass
+DB_URI = f"postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_database}"
+engine = create_engine(DB_URI)
 
 
-db = SQLAlchemy(model_class=Base)
+class Base(DeclarativeBase):
+    pass

backend/db/implementation/SqlAbstractDAO.py

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

backend/db/implementation/SqlAdminDAO.py

@@ -1,27 +1,19 @@
-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(SqlAbstractDAO[Admin, AdminDataclass], AdminDAO):
+    def __init__(self) -> None:
+        self.model_class = Admin
 
     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,69 @@
+from sqlalchemy.orm import Session
+
 from db.errors.database_errors import ItemNotFoundError, UniqueConstraintError
-from db.extensions import db
+from db.extensions import engine
+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()
+class SqlGroupDAO(SqlAbstractDAO[Group, GroupDataclass], GroupDAO):
+    def __init__(self) -> None:
+        self.model_class = Group
 
-    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 create_group(self, 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)
+            session.commit()
+            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]
+        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]
+        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)
+        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]
+        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,40 @@
 from datetime import datetime
 
+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.ProjectDAO import ProjectDAO
 from db.models.models import Project, Subject
 from domain.models.ProjectDataclass import ProjectDataclass
 
 
-class SqlProjectDAO(ProjectDAO):
+class SqlProjectDAO(SqlAbstractDAO[Project, ProjectDataclass], ProjectDAO):
+    def __init__(self) -> None:
+        self.model_class = Project
+
     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()
+        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)
 
-    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()
+            session.add(new_project)
+            session.commit()
+            return new_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, 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,19 @@
-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.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]
+class SqlStudentDAO(SqlAbstractDAO[Student, StudentDataclass], StudentDAO):
+    def __init__(self) -> None:
+        self.model_class = Student
 
     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()
+        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()