feat(api): global error handling & direct response usage in create endpoints

Author: SpliiT

requirements.txt

@@ -36,7 +36,7 @@ markupsafe~=3.0
     #   flask
     #   jinja2
     #   werkzeug
-opengeode-core==15.27.4
+opengeode-core==15.29.0
     # via
     #   -r requirements.in
     #   geode-common

src/opengeodeweb_back/app.py

@@ -1,34 +1,28 @@
 """Packages"""
-
 import argparse
 import os
 import time
 from typing import Any
-
 import flask
 import flask_cors  # type: ignore
 from flask import Flask, Response
 from flask_cors import cross_origin
 from werkzeug.exceptions import HTTPException
-
 from opengeodeweb_back import utils_functions, app_config
 from opengeodeweb_back.routes import blueprint_routes
 from opengeodeweb_back.routes.models import blueprint_models
 from opengeodeweb_back.routes.create import blueprint_create
 from opengeodeweb_microservice.database.connection import init_database
 
-
 """ Global config """
 app: Flask = flask.Flask(__name__)
 
 """ Config variables """
 FLASK_DEBUG = True if os.environ.get("FLASK_DEBUG", default=None) == "True" else False
-
 if FLASK_DEBUG == False:
     app.config.from_object(app_config.ProdConfig)
 else:
     app.config.from_object(app_config.DevConfig)
-
 DEFAULT_HOST: str = app.config.get("DEFAULT_HOST") or "localhost"
 DEFAULT_PORT: int = int(app.config.get("DEFAULT_PORT") or 5000)
 DEFAULT_DATA_FOLDER_PATH: str = app.config.get("DEFAULT_DATA_FOLDER_PATH") or "./data"
@@ -39,19 +33,16 @@
     app.config.get("SECONDS_BETWEEN_SHUTDOWNS") or 60.0
 )
 
-
 app.register_blueprint(
     blueprint_routes.routes,
     url_prefix="/opengeodeweb_back",
     name="opengeodeweb_back",
 )
-
 app.register_blueprint(
     blueprint_models.routes,
     url_prefix="/opengeodeweb_back/models",
     name="opengeodeweb_models",
 )
-
 app.register_blueprint(
     blueprint_create.routes,
     url_prefix="/opengeodeweb_back/create",
@@ -63,11 +54,13 @@
         utils_functions.kill_task, SECONDS_BETWEEN_SHUTDOWNS, app
     )
 
-
 @app.errorhandler(HTTPException)
 def errorhandler(e: HTTPException) -> tuple[dict[str, Any], int] | Response:
     return utils_functions.handle_exception(e)
 
