feat: add project support

Author: goncalorazevedo

app/app.py

@@ -11,11 +11,13 @@
 from app.extensions import migrate
 
 from app.controllers.member_controller import create_member_bp
+from app.controllers.project_controller import create_project_bp
 
 from app.repositories.member_repository import MemberRepository
+from app.repositories.project_repository import ProjectRepository
 
 
-def create_app(config_class=Config, *, member_repo=None):
+def create_app(config_class=Config, *, member_repo=None, project_repo=None, access_control=False):
     flask_app = Flask(__name__)
     flask_app.config.from_object(config_class)
 
@@ -28,6 +30,11 @@ def create_app(config_class=Config, *, member_repo=None):
     member_bp = create_member_bp(member_repo=member_repo)
     flask_app.register_blueprint(member_bp)
 
+    if project_repo is None:
+        project_repo = ProjectRepository(db=db)
+    project_bp = create_project_bp(project_repo=project_repo)
+    flask_app.register_blueprint(project_bp)
+
     from werkzeug.exceptions import HTTPException
     flask_app.register_error_handler(HTTPException, handle_http_exception)
     from pydantic import ValidationError

app/controllers/auth_controller.py

@@ -0,0 +1,175 @@
+import bcrypt
+import secrets
+import requests
+import random
+import base62
+
+from pydantic import ValidationError
+
+from urllib.parse import urlencode
+
+from flask import Blueprint, request, abort, session, current_app, redirect
+from http import HTTPStatus
+
+from app.schemas.login_schema import LoginSchema, FenixOAuthSchema, FenixStudentSchema
+from app.schemas.member_schema import MemberSchema
+
+from app.repositories.member_repository import MemberRepository
+
+
+def create_auth_bp(member_repo: MemberRepository):
+    bp = Blueprint("auth", __name__)
+
+    @bp.route("/login", methods=["POST"])
+    def login():
+        login_data = LoginSchema(**request.json())
+
+        if (member := member_repo.get_member_by_username(login_data.username)) is None:
+            abort(HTTPStatus.UNAUTHORIZED)
+
+        if member.password is None:  # fenix authenticated users must login through Fenix
+            abort(HTTPStatus.UNAUTHORIZED)
+
+        if not bcrypt.checkpw(login_data.password.encode("utf-8"), member.password.encode("utf-8")):
+            abort(HTTPStatus.UNAUTHORIZED)
+
+        session.clear()
+
+        session["username"] = member.username
+        return {"message": f"Welcome {member.username}!"}
+
+    @bp.route("/logout", methods=["GET"])
+    def logout():
+        session.clear()
+        return {"message": "Logged out sucessfully"}
+
+    @bp.route("/fenix-auth")
+    def fenix_auth():
+        if (
+            current_app.config.get("CLIENT_ID", "") == ""
+            or current_app.config.get("CLIENT_SECRET", "") == ""
+            or current_app.config.get("FENIX_REDIRECT_URL") == ""
+        ):
+            abort(HTTPStatus.NOT_IMPLEMENTED)
+
+        state = secrets.token_hex(16)
+
+        session.clear()
+        session["state"] = state
+        params = {
+            "client_id": current_app.config.get("CLIENT_ID"),
+            "redirect_uri": current_app.config.get("FENIX_REDIRECT_URL"),
+            "state": state,
+        }
+        return redirect(
+            "https://fenix.tecnico.ulisboa.pt/oauth/userdialog?" + urlencode(params)
+        )
+
+    @bp.route("/fenix-auth-callback")
+    def fenix_auth_callback():
+        try:
+            oauth_data = FenixOAuthSchema(**request.args)
+        except ValidationError:
+            # this fails on invalid state parameters
+            abort(HTTPStatus.UNAUTHORIZED)
+
+        if oauth_data.state is None or oauth_data.state != session.get("state", ""):
+            abort(HTTPStatus.UNAUTHORIZED)
+
+        if oauth_data.error is not None:
+            abort(HTTPStatus.BAD_GATEWAY, {"error": oauth_data.error})
+
+        params = {
+            "client_id": current_app.config.get("CLIENT_ID"),
+            "client_secret": current_app.config.get("CLIENT_SECRET"),
+            "redirect_uri": current_app.config.get("FENIX_REDIRECT_URL"),
+            "code": oauth_data.code,
+            "grant_type": "authorization_code",
+        }
+
+        if (
+            access_token := __fetch_access_token(
+                "https://fenix.tecnico.ulisboa.pt/oauth/access_token?"
+                + urlencode(params)
+            )
+        ) is None:
+            abort(HTTPStatus.BAD_GATEWAY)
+
+        if (user_data := __get_user_info(access_token)) is None:
+            abort(HTTPStatus.BAD_GATEWAY)
+
+        session.clear()
+        if (member := member_repo.get_member_by_ist_id(user_data.username)) is None:
+            # attempt to create a user with a random username
+            while True:
+                while True:
+                    # generate a random username
+                    username = base62.encodebytes(random.randbytes(8)).lower()
+                    if member_repo.get_member_by_username(username) is None:
+                        break
+
+                try:
+                    pass
+                    # member = member_service.create_user(Member(MemberSchema(**user_data)))
+                except Exception as e:
+                    print(e)
+                    # TODO replace with conflict exception
+                    continue
+                break
+
+        session["username"] = member.username
+        return {"message": "Registered"}
+
+    return bp
+
+
+def __fetch_access_token(url: str) -> str | None:
+    r = requests.post(url)
+    try:
+        r.raise_for_status()
+    except requests.HTTPError as e:
+        current_app.logger.info(f"Failed requesting access token: {e}")
+        return None
+
+    if r.status_code != 200:
+        current_app.logger.info(
+            f"Failed requesting access token with non 200 response. Status code: {r.status_code}\nMessage: {r.connection}"
+        )
+        return None
+
+    access_token: str
+    try:
+        access_token = r.json()["access_token"]
+    except Exception as e:
+        current_app.logger.info(f"Failed requesting access token: {e}")
+        return None
+
+    return access_token
+
+
+def __get_user_info(
+    access_token: str,
+) -> FenixStudentSchema | None:
+    r = requests.get(
+        "https://fenix.tecnico.ulisboa.pt/api/fenix/v1/person?"
+        + urlencode({"access_token": access_token})
+    )
+    try:
+        r.raise_for_status()
+    except requests.HTTPError as e:
+        current_app.logger.info(f"Failed requesting user info: {e}")
+        return None
+
+    if r.status_code != 200:
+        current_app.logger.info(
+            f"Failed requesting user info with non 200 status code. Status Code: {r.status_code}\nMessage: {r.content}"
+        )
+        return None
+
+    try:
+        user_data = FenixStudentSchema(**r.json())
+    except Exception as e:
+        current_app.logger.info(f"Failed requesting user info: {e}")
+        return None
+
+    return user_data

