refactor: migrate web framework from Flask to FastAPI

app.py

@@ -1,52 +1,73 @@
 import os
 import logging
-from flask import Flask
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from fastapi.exceptions import RequestValidationError
 from model import *
 from mongo import *
 
 
-def app():
-    # Create a flask app
-    app = Flask(__name__)
-    app.config['PREFERRED_URL_SCHEME'] = 'https'
-    app.url_map.strict_slashes = False
-    setup_smtp(app)
-    # Register flask blueprint
-    api2prefix = [
-        (auth_api, '/auth'),
-        (profile_api, '/profile'),
-        (problem_api, '/problem'),
-        (submission_api, '/submission'),
-        (course_api, '/course'),
-        (homework_api, '/homework'),
-        (test_api, '/test'),
-        (ann_api, '/ann'),
-        (ranking_api, '/ranking'),
-        (post_api, '/post'),
-        (copycat_api, '/copycat'),
-        (health_api, '/health'),
-        (user_api, '/user'),
-    ]
-    for api, prefix in api2prefix:
-        app.register_blueprint(api, url_prefix=prefix)
+def create_app() -> FastAPI:
+    app = FastAPI()
 
+    from model.utils.response import NOJException
+
+    @app.exception_handler(NOJException)
+    async def noj_exception_handler(request: Request, exc: NOJException):
+        return JSONResponse(
+            {
+                'status': 'err',
+                'message': exc.message,
+                'data': exc.data,
+            },
+            status_code=exc.status_code,
+        )
+
+    @app.exception_handler(RequestValidationError)
+    async def validation_exception_handler(request: Request,
+                                           exc: RequestValidationError):
+        return JSONResponse(
+            {
+                'status': 'err',
+                'message': 'Invalid request body',
+                'data': None,
+            },
+            status_code=400,
+        )
+
+    app.include_router(auth_router, prefix='/auth')
+    app.include_router(profile_router, prefix='/profile')
+    app.include_router(problem_router, prefix='/problem')
+    app.include_router(submission_router, prefix='/submission')
+    app.include_router(course_router, prefix='/course')
+    app.include_router(homework_router, prefix='/homework')
+    app.include_router(test_router, prefix='/test')
+    app.include_router(ann_router, prefix='/ann')
+    app.include_router(ranking_router, prefix='/ranking')
+    app.include_router(post_router, prefix='/post')
+    app.include_router(copycat_router, prefix='/copycat')
+    app.include_router(health_router, prefix='/health')
+    app.include_router(user_router, prefix='/user')
+    app.include_router(user_options_router, prefix='/user')
+
+    _seed_db()
+    setup_smtp()
+
+    return app
+
+
+def _seed_db():
     if not User('first_admin'):
         ADMIN = {
             'username': 'first_admin',
             'password': 'firstpasswordforadmin',
-            'email': 'i.am.first.admin@noj.tw'
+            'email': 'i.am.first.admin@noj.tw',
         }
         PROFILE = {
             'displayed_name': 'the first admin',
-            'bio': 'I am super good!!!!!'
+            'bio': 'I am super good!!!!!',
         }
         admin = User.signup(**ADMIN)