+@app.errorhandler(Exception)
+def handle_generic_exception(e: Exception) -> Response:
+    return flask.make_response({"error": str(e)}, 500)
 
 @app.route(
     "/error",
@@ -77,20 +70,17 @@ def return_error() -> Response:
     flask.abort(500, f"Test")
     return flask.make_response({}, 500)
 
-
 @app.route("/", methods=["POST"])
 @cross_origin()
 def root() -> Response:
     return flask.make_response({}, 200)
 
-
 @app.route("/kill", methods=["POST"])
 @cross_origin()
 def kill() -> None:
     print("Manual server kill, shutting down...", flush=True)
     os._exit(0)
 
-
 def run_server() -> None:
     parser = argparse.ArgumentParser(
         prog="OpenGeodeWeb-Back", description="Backend server for OpenGeodeWeb"
@@ -133,31 +123,25 @@ def run_server() -> None:
         help="Number of minutes before the server times out",
     )
     args = parser.parse_args()
-
     app.config.update(DATA_FOLDER_PATH=args.data_folder_path)
     app.config.update(UPLOAD_FOLDER=args.upload_folder_path)
     app.config.update(MINUTES_BEFORE_TIMEOUT=args.timeout)
-
     flask_cors.CORS(app, origins=args.allowed_origins)
-
     print(
         f"Host: {args.host}, Port: {args.port}, Debug: {args.debug}, "
         f"Data folder path: {args.data_folder_path}, Timeout: {args.timeout}, "
         f"Origins: {args.allowed_origins}",
         flush=True,
     )
-
     db_filename: str = app.config.get("DATABASE_FILENAME") or "database.db"
     db_path = os.path.join(args.data_folder_path, db_filename)
     os.makedirs(os.path.dirname(db_path), exist_ok=True)
     app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
     init_database(app, db_filename)
     print(f"Database initialized at: {db_path}", flush=True)
-
     app.run(debug=args.debug, host=args.host, port=args.port, ssl_context=SSL)
 
-
 # ''' Main '''
 if __name__ == "__main__":
     run_server()

src/opengeodeweb_back/routes/create/blueprint_create.py

@@ -1,13 +1,11 @@
 # Standard library imports
 import json
 import os
-from typing import Dict, Any, List, TypedDict, cast
-
+from typing import Any, TypedDict, cast
 # Third party imports
 import flask
 import opengeode
 import werkzeug
-
 # Local application imports
 from opengeodeweb_back import geode_functions, utils_functions
 from opengeodeweb_microservice.database.data import Data
@@ -30,12 +28,12 @@ class CreatePointRequest(TypedDict):
 
 class CreateAOIRequest(TypedDict):
     name: str
-    points: List[Point]
+    points: list[Point]
     z: float
 
 # Load schema for point creation
 with open(os.path.join(schemas, "create_point.json"), "r") as file:
-    create_point_json = cast(Dict[str, Any], json.load(file))
+    create_point_json = cast(dict[str, Any], json.load(file))
 
 @routes.route(
     create_point_json["route"],
@@ -46,50 +44,33 @@ def create_point() -> flask.Response:
     print(f"create_point : {flask.request=}", flush=True)
     utils_functions.validate_request(flask.request, create_point_json)
 
-    try:
-        # Extract and validate data from request
-        request_data = cast(CreatePointRequest, flask.request.json)
-        name: str = request_data["name"]
-        x: float = request_data["x"]
-        y: float = request_data["y"]
-        z: float = request_data["z"]
-
-        # Create the point set
-        class_ = geode_functions.geode_object_class("PointSet3D")
-        point_set = class_.create()
-        builder = geode_functions.create_builder("PointSet3D", point_set)
-        builder.create_point(opengeode.Point3D([x, y, z]))  # type: ignore
-        builder.set_name(name)
-
-        # Save and get info
-        result = save_all_viewables_and_return_info(
-            geode_object="PointSet3D",
-            data=point_set,
-        )
-
-        # Vérifier que binary_light_viewable est présent
-        if "binary_light_viewable" not in result:
-            raise ValueError("binary_light_viewable is missing in the result")
-
-        # Prepare response
-        response: Dict[str, Any] = {
-            "viewable_file_name": result["viewable_file_name"],
-            "id": result["id"],
-            "name": name,
-            "native_file_name": result["native_file_name"],
-            "object_type": result["object_type"],
-            "geode_object": result["geode_object"],
-            "binary_light_viewable": result["binary_light_viewable"],
-        }
-
-        return flask.make_response(response, 200)
-
-    except Exception as e:
-        return flask.make_response({"error": str(e)}, 500)
+    # Extract and validate data from request
+    request_data = cast(CreatePointRequest, flask.request.json)
+    name: str = request_data["name"]
+    x: float = request_data["x"]
+    y: float = request_data["y"]
+    z: float = request_data["z"]
+
+    # Create the point set
+    class_ = geode_functions.geode_object_class("PointSet3D")
+    point_set = class_.create()
+    builder = geode_functions.create_builder("PointSet3D", point_set)
+    builder.create_point(opengeode.Point3D([x, y, z]))
+    builder.set_name(name)
+
+    # Save and get info
+    result = save_all_viewables_and_return_info(
+        geode_object="PointSet3D",
+        data=point_set,
+    )
+    result["name"] = name
+    if "binary_light_viewable" not in result:
+        raise ValueError("binary_light_viewable is missing in the result")
+    return flask.make_response(result, 200)
 
 # Load schema for AOI creation
 with open(os.path.join(schemas, "create_aoi.json"), "r") as file:
-    create_aoi_json = cast(Dict[str, Any], json.load(file))
+    create_aoi_json = cast(dict[str, Any], json.load(file))
 
 @routes.route(
     create_aoi_json["route"],
@@ -100,55 +81,38 @@ def create_aoi() -> flask.Response:
     print(f"create_aoi : {flask.request=}", flush=True)
     utils_functions.validate_request(flask.request, create_aoi_json)
 
-    try:
-        # Extract and validate data from request
-        request_data = cast(CreateAOIRequest, flask.request.json)
-        name: str = request_data["name"]
-        points: List[Point] = request_data["points"]
-        z: float = request_data["z"]
-
-        # Create the edged curve
-        class_ = geode_functions.geode_object_class("EdgedCurve3D")
-        edged_curve = class_.create()
-        builder = geode_functions.create_builder("EdgedCurve3D", edged_curve)
-        builder.set_name(name)
-
-        # Create vertices first
-        vertex_indices: List[int] = []
-        for point in points:
-            vertex_id = builder.create_point(opengeode.Point3D([point["x"], point["y"], z]))  # type: ignore
-            vertex_indices.append(vertex_id)
-
-        # Create edges between consecutive vertices and close the loop
-        num_vertices = len(vertex_indices)
-        for i in range(num_vertices):
-            next_i = (i + 1) % num_vertices
-            edge_id = builder.create_edge()
-            builder.set_edge_vertex(opengeode.EdgeVertex(edge_id, 0), vertex_indices[i])  # type: ignore
-            builder.set_edge_vertex(opengeode.EdgeVertex(edge_id, 1), vertex_indices[next_i])  # type: ignore
-
-        # Save and get info
-        result = save_all_viewables_and_return_info(
-            geode_object="EdgedCurve3D",
-            data=edged_curve,
-        )
-
-        # Vérifier que binary_light_viewable est présent
-        if "binary_light_viewable" not in result:
-            raise ValueError("binary_light_viewable is missing in the result")
-
-        # Prepare response
-        response: Dict[str, Any] = {
-            "viewable_file_name": result["viewable_file_name"],
-            "id": result["id"],
-            "name": name,
-            "native_file_name": result["native_file_name"],
-            "object_type": result["object_type"],
-            "geode_object": result["geode_object"],
-            "binary_light_viewable": result["binary_light_viewable"],
-        }
-
-        return flask.make_response(response, 200)
-
-    except Exception as e:
-        return flask.make_response({"error": str(e)}, 500)
+    # Extract and validate data from request
+    request_data = cast(CreateAOIRequest, flask.request.json)
+    name: str = request_data["name"]
+    points: list[Point] = request_data["points"]
+    z: float = request_data["z"]
+
+    # Create the edged curve
+    class_ = geode_functions.geode_object_class("EdgedCurve3D")
+    edged_curve = class_.create()
+    builder = geode_functions.create_builder("EdgedCurve3D", edged_curve)
+    builder.set_name(name)
+
+    # Create vertices first
+    vertex_indices: list[int] = []
+    for point in points:
+        vertex_id = builder.create_point(opengeode.Point3D([point["x"], point["y"], z]))
+        vertex_indices.append(vertex_id)
+
+    # Create edges between consecutive vertices and close the loop
+    num_vertices = len(vertex_indices)
+    for i in range(num_vertices):
+        next_i = (i + 1) % num_vertices
+        edge_id = builder.create_edge()
+        builder.set_edge_vertex(opengeode.EdgeVertex(edge_id, 0), vertex_indices[i])
+        builder.set_edge_vertex(opengeode.EdgeVertex(edge_id, 1), vertex_indices[next_i])
+
+    # Save and get info
+    result = save_all_viewables_and_return_info(
+        geode_object="EdgedCurve3D",
+        data=edged_curve,
+    )
+    result["name"] = name
+    if "binary_light_viewable" not in result:
+        raise ValueError("binary_light_viewable is missing in the result")
+    return flask.make_response(result, 200)