121 lines
3.3 KiB
Python
121 lines
3.3 KiB
Python
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
|
|
|
|
|