Flask-side refactors

pyflask/apis/data.py

@@ -5,6 +5,7 @@
 from errorHandlers import notBadRequestException
 from flask_restx import Namespace, Resource, reqparse
 from manageNeuroconv import generate_dataset, generate_test_data
+from utils import catch_exception_and_abort, server_error_responses
 
 data_api = Namespace("data", description="API route for dataset generation in the NWB GUIDE.")
 
@@ -22,15 +23,14 @@ def exception_handler(error):
 @data_api.route("/generate")
 @data_api.expect(generate_test_data_parser)
 class GeneratetestData(Resource):
-    @data_api.doc(responses={200: "Success", 400: "Bad Request", 500: "Internal server error"})
+    @data_api.doc(
+        description="Generate example ecephys data using SpikeInterface.",
+        responses=server_error_responses(codes=[200, 500]),
+    )
+    @catch_exception_and_abort(api=data_api, code=500)
     def post(self):
-        try:
-            arguments = generate_test_data_parser.parse_args()
-            generate_test_data(output_path=arguments["output_path"])
-        except Exception as exception:
-            if notBadRequestException(exception):
-                data_api.abort(500, str(exception))
-            raise exception
+        arguments = generate_test_data_parser.parse_args()
+        generate_test_data(output_path=arguments["output_path"])
 
 
 generate_test_dataset_parser = reqparse.RequestParser()

pyflask/apis/neuroconv.py

@@ -1,6 +1,7 @@
 """API endpoint definitions for interacting with NeuroConv."""
 
 import traceback
+from typing import Dict
 
 from errorHandlers import notBadRequestException
 from flask import Response, request
@@ -25,16 +26,18 @@
 )
 from manageNeuroconv.info import announcer
 
-neuroconv_api = Namespace("neuroconv", description="Neuroconv neuroconv_api for the NWB GUIDE.")
+neuroconv_api = Namespace(name="neuroconv", description="Neuroconv neuroconv_api for the NWB GUIDE.")
 
 parser = reqparse.RequestParser()
-parser.add_argument("interfaces", type=str, action="split", help="Interfaces cannot be converted")
+parser.add_argument("interfaces", type=str, action="split", help="Interfaces cannot be converted.")
 
 
 @neuroconv_api.errorhandler(Exception)
-def exception_handler(error):
-    exceptiondata = traceback.format_exception(type(error), error, error.__traceback__)
-    return {"message": exceptiondata[-1], "traceback": "".join(exceptiondata)}
+def exception_handler(error: Exception) -> Dict[str, str]:
+    full_traceback = traceback.format_exception(type(error), error, error.__traceback__)
+    message = full_traceback[-1]
+    remaining_traceback = "".join(full_traceback[:-1])
+    return {"message": message, "traceback": remaining_traceback}
 
 
 @neuroconv_api.route("/")

pyflask/app.py

@@ -12,14 +12,12 @@
 from signal import SIGINT
 from urllib.parse import unquote
 
-from errorHandlers import notBadRequestException
-
 # https://stackoverflow.com/questions/32672596/pyinstaller-loads-script-multiple-times#comment103216434_32677108
 multiprocessing.freeze_support()
 
 
 from apis import data_api, neuroconv_api, startup_api
-from flask import Flask, request, send_file, send_from_directory
+from flask import Flask, Response, request, send_file, send_from_directory
 from flask_cors import CORS
 from flask_restx import Api, Resource
 from manageNeuroconv.info import (
@@ -28,12 +26,13 @@
     STUB_SAVE_FOLDER_PATH,
     resource_path,
 )
+from utils import catch_exception_and_abort, server_error_responses
 
-app = Flask(__name__)
+flask_app = Flask(__name__)
 
 # Always enable CORS to allow distinct processes to handle frontend vs. backend
-CORS(app)
-app.config["CORS_HEADERS"] = "Content-Type"
+CORS(flask_app)
+flask_app.config["CORS_HEADERS"] = "Content-Type"
 
 # Create logger configuration
 LOG_FOLDER = Path(GUIDE_ROOT_FOLDER, "logs")