app/controllers/project_controller.py

@@ -0,0 +1,65 @@
+from http import HTTPStatus
+
+from flask import Blueprint
+from flask import request
+from flask import abort
+
+from app.utils import slugify
+
+from app.schemas.project_schema import ProjectSchema
+from app.schemas.update_project_schema import UpdateProjectSchema
+
+from app.repositories.project_repository import ProjectRepository
+
+from app.models.project_model import Project
+
+
+def create_project_bp(*, project_repo: ProjectRepository):
+    bp = Blueprint("projects", __name__)
+
+    @bp.route("/projects", methods=["POST"])
+    def create_project():
+        project_data = ProjectSchema(**request.json)
+        if project_repo.get_project_by_name(project_data.name) is not None:
+            return abort(HTTPStatus.CONFLICT, description=f'Project with name "{project_data.name}" already exists')
+
+        slug = slugify(project_data.name)
+        if project_repo.get_project_by_slug(slug):
+            return abort(HTTPStatus.CONFLICT,
+                         description=f'A slug already exists for this name, please pick a new one: "{project_data.name}')
+
+        project = project_repo.create_project(Project.from_schema(project_data))
+        return ProjectSchema.from_project(project).model_dump()
+
+    @bp.route("/projects", methods=["GET"])
+    def get_projects():
+        return [ProjectSchema.from_project(p).model_dump() for p in project_repo.get_projects()]
+
+    @bp.route("/projects/<slug>", methods=["GET"])
+    def get_project_by_slug(slug):
+        if (project := project_repo.get_project_by_slug(slug)) is None:
+            return abort(HTTPStatus.NOT_FOUND, description=f'Project "{slug}" not found"')
+        return ProjectSchema.from_project(project).model_dump()
+
+    @bp.route("/projects/<slug>", methods=["PUT"])
+    def update_project_by_slug(slug):
+        if (project := project_repo.get_project_by_slug(slug)) is None:
+            return abort(HTTPStatus.NOT_FOUND, description=f'Project "{slug}" not found')
+
+        project_update = UpdateProjectSchema(**request.json)
+        if project_update.name and project_repo.get_project_by_slug(slugify(project_update.name)) is not None:
+            return abort(HTTPStatus.CONFLICT,
+                         description=f'A slug already exists for this name, please pick a new one: "{project_update.name}')
+
+        updated_project = project_repo.update_project(project, project_update)
+        return ProjectSchema.from_project(updated_project).model_dump()
+
+    @bp.route("/projects/<slug>", methods=["DELETE"])
+    def delete_project_by_slug(slug):
+        if (project := project_repo.get_project_by_slug(slug)) is None:
+            return abort(HTTPStatus.NOT_FOUND, description=f'Project "{slug}" not found')
+
+        name = project_repo.delete_project(project)
+        return {f"description": "Project deleted successfully", "name": name}
+
+    return bp

