fix(Schemas): using typed schemas

Author: BotellaA

src/opengeodeweb_back/geode_functions.py

@@ -175,7 +175,7 @@ def filter_geode_objects(key: str = None):
     return geode_objects_filtered_list
 
 
-def list_input_extensions(key: str = None):
+def list_input_extensions(key: str | None = None):
     extensions_list = []
     geode_objects_filtered_list = filter_geode_objects(key)
     for geode_object in geode_objects_filtered_list:
@@ -192,7 +192,7 @@ def has_creator(geode_object: str, extension: str):
 
 def list_geode_objects(
     file_absolute_path: str,
-    key: str = None,
+    key: str | None = None,
 ):
     return_dict = {}
     file_extension = utils_functions.extension_from_filename(

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -5,14 +5,12 @@
 
 # Third party imports
 import flask
-import opengeode
 import werkzeug
 
 # Local application imports
-from .. import geode_functions, utils_functions
-from opengeodeweb_microservice.database.data import Data
-from opengeodeweb_microservice.database.connection import get_session
+from opengeodeweb_back import geode_functions, utils_functions
 from .models import blueprint_models
+from . import schemas
 
 routes = flask.Blueprint("routes", __name__, url_prefix="/opengeodeweb_back")
 
@@ -23,39 +21,38 @@
     name=blueprint_models.routes.name,
 )
 
-schemas = os.path.join(os.path.dirname(__file__), "schemas")
+schemas_folder = os.path.join(os.path.dirname(__file__), "schemas")
 
 with open(
-    os.path.join(schemas, "allowed_files.json"),
+    os.path.join(schemas_folder, "allowed_files.json"),
     "r",
 ) as file:
-    allowed_files_json = json.load(file)
+    allowed_files_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     allowed_files_json["route"],
     methods=allowed_files_json["methods"],
 )
-def allowed_files():
+def allowed_files() -> flask.Response:
     utils_functions.validate_request(flask.request, allowed_files_json)
-    extensions = geode_functions.list_input_extensions(
-        flask.request.get_json()["supported_feature"]
-    )
+    params = schemas.AllowedFiles(**flask.request.get_json())
+    extensions = geode_functions.list_input_extensions(params.supported_feature)
     return flask.make_response({"extensions": extensions}, 200)
 
 
 with open(
-    os.path.join(schemas, "upload_file.json"),
+    os.path.join(schemas_folder, "upload_file.json"),
     "r",
 ) as file:
-    upload_file_json = json.load(file)
+    upload_file_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     upload_file_json["route"],
     methods=upload_file_json["methods"],
 )
-def upload_file():
+def upload_file() -> flask.Response:
     if flask.request.method == "OPTIONS":
         return flask.make_response({}, 200)
 
@@ -69,47 +66,47 @@ def upload_file():
 
 
 with open(
-    os.path.join(schemas, "allowed_objects.json"),
+    os.path.join(schemas_folder, "allowed_objects.json"),
     "r",
 ) as file:
-    allowed_objects_json = json.load(file)
+    allowed_objects_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     allowed_objects_json["route"],
     methods=allowed_objects_json["methods"],
 )
-def allowed_objects():
+def allowed_objects() -> flask.Response:
     if flask.request.method == "OPTIONS":
         return flask.make_response({}, 200)
 
     utils_functions.validate_request(flask.request, allowed_objects_json)
-    file_absolute_path = geode_functions.upload_file_path(
-        flask.request.get_json()["filename"]
-    )
+    params = schemas.AllowedObjects(**flask.request.get_json())
+    file_absolute_path = geode_functions.upload_file_path(params.filename)
     allowed_objects = geode_functions.list_geode_objects(
-        file_absolute_path, flask.request.get_json()["supported_feature"]
+        file_absolute_path, params.supported_feature
     )
     return flask.make_response({"allowed_objects": allowed_objects}, 200)
 
 
 with open(
-    os.path.join(schemas, "missing_files.json"),
+    os.path.join(schemas_folder, "missing_files.json"),
     "r",
 ) as file:
-    missing_files_json = json.load(file)
+    missing_files_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     missing_files_json["route"],
     methods=missing_files_json["methods"],
 )
-def missing_files():
+def missing_files() -> flask.Response:
     utils_functions.validate_request(flask.request, missing_files_json)
-    file_path = geode_functions.upload_file_path(flask.request.get_json()["filename"])
+    params = schemas.MissingFiles(**flask.request.get_json())
+    file_path = geode_functions.upload_file_path(params.filename)
 
     additional_files = geode_functions.additional_files(
-        flask.request.get_json()["input_geode_object"],
+        params.input_geode_object,
         file_path,
     )
 
