Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/continuous-integration.yml
Original file line number Diff line number Diff line change
Expand Up @@ -75,5 +75,8 @@ jobs:
- name: Run tests
run: poetry run pytest

- name: Run type check
run: poetry run pyright

- name: Run lint
run: poetry run ruff check .
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -17,3 +17,6 @@ __pycache__/
.vscode/
.idea/
.DS_Store

# 📦 Poetry (config locale)
poetry.toml
20 changes: 17 additions & 3 deletions app/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,29 @@
from app.controllers.article_controller import article_bp
from app.controllers.comment_controller import comment_bp
from app.controllers.login_controller import login_bp
from configurations.configuration_variables import env_vars
from config.configuration_variables import env_vars
from database.database_setup import db_session


def shutdown_session(exception=None):
def shutdown_session(exception: BaseException | None = None) -> None:
"""
Removes the database session at the end of the request.

Args:
exception (BaseException | None): The exception that triggered the teardown, if any.
"""

db_session.remove()


def initialize_flask_application():
def initialize_flask_application() -> Flask:
"""
Initializes and configures the Flask application.
Sets the secret key based on the environment and registers blueprints.

Returns:
Flask: The configured Flask application instance.
"""
app = Flask(__name__)
if os.getenv("PYTEST_CURRENT_TEST") or "pytest" in sys.modules:
app.secret_key = env_vars.test_secret_key
Expand Down
26 changes: 26 additions & 0 deletions app/constants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
from enum import Enum


class Role(str, Enum):
"""
Enum representing user roles within the application.
"""
ADMIN = "admin"
AUTHOR = "author"
USER = "user"


class SessionKey(str, Enum):
"""
Enum representing keys used in the Flask session.
"""
USER_ID = "user_id"
ROLE = "role"
USERNAME = "username"


class PaginationConfig:
"""
Configuration for pagination settings.
"""
ARTICLES_PER_PAGE = 10
144 changes: 117 additions & 27 deletions app/controllers/article_controller.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,18 @@
import math

from flask import Blueprint, flash, redirect, render_template, request, session, url_for
from flask import (
Blueprint,
flash,
redirect,
render_template,
request,
session,
url_for,
)
from werkzeug.wrappers import Response

from app.constants import PaginationConfig, Role, SessionKey
from app.controllers.decorators import roles_accepted
from app.services.article_service import ArticleService
from app.services.comment_service import CommentService
from database.database_setup import db_session
Expand All @@ -10,68 +21,147 @@


@article_bp.route("/")
def list_articles():
current_page_number = 1
page_number = request.args.get("page", current_page_number, type=int)
articles_per_page = 10
articles = ArticleService.get_paginated_articles(page_number, articles_per_page)
total_articles = ArticleService.get_total_count()
def list_articles() -> str:
"""
Renders the homepage with a paginated list of articles.

Returns:
str: The rendered HTML template for the homepage.
"""
page_number = request.args.get("page", 1, type=int)
articles_per_page = PaginationConfig.ARTICLES_PER_PAGE

article_service = ArticleService(db_session)
articles = article_service.get_paginated_articles(page_number, articles_per_page)
total_articles = article_service.get_total_count()
total_pages = math.ceil(total_articles / articles_per_page)
return render_template("index.html", articles=articles, page_number=page_number, total_pages=total_pages)

return render_template(
"index.html",
articles=articles,
page_number=page_number,
total_pages=total_pages
)


@article_bp.route("/article/<int:article_id>")
def view_article(article_id):
article = ArticleService.get_by_id(article_id)
def view_article(article_id: int) -> str | Response:
"""
Displays the details of a specific article and its comments.

Args:
article_id (int): ID of the article to view.

Returns:
str | Response: The rendered HTML template for the article or a redirect if the article is not found.
"""
article_service = ArticleService(db_session)
article = article_service.get_by_id(article_id)
if not article:
flash("Article not found.")
return redirect(url_for("article.list_articles"))

comments = CommentService.get_tree_by_article_id(article_id)
comment_service = CommentService(db_session)
comments = comment_service.get_tree_by_article_id(article_id)
return render_template("article_detail.html", article=article, comments=comments)


@article_bp.route("/article/new", methods=["GET", "POST"])
def create_article():
if session.get("role") not in ["admin", "author"]:
flash("Access restricted.")
return redirect(url_for("article.list_articles"))
@roles_accepted(Role.ADMIN, Role.AUTHOR)
def create_article() -> str | Response:
"""
Handles the creation of a new blog article.
Restricted to 'admin' and 'author' roles.

Returns:
str | Response: The rendered HTML form (GET) or a redirect to the article list after creation (POST).
"""
if request.method == "POST":
ArticleService.create_article(request.form.get("title"), request.form.get("content"), session["user_id"])
article_service = ArticleService(db_session)
title = str(request.form.get("title") or "")
content = str(request.form.get("content") or "")
user_id = int(session.get(SessionKey.USER_ID) or 0)

article_service.create_article(
title=title,
content=content,
author_id=user_id
)
db_session.commit()
flash("Article published!")
return redirect(url_for("article.list_articles"))

return render_template("article_form.html", article=None)


