71 lines
1.7 KiB
Python
71 lines
1.7 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('/forum/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('/forum/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: str = flask_login.current_user.get_id()
|
|
|
|
# Get all forums the user has access to
|
|
user_forums: list[mgU.Forum] = database.get_user_forums(int(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
|
|
|