@@ -140,21 +137,20 @@ def missing_files():
 
 
 with open(
-    os.path.join(schemas, "geographic_coordinate_systems.json"),
+    os.path.join(schemas_folder, "geographic_coordinate_systems.json"),
     "r",
 ) as file:
-    geographic_coordinate_systems_json = json.load(file)
+    geographic_coordinate_systems_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     geographic_coordinate_systems_json["route"],
     methods=geographic_coordinate_systems_json["methods"],
 )
-def crs_converter_geographic_coordinate_systems():
+def crs_converter_geographic_coordinate_systems() -> flask.Response:
     utils_functions.validate_request(flask.request, geographic_coordinate_systems_json)
-    infos = geode_functions.geographic_coordinate_systems(
-        flask.request.get_json()["input_geode_object"]
-    )
+    params = schemas.GeographicCoordinateSystems(**flask.request.get_json())
+    infos = geode_functions.geographic_coordinate_systems(params.input_geode_object)
     crs_list = []
     for info in infos:
         crs = {}
@@ -167,54 +163,51 @@ def crs_converter_geographic_coordinate_systems():
 
 
 with open(
-    os.path.join(schemas, "inspect_file.json"),
+    os.path.join(schemas_folder, "inspect_file.json"),
     "r",
 ) as file:
-    inspect_file_json = json.load(file)
+    inspect_file_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     inspect_file_json["route"],
     methods=inspect_file_json["methods"],
 )
-def inspect_file():
+def inspect_file() -> flask.Response:
     utils_functions.validate_request(flask.request, inspect_file_json)
-
-    file_path = geode_functions.upload_file_path(flask.request.get_json()["filename"])
-    data = geode_functions.load(
-        flask.request.get_json()["input_geode_object"], file_path
-    )
-    class_inspector = geode_functions.inspect(
-        flask.request.get_json()["input_geode_object"], data
-    )
+    params = schemas.InspectFile(**flask.request.get_json())
+    file_path = geode_functions.upload_file_path(params.filename)
+    data = geode_functions.load(params.input_geode_object, file_path)
+    class_inspector = geode_functions.inspect(params.input_geode_object, data)
     inspection_result = geode_functions.get_inspector_children(class_inspector)
     return flask.make_response({"inspection_result": inspection_result}, 200)
 
 
 with open(
-    os.path.join(schemas, "geode_objects_and_output_extensions.json"),
+    os.path.join(schemas_folder, "geode_objects_and_output_extensions.json"),
     "r",
 ) as file:
-    geode_objects_and_output_extensions_json = json.load(file)
+    geode_objects_and_output_extensions_json: utils_functions.SchemaDict = json.load(
+        file
+    )
 
 
 @routes.route(
     geode_objects_and_output_extensions_json["route"],
     methods=geode_objects_and_output_extensions_json["methods"],
 )
-def geode_objects_and_output_extensions():
+def geode_objects_and_output_extensions() -> flask.Response:
     utils_functions.validate_request(
         flask.request, geode_objects_and_output_extensions_json
     )
-    file_path = geode_functions.upload_file_path(flask.request.get_json()["filename"])
+    params = schemas.GeodeObjectsAndOutputExtensions(**flask.request.get_json())
+    file_path = geode_functions.upload_file_path(params.filename)
     data = geode_functions.load(
-        flask.request.get_json()["input_geode_object"],
+        params.input_geode_object,
         file_path,
     )
     geode_objects_and_output_extensions = (
-        geode_functions.geode_objects_output_extensions(
-            flask.request.get_json()["input_geode_object"], data
-        )
+        geode_functions.geode_objects_output_extensions(params.input_geode_object, data)
     )
     return flask.make_response(
         {"geode_objects_and_output_extensions": geode_objects_and_output_extensions},
@@ -223,56 +216,59 @@ def geode_objects_and_output_extensions():
 
 
 with open(
-    os.path.join(schemas, "save_viewable_file.json"),
+    os.path.join(schemas_folder, "save_viewable_file.json"),
     "r",
 ) as file:
-    save_viewable_file_json = json.load(file)
+    save_viewable_file_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     save_viewable_file_json["route"],
     methods=save_viewable_file_json["methods"],
 )
-def save_viewable_file():
+def save_viewable_file() -> flask.Response:
     utils_functions.validate_request(flask.request, save_viewable_file_json)
+    params = schemas.SaveViewableFile(**flask.request.get_json())
     return flask.make_response(
         utils_functions.generate_native_viewable_and_light_viewable_from_file(
-            geode_object=flask.request.get_json()["input_geode_object"],
-            input_filename=flask.request.get_json()["filename"],
+            geode_object=params.input_geode_object,
+            input_filename=params.filename,
         ),
         200,
     )
 
 