@article_bp.route("/article/<int:article_id>/edit", methods=["GET", "POST"])
def edit_article(article_id):
@roles_accepted(Role.ADMIN, Role.AUTHOR, Role.USER)
def edit_article(article_id: int) -> str | Response:
"""
Handles the editing of an existing article.
Ensures the user is authorized to perform the update.

Args:
article_id (int): ID of the article to edit.

Returns:
str | Response: The rendered HTML form (GET) or a redirect to the updated article (POST).
"""
article_service = ArticleService(db_session)

if request.method == "POST":
article = ArticleService.update_article(
article_id,
session.get("user_id"),
session.get("role"),
request.form.get("title"),
request.form.get("content")
user_id = int(session.get(SessionKey.USER_ID) or 0)
role = str(session.get(SessionKey.ROLE) or "")
title = str(request.form.get("title") or "")
content = str(request.form.get("content") or "")

article = article_service.update_article(
article_id=article_id,
user_id=user_id,
role=role,
title=title,
content=content
)
if article:
db_session.commit()
flash("Article updated!")
return redirect(url_for("article.view_article", article_id=article_id))

flash("Update failed: Unauthorized or not found.")
return redirect(url_for("article.list_articles"))
article = ArticleService.get_by_id(article_id)

article = article_service.get_by_id(article_id)
if not article:
flash("Article not found.")
return redirect(url_for("article.list_articles"))

return render_template("article_form.html", article=article)


@article_bp.route("/article/<int:article_id>/delete")
def delete_article(article_id):
if ArticleService.delete_article(article_id, session.get("user_id"), session.get("role")):
@roles_accepted(Role.ADMIN, Role.AUTHOR, Role.USER)
def delete_article(article_id: int) -> Response:
"""
Handles the deletion of an article.

Args:
article_id (int): ID of the article to delete.

Returns:
Response: A redirect to the article list after deletion.
"""
article_service = ArticleService(db_session)
user_id = int(session.get(SessionKey.USER_ID) or 0)
role = str(session.get(SessionKey.ROLE) or "")

if article_service.delete_article(
article_id=article_id,
user_id=user_id,
role=role
):
db_session.commit()
flash("Article deleted.")
else:
flash("Delete failed.")
flash("Delete failed: Unauthorized or not found.")

return redirect(url_for("article.list_articles"))
73 changes: 59 additions & 14 deletions app/controllers/comment_controller.py
Original file line number Diff line number Diff line change
@@ -1,33 +1,62 @@
from flask import Blueprint, flash, redirect, request, session, url_for
from werkzeug.wrappers import Response

from app.constants import Role, SessionKey
from app.controllers.decorators import login_required, roles_accepted
from app.services.comment_service import CommentService
from database.database_setup import db_session

comment_bp = Blueprint("comment", __name__, url_prefix="/comments")


@comment_bp.route("/create/<int:article_id>", methods=["POST"])
def create_comment(article_id):
# Exception to add here
if not session.get("user_id"):
flash("Login required.")
return redirect(url_for("login.render_login_page"))
if CommentService.create_comment(article_id, session["user_id"], request.form.get("content")):
@login_required
def create_comment(article_id: int) -> Response:
"""
Handles the creation of a new comment on an article.
Requires the user to be logged in.

Args:
article_id (int): ID of the article being commented on.

Returns:
Response: A redirect to the article view or login page.
"""
comment_service = CommentService(db_session)
content = str(request.form.get("content") or "")
if comment_service.create_comment(
article_id=article_id,
user_id=session[SessionKey.USER_ID],
content=content
):
db_session.commit()
flash("Comment added.")
else:
flash("Error adding comment.")

return redirect(url_for("article.view_article", article_id=article_id))


@comment_bp.route("/reply/<int:parent_comment_id>", methods=["POST"])
def reply_to_comment(parent_comment_id):
# Exception to add here
if not session.get("user_id"):
flash("Login required.")
return redirect(url_for("login.render_login_page"))
@login_required
def reply_to_comment(parent_comment_id: int) -> Response:
"""
Handles the creation of a reply to an existing comment.
Requires the user to be logged in.

article_id = CommentService.create_reply(parent_comment_id, session["user_id"], request.form.get("content"))
Args:
parent_comment_id (int): ID of the comment being replied to.

Returns:
Response: A redirect to the article view or the article list in case of error.
"""
comment_service = CommentService(db_session)
content = str(request.form.get("content") or "")
article_id = comment_service.create_reply(
parent_comment_id=parent_comment_id,
user_id=session[SessionKey.USER_ID],
content=content
)
if article_id:
db_session.commit()
return redirect(url_for("article.view_article", article_id=article_id))
Expand All @@ -37,8 +66,24 @@ def reply_to_comment(parent_comment_id):


@comment_bp.route("/delete/<int:comment_id>")
def delete_comment(comment_id):
article_id = CommentService.delete_comment(comment_id, session.get("role"))
@roles_accepted(Role.ADMIN)
def delete_comment(comment_id: int) -> Response:
"""
Handles the deletion of a comment.
Restricted to users with the 'admin' role.

Args:
comment_id (int): ID of the comment to delete.

Returns:
Response: A redirect to the article view or article list after deletion.
"""
comment_service = CommentService(db_session)
role = str(session.get(SessionKey.ROLE) or "")
article_id = comment_service.delete_comment(
comment_id=comment_id,
role=role
)
if article_id:
db_session.commit()
flash("Comment deleted.")
Expand Down
Loading