-        # TODO: use a single method to active.
-        #       we won't call `activate` here because it required the
-        #       course 'Public' should exist, but create a course
-        #       also need a teacher.
-        #       but at least make it can work now...
-        # admin.activate(PROFILE)
         admin.update(
             active=True,
             role=0,
@@ -55,29 +76,18 @@ def app():
     if not Course('Public'):
         Course.add_course('Public', 'first_admin')
 
-    if __name__ != '__main__':
-        logger = logging.getLogger('gunicorn.error')
-        app.logger.handlers = logger.handlers
-        app.logger.setLevel(logger.level)
-
-    return app
 
-
-def setup_smtp(app: Flask):
-    logger = logging.getLogger('gunicorn.error')
+def setup_smtp():
+    logger = logging.getLogger(__name__)
     if os.getenv('SMTP_SERVER') is None:
         logger.info(
-            '\'SMTP_SERVER\' is not set. email-related function will be disabled'
+            "'SMTP_SERVER' is not set. email-related function will be disabled"
         )
         return
     if os.getenv('SMTP_NOREPLY') is None:
-        raise RuntimeError('missing required configuration \'SMTP_NOREPLY\'')
+        raise RuntimeError("missing required configuration 'SMTP_NOREPLY'")
     if os.getenv('SMTP_NOREPLY_PASSWORD') is None:
-        logger.info('\'SMTP_NOREPLY\' set but \'SMTP_NOREPLY_PASSWORD\' not')
-    # config for external URLs
-    server_name = os.getenv('SERVER_NAME')
-    if server_name is None:
-        raise RuntimeError('missing required configuration \'SERVER_NAME\'')
-    app.config['SERVER_NAME'] = server_name
-    if (application_root := os.getenv('APPLICATION_ROOT')) is not None:
-        app.config['APPLICATION_ROOT'] = application_root
+        logger.info("'SMTP_NOREPLY' set but 'SMTP_NOREPLY_PASSWORD' not")
+
+
+app = create_app()

model/announcement.py

@@ -1,65 +1,71 @@
 from datetime import datetime
-from flask import Blueprint
+from typing import Optional
+from fastapi import APIRouter, Depends
 
 from mongo import *
 from mongo.utils import *
-from .auth import *
+from .auth import login_required
 from .utils import *
 from .schemas import CreateAnnouncementBody, UpdateAnnouncementBody, DeleteAnnouncementBody
-from .course import *
+from .course import course_router
 
-__all__ = ['ann_api']
+__all__ = ['ann_router']
 
-ann_api = Blueprint('ann_api', __name__)
+ann_router = APIRouter()
 
 
-@ann_api.route('/', methods=['GET'])
-@ann_api.route('/<ann_id>', methods=['GET'])
-def get_sys_ann(ann_id=None):
-    public_name = Course.get_public().course_name
-    anns = Announcement.ann_list(None, public_name)
-    data = [{
+def _format_ann(an):
+    return {
         'annId': str(an.id),
         'title': an.title,
         'createTime': int(an.create_time.timestamp()),
         'updateTime': int(an.update_time.timestamp()),
         'creator': an.creator.info,
         'updater': an.updater.info,
         'markdown': an.markdown,
-        'pinned': an.pinned
-    } for an in anns if ann_id == None or str(an.id) == ann_id]
+        'pinned': an.pinned,
+    }
+
+
+@ann_router.get('')
+@ann_router.get('/{ann_id}')
+def get_sys_ann(ann_id: Optional[str] = None):
+    public_name = Course.get_public().course_name
+    anns = Announcement.ann_list(None, public_name)
+    data = [
+        _format_ann(an) for an in anns
+        if ann_id is None or str(an.id) == ann_id
+    ]
     return HTTPResponse('Sys Ann bro', data=data)
 
 
-@course_api.get('/<course_name>/ann')
-@ann_api.get('/<course_name>/<ann_id>')
-@login_required
-def get_announcements(user, course_name=None, ann_id=None):
-    # Get an announcement list
+@course_router.get('/{course_name}/ann')
+def get_course_announcements(course_name: str, user=Depends(login_required)):
     try:
-        anns = Announcement.ann_list(user.obj, course_name or 'Public')
+        anns = Announcement.ann_list(user.obj, course_name)
     except (DoesNotExist, ValidationError):
         return HTTPError('Cannot Access a Announcement', 403)
     if anns is None:
         return HTTPError('Announcement Not Found', 404)
-    data = [{
-        'annId': str(an.id),
-        'title': an.title,
-        'createTime': int(an.create_time.timestamp()),
-        'updateTime': int(an.update_time.timestamp()),
-        'creator': an.creator.info,
-        'updater': an.updater.info,
-        'markdown': an.markdown,
-        'pinned': an.pinned
-    } for an in anns if ann_id is None or str(an.id) == ann_id]
+    data = [_format_ann(an) for an in anns]
     return HTTPResponse('Announcement List', data=data)
 
 
-@ann_api.post('/')
-@login_required
-@parse_body(CreateAnnouncementBody)
-def create_announcement(user, body: CreateAnnouncementBody):
-    # Create a new announcement
+@ann_router.get('/{course_name}/{ann_id}')
+def get_ann_by_id(course_name: str, ann_id: str, user=Depends(login_required)):
+    try:
+        anns = Announcement.ann_list(user.obj, course_name)
+    except (DoesNotExist, ValidationError):
+        return HTTPError('Cannot Access a Announcement', 403)
+    if anns is None:
+        return HTTPError('Announcement Not Found', 404)
+    data = [_format_ann(an) for an in anns if str(an.id) == ann_id]
+    return HTTPResponse('Announcement List', data=data)
+
+
+@ann_router.post('')
+def create_announcement(body: CreateAnnouncementBody,
+                        user=Depends(login_required)):
     try:
         ann = Announcement.new_ann(
             title=body.title,
@@ -76,20 +82,17 @@ def create_announcement(user, body: CreateAnnouncementBody):
         return HTTPError('Failed to Create Announcement', 403)
     data = {
         'annId': str(ann.id),
-        'createTime': int(ann.create_time.timestamp())
+        'createTime': int(ann.create_time.timestamp()),
     }
     return HTTPResponse('Announcement Created', data=data)
 
 
-@ann_api.put('/')
-@login_required
-@parse_body(UpdateAnnouncementBody)
-def update_announcement(user, body: UpdateAnnouncementBody):
-    # Update an announcement
+@ann_router.put('')
+def update_announcement(body: UpdateAnnouncementBody,
+                        user=Depends(login_required)):
     ann = Announcement(body.ann_id)
     if not ann:
         return HTTPError('Announcement Not Found', 404)
-
     course = Course(ann.course)
     if not course.permission(user, Course.Permission.GRADE):
         return HTTPError('Failed to Update Announcement', 403)
@@ -102,23 +105,18 @@ def update_announcement(user, body: UpdateAnnouncementBody):
             pinned=body.pinned,
         )
     except ValidationError as ve:
-        return HTTPError(
-            'Failed to Update Announcement',
-            400,
-            data=ve.to_dict(),
-        )
+        return HTTPError('Failed to Update Announcement',
+                         400,
+                         data=ve.to_dict())
     return HTTPResponse('Updated')
 
 
-@ann_api.delete('/')
-@login_required
-@parse_body(DeleteAnnouncementBody)
-def delete_announcement(user, body: DeleteAnnouncementBody):
-    # Delete an announcement
+@ann_router.delete('')
+def delete_announcement(body: DeleteAnnouncementBody,
+                        user=Depends(login_required)):
     ann = Announcement(body.ann_id)
     if not ann:
         return HTTPError('Announcement Not Found', 404)
-
     course = Course(ann.course)
     if not course.permission(user, Course.Permission.GRADE):
         return HTTPError('Failed to Delete Announcement', 403)