app/models/project_model.py

@@ -0,0 +1,97 @@
+from typing import TYPE_CHECKING
+from sqlalchemy.orm import Mapped, mapped_column, validates
+from sqlalchemy import select, Enum
+
+from app.extensions import db
+
+from app.utils import is_valid_datestring, slugify, ProjectStateEnum
+
+if TYPE_CHECKING:
+    from app.schemas.project_schema import ProjectSchema
+
+class Project(db.Model):
+    __tablename__ = "projects"
+
+    _name: Mapped[str] = mapped_column("name", primary_key=True)
+    slug: Mapped[str] = mapped_column(unique=True)
+    state: Mapped[ProjectStateEnum] = mapped_column(Enum(ProjectStateEnum, native_enum=False))
+    start_date: Mapped[str] = mapped_column()
+
+    end_date: Mapped[str] = mapped_column(nullable=True)
+    description: Mapped[str] = mapped_column(nullable=True)
+
+    @classmethod
+    def from_schema(cls, schema: "ProjectSchema"):
+        data = schema.model_dump()
+        if "slug" in data:
+            del data["slug"]
+        return cls(**data)
+
+    def __init__(self, *, name=None, state=None, start_date=None, end_date=None, description=None):
+        self.name = name
+        self.state = state
+        self.start_date = start_date
+        self.end_date = end_date
+        self.description = description
+
+    @property
+    def name(self):
+        return self._name
+
+    @name.setter
+    def name(self, value):
+        self._name = self.validate_name("name", value)
+        self.slug = slugify(value)
+
+    @validates("name")
+    def validate_name(self, k, v):
+        if not isinstance(v, str):
+            raise ValueError(f'Invalid name type: "{type(v)}"')
+        if isinstance(v, str) and not 2 <= len(v) <= 64:
+            raise ValueError(f'Invalid name length, minimum 2 and maximum 64 characters: "{v}"')
+        return v
+
+    @validates("slug")
+    def validate_slug(self, k, v):
+        if db.session.execute(select(Project).where(Project.slug == v)).one_or_none() is not None:
+            raise ValueError(f'A slug already exists for this name, please pick a new one: "{v}"')
+        return v
+
+    @validates("state")
+    def validate_state(self, k, v):
+        if not isinstance(v, ProjectStateEnum):
+            raise ValueError(f'Invalid state type: "{type(v)}"')
+        return v
+
+    @validates("start_date")
+    def validate_start_date(self, k, v):
+        if not isinstance(v, str):
+            raise ValueError(f'Invalid start_date type: "{type(v)}"')
+        if not is_valid_datestring(v):
+            raise ValueError(f'Invalid start_date format, expected "YYYY-MM-DD": "{v}"')
+        return v
+
+    @validates("end_date")
+    def validate_end_date(self, k, v):
+        if v is None:
+            return None
+        if not isinstance(v, str):
+            raise ValueError(f'Invalid end_date type: "{type(v)}"')
+        if not is_valid_datestring(v):
+            raise ValueError(f'Invalid end_date format, expected "YYYY-MM-DD": "{v}"')
+        return v
+
+    @validates("description")
+    def validate_description(self, k, v):
+        if v is None:
+            return None
+        if not isinstance(v, str):
+            raise ValueError(f'Invalid {k} type: "{type(v)}"')
+        if len(v) > 2048:
+            raise ValueError(f'Invalid {k} length, minimum 0 and maximum 2048 characters: "{v}"')
+        return v
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__}({', '.join(f'{k}={v!r}' for k, v in self.__dict__.items())})>"
+
+