Merge pull request #1438 from Sage-Bionetworks/develop-add-errors

feat: surfaced schematic exceptions in flask APIs

schematic_api/api/__init__.py

@@ -1,13 +1,20 @@
 import os
 
 import connexion
+from typing import Tuple
+
+import traceback
+from synapseclient.core.exceptions import (
+    SynapseAuthenticationError,
+)
+from schematic.exceptions import AccessCredentialsError
 
-from schematic import CONFIG
 
 def create_app():
     connexionapp = connexion.FlaskApp(__name__, specification_dir="openapi/")
-    connexionapp.add_api("api.yaml", arguments={"title": "Schematic REST API"}, pythonic_params=True)
-
+    connexionapp.add_api(
+        "api.yaml", arguments={"title": "Schematic REST API"}, pythonic_params=True
+    )
 
     # get the underlying Flask app instance
     app = connexionapp.app
@@ -20,22 +27,38 @@ def create_app():
     app.config["SCHEMATIC_CONFIG"] = schematic_config
     app.config["SCHEMATIC_CONFIG_CONTENT"] = schematic_config_content
 
-    # Configure flask app
-    # app.config[] = schematic[]
-    # app.config[] = schematic[]
-    # app.config[] = schematic[]
+    # handle exceptions in schematic when an exception gets raised
+    @app.errorhandler(Exception)
+    def handle_exception(e: Exception) -> Tuple[str, int]:
+        """handle exceptions in schematic APIs"""
+        # Ensure the application context is available
+        with app.app_context():
+            # Get the last line of error from the traceback
+            last_line = traceback.format_exc().strip().split("\n")[-1]
+
+            # Log the full trace
+            app.logger.error(traceback.format_exc())
+
+            # Return a JSON response with the last line of the error
+            return last_line, 500
 
-    # Initialize extension schematic
-    # import MyExtension
-    # myext = MyExtension()
-    # myext.init_app(app)
+    @app.errorhandler(SynapseAuthenticationError)
+    def handle_synapse_auth_error(e: Exception) -> Tuple[str, int]:
+        """handle synapse authentication error"""
+        return str(e), 401
+
+    @app.errorhandler(AccessCredentialsError)
+    def handle_synapse_access_error(e: Exception) -> Tuple[str, int]:
+        """handle synapse access error"""
+        return str(e), 403
 
     return app
 
+
 app = create_app()
 
 
 # def route_code():
 #     import flask_schematic as sc
 #     sc.method1()
-#
+#

tests/test_api.py

@@ -132,8 +132,30 @@ def request_headers(syn_token):
     yield headers
 
 
+@pytest.fixture
+def request_invalid_headers():
+    headers = {"Authorization": "Bearer invalid headers"}
+    yield headers
+
+
 @pytest.mark.schematic_api
 class TestSynapseStorage:
+    def test_invalid_authentication(self, client, request_invalid_headers):
+        response = client.get(
+            "http://localhost:3001/v1/storage/assets/tables",
+            query_string = {"asset_view": "syn23643253", "return_type": "csv"},
+            headers=request_invalid_headers,
+        )
+        assert response.status_code == 401
+
+    def test_insufficent_auth(self, client, request_headers):
+        response = client.get(
+            "http://localhost:3001/v1/storage/assets/tables",
+            query_string = {"asset_view": "syn23643252", "return_type": "csv"},
+            headers=request_headers,
+        )
+        assert response.status_code == 403
+
     @pytest.mark.synapse_credentials_needed
     @pytest.mark.parametrize("return_type", ["json", "csv"])
     def test_get_storage_assets_tables(self, client, return_type, request_headers):
@@ -791,6 +813,19 @@ def test_generate_manifest_not_file_based_with_annotations(
             ]
         )
 
+    def test_generate_manifest_data_type_not_found(self, client, data_model_jsonld):
+        params = {
+            "schema_url": data_model_jsonld,
+            "data_type": "wrong data type",
+            "use_annotations": False,
+        }
+        response = client.get(
+            "http://localhost:3001/v1/manifest/generate", query_string=params
+        )
+
+        assert response.status_code == 500
+        assert "LookupError" in str(response.data)
+
     def test_populate_manifest(self, client, data_model_jsonld, test_manifest_csv):
         # test manifest
         test_manifest_data = open(test_manifest_csv, "rb")