@@ -54,46 +53,85 @@
 api.add_namespace(startup_api)
 api.add_namespace(neuroconv_api)
 api.add_namespace(data_api)
-api.init_app(app)
+api.init_app(flask_app)
+
+# 'nwbfile_registry' is a global list that keeps track of all NWB files that have been registered with the server
+nwbfile_registry = []
+
+
+# TODO: is there any advantage to using the api.route instead of app for resources added in this file?
+@api.route(rule="/log")
+@api.doc(
+    description="Any exception that occurs on the Flask server will save a full traceback to a log file on disk.",
+    responses=server_error_responses(codes=[200, 400, 500]),
+)
+class Log(Resource):
+    @api.doc(description="Nicely format the exception and the payload that caused it.", responses=all_error_responses)
+    @catch_exception_and_abort(api=api, code=500)
+    def post(self):
+        payload = api.payload
+        type = payload["type"]
+        header = payload["header"]
+        inputs = payload["inputs"]
+        exception_traceback = payload["traceback"]
 
-registered = {}
+        message = f"{header}\n{'-'*len(header)}\n\n{json.dumps(inputs, indent=2)}\n\n{exception_traceback}\n"
+        selected_logger = getattr(api.logger, type)
+        selected_logger(message)
 
 
-@app.route("/files")
+@flask_app.route(rule="/files")
+@api.doc(
+    description="Get a list of all files that have been nwbfile_registry with the server.",
+    responses=server_error_responses(codes=[200, 400, 500]),
+)
+@catch_exception_and_abort(api=api, code=500)
 def get_all_files():
-    return list(registered.keys())
+    return list(nwbfile_registry.keys())
 
 
-@app.route("/files/<path:path>", methods=["GET", "POST"])
-def handle_file_request(path):
-    if request.method == "GET":
-        if registered[path]:
-            path = unquote(path)
-            if not isabs(path):
-                path = f"/{path}"
-            return send_file(path)
-        else:
-            app.abort(404, "Resource is not accessible.")
+@flask_app.route(rule="/files/<file_path:path>", methods=["GET", "POST"])
+@api.doc(
+    description="Handle adding and fetching NWB files from the global file registry.",
+    responses=server_error_responses(codes=[200, 400, 404, 500]),
+)
+def handle_file_request(file_path: str) -> Union[str, Response, None]:
+    if ".nwb" not in file_path:
+        code = 400
+        base_message = server_error_responses(codes=[code])[code]
+        message = f"{base_message}: Path does not point to an NWB file."
+        flask_app.abort(code=code, message=message)
+        return
+
+    if request.method == "GET" and file_path not in nwbfile_registry:
+        code = 404
+        base_message = server_error_responses(codes=[code])[code]
+        message = f"{base_message}: Path does not point to an NWB file."
+        flask_app.abort(code=code, message=message)
+        return
 
-    else:
-        if ".nwb" in path:
-            registered[path] = True
-            return request.base_url
-        else:
-            app.abort(400, str("Path does not point to an NWB file."))
+    if request.method == "GET":
+        parsed_file_path = unquote(file_path)
+        is_file_relative = not isabs(parsed_file_path)
+        if is_file_relative:
+            parsed_file_path = f"/{parsed_file_path}"
+        return send_file(path_or_file=parsed_file_path)
+    elif request.method == "POST":
+        nwbfile_registry.append(file_path)
+        return request.base_url
 
 
-@app.route("/conversions/<path:path>")
+@flask_app.route("/conversions/<path:path>")
 def send_conversions(path):
     return send_from_directory(CONVERSION_SAVE_FOLDER_PATH, path)
 
 
-@app.route("/preview/<path:path>")
+@flask_app.route("/preview/<path:path>")
 def send_preview(path):
     return send_from_directory(STUB_SAVE_FOLDER_PATH, path)
 
 
-@app.route("/cpus")
+@flask_app.route("/cpus")
 def get_cpu_count():
     from psutil import cpu_count
 
