import time import dataclasses import flask import flask_login from . import helpers import database import mgU forum = flask.Blueprint( name='forum_api', import_name=__name__, url_prefix='/forum/' ) @forum.route('/send_message', methods=['POST']) @flask_login.login_required def send_message() -> tuple[str, int]: data = flask.request.json if not data: return '', 400 if any([not p in data for p in ['content', 'thread']]): return '', 400 new_message = database.MessageDTO( user_id=flask_login.current_user.get_id(), content=data['content'], thread_id=data['thread'], time=int(time.time()) ) if not helpers.user_in_thread(flask_login.current_user.get_id(), data['thread']): return '', 401 database.insert_message(new_message) return '', 201 @forum.route('/get_threads') @flask_login.login_required def get_forum_threads() -> tuple[flask.Response, int]: """ Get all threads from forums the current user has access to. """ user_id: int = flask_login.current_user.get_id() # Get all forums the user has access to user_forums: list[mgU.Forum] = database.get_user_forums(user_id) if not user_forums: return flask.jsonify({"threads": []}), 200 # Get all threads from all user's forums all_threads: list[mgU.Thread] = [] for forum in user_forums: if forum.id: # Make sure forum has an ID forum_threads = database.get_forum_threads(forum.id) all_threads.extend(forum_threads) # Convert threads to dictionary format threads_dict = {"threads": [dataclasses.asdict(thread) for thread in all_threads]} return flask.jsonify(threads_dict), 200 @forum.route('/get_messages') @flask_login.login_required def get_thread_messages() -> tuple[flask.Response, int]: thread_id = flask.request.args.get("thread", type=int) if thread_id is None: return flask.jsonify({"error": "Missing thread parameter"}), 400 messages = database.get_messages_by_thread(thread_id) messages_dicts = {"messages" : [dataclasses.asdict(msg) for msg in messages]} return flask.jsonify(messages_dicts), 200 @forum.route('/create_thread', methods=['POST']) @flask_login.login_required def create_forum_thread() -> tuple[flask.Response, int] | tuple[str, int]: data = flask.request.json if not data: return '', 400 if any([not key in data for key in ['forum_id', 'name']]): return '', 400 if type(data['forum_id']) != int: return '', 400 forum_id: int = data.get('forum_id') thread_name: str = data.get('name') if not thread_name or not thread_name.strip(): return flask.jsonify({"error": "Thread name cannot be empty"}), 400 user_id: int = flask_login.current_user.get_id() user_owns_forum: bool = helpers.user_owns_forum(user_id, forum_id) if not user_owns_forum: return flask.jsonify({"error": "You don't have permission to create threads in this forum"}), 403 # Create the thread new_thread = database.create_thread(thread_name.strip(), forum_id) if new_thread is None: return flask.jsonify({"error": "Failed to create thread"}), 500 # Return the created thread thread_dict = dataclasses.asdict(new_thread) return flask.jsonify({"thread": thread_dict, "message": "Thread created successfully"}), 201