-with open(os.path.join(schemas, "texture_coordinates.json"), "r") as file:
-    texture_coordinates_json = json.load(file)
+with open(os.path.join(schemas_folder, "texture_coordinates.json"), "r") as file:
+    texture_coordinates_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     texture_coordinates_json["route"],
     methods=texture_coordinates_json["methods"],
 )
-def texture_coordinates():
+def texture_coordinates() -> flask.Response:
     utils_functions.validate_request(flask.request, texture_coordinates_json)
-    data = geode_functions.load_data(flask.request.get_json().get("id"))
+    params = schemas.TextureCoordinates(**flask.request.get_json())
+    data = geode_functions.load_data(params.id)
     texture_coordinates = data.texture_manager().texture_names()
     return flask.make_response({"texture_coordinates": texture_coordinates}, 200)
 
 
 with open(
-    os.path.join(schemas, "vertex_attribute_names.json"),
+    os.path.join(schemas_folder, "vertex_attribute_names.json"),
     "r",
 ) as file:
-    vertex_attribute_names_json = json.load(file)
+    vertex_attribute_names_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     vertex_attribute_names_json["route"],
     methods=vertex_attribute_names_json["methods"],
 )
-def vertex_attribute_names():
+def vertex_attribute_names() -> flask.Response:
     utils_functions.validate_request(flask.request, vertex_attribute_names_json)
-    data = geode_functions.load_data(flask.request.get_json().get("id"))
+    params = schemas.VertexAttributeNames(**flask.request.get_json())
+    data = geode_functions.load_data(params.id)
     vertex_attribute_names = data.vertex_attribute_manager().attribute_names()
     return flask.make_response(
         {
@@ -283,19 +279,20 @@ def vertex_attribute_names():
 
 
 with open(
-    os.path.join(schemas, "polygon_attribute_names.json"),
+    os.path.join(schemas_folder, "polygon_attribute_names.json"),
     "r",
 ) as file:
-    polygon_attribute_names_json = json.load(file)
+    polygon_attribute_names_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     polygon_attribute_names_json["route"],
     methods=polygon_attribute_names_json["methods"],
 )
-def polygon_attribute_names():
+def polygon_attribute_names() -> flask.Response:
     utils_functions.validate_request(flask.request, polygon_attribute_names_json)
-    data = geode_functions.load_data(flask.request.get_json().get("id"))
+    params = schemas.PolygonAttributeNames(**flask.request.get_json())
+    data = geode_functions.load_data(params.id)
     polygon_attribute_names = data.polygon_attribute_manager().attribute_names()
     return flask.make_response(
         {
@@ -306,19 +303,20 @@ def polygon_attribute_names():
 
 
 with open(
-    os.path.join(schemas, "polyhedron_attribute_names.json"),
+    os.path.join(schemas_folder, "polyhedron_attribute_names.json"),
     "r",
 ) as file:
-    polyhedron_attribute_names_json = json.load(file)
+    polyhedron_attribute_names_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     polyhedron_attribute_names_json["route"],
     methods=polyhedron_attribute_names_json["methods"],
 )
-def polyhedron_attribute_names():
+def polyhedron_attribute_names() -> flask.Response:
     utils_functions.validate_request(flask.request, polyhedron_attribute_names_json)
-    data = geode_functions.load_data(flask.request.get_json().get("id"))
+    params = schemas.PolyhedronAttributeNames(**flask.request.get_json())
+    data = geode_functions.load_data(params.id)
     polyhedron_attribute_names = data.polyhedron_attribute_manager().attribute_names()
     return flask.make_response(
         {
@@ -329,27 +327,27 @@ def polyhedron_attribute_names():
 
 
 with open(
-    os.path.join(schemas, "ping.json"),
+    os.path.join(schemas_folder, "ping.json"),
     "r",
 ) as file:
-    ping_json = json.load(file)
+    ping_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(
     ping_json["route"],
     methods=ping_json["methods"],
 )
-def ping():
+def ping() -> flask.Response:
     utils_functions.validate_request(flask.request, ping_json)
     flask.current_app.config.update(LAST_PING_TIME=time.time())
     return flask.make_response({"message": "Flask server is running"}, 200)
 
 
 with open(
-    os.path.join(schemas, "kill.json"),
+    os.path.join(schemas_folder, "kill.json"),
     "r",
 ) as file:
-    kill_json = json.load(file)
+    kill_json: utils_functions.SchemaDict = json.load(file)
 
 
 @routes.route(kill_json["route"], methods=kill_json["methods"])