@@ -103,45 +141,24 @@ def get_cpu_count():
     return dict(physical=physical, logical=logical)
 
 
-@app.route("/get-recommended-species")
+@flask_app.route("/get-recommended-species")
 def get_species():
     from dandi.metadata.util import species_map
 
     return species_map
 
 
-@api.route("/log")
-class Log(Resource):
-    @api.doc(responses={200: "Success", 400: "Bad Request", 500: "Internal server error"})
-    def post(self):
-        try:
-
-            payload = api.payload
-            type = payload["type"]
-            header = payload["header"]
-            inputs = payload["inputs"]
-            traceback = payload["traceback"]
-
-            message = f"{header}\n{'-'*len(header)}\n\n{json.dumps(inputs, indent=2)}\n\n{traceback}\n"
-            selected_logger = getattr(api.logger, type)
-            selected_logger(message)
-
-        except Exception as exception:
-            if notBadRequestException(exception):
-                api.abort(500, str(exception))
-
-
 @api.route("/server_shutdown", endpoint="shutdown")
 class Shutdown(Resource):
-    def get(self):
-        func = request.environ.get("werkzeug.server.shutdown")
-        api.logger.info("Shutting down server")
+    def get(self) -> None:
+        werkzeug_shutdown_function = request.environ.get("werkzeug.server.shutdown")
+        api.logger.info("Shutting down server...")
 
-        if func is None:
+        if werkzeug_shutdown_function is None:
             kill(getpid(), SIGINT)
             return
 
-        func()
+        werkzeug_shutdown_function()
 
 
 if __name__ == "__main__":
@@ -156,13 +173,13 @@ def get(self):
         )
         log_handler.setFormatter(log_formatter)
 
-        app.logger.addHandler(log_handler)
-        app.logger.setLevel(DEBUG)
+        flask_app.logger.addHandler(log_handler)
+        flask_app.logger.setLevel(DEBUG)
 
-        app.logger.info(f"Logging to {LOG_FILE_PATH}")
+        flask_app.logger.info(f"Logging to {LOG_FILE_PATH}")
 
         # Run the server
         api.logger.info(f"Starting server on port {port}")
-        app.run(host="127.0.0.1", port=port)
+        flask_app.run(host="127.0.0.1", port=port)
     else:
         raise Exception("No port provided for the NWB GUIDE backend.")

pyflask/tests/conftest.py

@@ -3,7 +3,7 @@
 
 
 def pytest_addoption(parser):
-    parser.addoption("--target", action="store", help="Run the executable instead of the standard Flask app")
+    parser.addoption("--target", action="store", help="Run the executable instead of the standard Flask flask_app")
 
 
 @pytest.fixture(scope="session")
@@ -12,7 +12,7 @@ def client(request):
     if target:
         return target
     else:
-        app = flask.app
+        app = flask.flask_app
         app.config.update(
             {
                 "TESTING": True,

pyflask/utils/__init__.py

@@ -0,0 +1,3 @@
+from ._catch_flask_errors import catch_exception_and_abort, server_error_responses
+
+__all__ = ["catch_exception_and_abort", "server_error_responses"]

pyflask/utils/_catch_flask_errors.py

@@ -0,0 +1,27 @@
+import contextlib
+
+import flask_restx
+from tpying import List, Union
+
+
+def server_error_responses(*, codes: List[str]) -> Dict[int, str]:
+    all_server_error_responses = {
+        200: "Success",
+        400: "Bad request",
+        404: "Resource is not accessible.",
+        500: ("Internalerver error"),
+    }
+
+    selected_responses = {code: all_server_error_responses[code] for code in codes}
+    return selected_responses
+
+
+@contextlib.contextmanager
+def catch_exception_and_abort(*, api: Union[flask_restx.Api, flask_restx.Namespace], code: int) -> None:
+    try:
+        yield
+    except Exception as exception:
+        exception_type = type(exception)
+        exception_message = str(exception)
+        api.abort(code=code, message=f"{exception_type}: {exception_message}")
+        raise exception