feat(export_and_import_routes): add export, import route and schemas for save and load feature

Author: MaxNumerique

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -5,6 +5,8 @@
 # Third party imports
 import flask
 import werkzeug
+import zipfile
+import glob
 from opengeodeweb_microservice.schemas import get_schemas_dict
 
 # Local application imports
@@ -267,3 +269,29 @@ def kill() -> flask.Response:
     print("Manual server kill, shutting down...", flush=True)
     os._exit(0)
     return flask.make_response({"message": "Flask server is dead"}, 200)
+
+
+@routes.route(
+    schemas_dict["export_project"]["route"],
+    methods=schemas_dict["export_project"]["methods"],
+)
+def export_project() -> flask.Response:
+    utils_functions.validate_request(flask.request, schemas_dict["export_project"])
+    params = schemas.ExportProject.from_dict(flask.request.get_json())
+
+    data_folder_path: str = flask.current_app.config["DATA_FOLDER_PATH"]
+    upload_folder: str = flask.current_app.config["UPLOAD_FOLDER"]
+    os.makedirs(upload_folder, exist_ok=True)
+
+    filename: str = params.filename or f"project_{int(time.time())}.zip"
+    export_zip_path = os.path.join(upload_folder, filename)
+
+    with zipfile.ZipFile(export_zip_path, "w", compression=8) as zipf:
+        pattern = os.path.join(data_folder_path, "**", "*")
+        for full_path in glob.glob(pattern, recursive=True):
+            if os.path.isfile(full_path):
+                archive_name = os.path.relpath(full_path, start=data_folder_path)
+                zipf.write(full_path, archive_name)
+        zipf.writestr("snapshot.json", flask.json.dumps(params.snapshot))
+
+    return utils_functions.send_file(upload_folder, [export_zip_path], filename)

src/opengeodeweb_back/routes/schemas/__init__.py

@@ -12,3 +12,4 @@
 from .geode_objects_and_output_extensions import *
 from .allowed_objects import *
 from .allowed_files import *
+from .export_project import *

src/opengeodeweb_back/routes/schemas/export_project.json

@@ -0,0 +1,20 @@
+{
+  "route": "/export_project",
+  "methods": [
+    "POST"
+  ],
+  "type": "object",
+  "properties": {
+    "snapshot": {
+      "type": "object"
+    },
+    "filename": {
+      "type": "string",
+      "minLength": 1
+    }
+  },
+  "required": [
+    "snapshot"
+  ],
+  "additionalProperties": false
+}

src/opengeodeweb_back/routes/schemas/export_project.py

@@ -0,0 +1,9 @@
+from dataclasses_json import DataClassJsonMixin
+from dataclasses import dataclass
+from typing import Dict, Any, Optional
+
+
+@dataclass
+class ExportProject(DataClassJsonMixin):
+    snapshot: Dict[str, Any]
+    filename: Optional[str] = None

src/opengeodeweb_back/utils_functions.py

@@ -136,15 +136,17 @@ def send_file(
     else:
         mimetype = "application/zip"
         new_file_name = os.path.splitext(new_file_name)[0] + ".zip"
-        with zipfile.ZipFile(os.path.join(upload_folder, new_file_name), "w") as zipObj:
+        with zipfile.ZipFile(
+            os.path.join(os.path.abspath(upload_folder), new_file_name), "w"
+        ) as zipObj:
             for saved_file_path in saved_files:
                 zipObj.write(
                     saved_file_path,
                     os.path.basename(saved_file_path),
                 )
 
     response = flask.send_from_directory(
-        directory=upload_folder,
+        directory=os.path.abspath(upload_folder),
         path=new_file_name,
         as_attachment=True,
         mimetype=mimetype,

tests/test_models_routes.py

@@ -5,6 +5,9 @@
 from opengeodeweb_back import geode_functions
 from opengeodeweb_microservice.database.data import Data
 from opengeodeweb_microservice.database.connection import get_session
+import zipfile
+import json
+import io
 
 
 def test_model_mesh_components(client, test_id):
@@ -55,3 +58,26 @@ def test_extract_brep_uuids(client, test_id):
         assert "uuid_dict" in response.json
         uuid_dict = response.json["uuid_dict"]
         assert isinstance(uuid_dict, dict)
+
+
+def test_export_project_route(client):
+    route = "/opengeodeweb_back/export_project"
+    snapshot = {"styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}}}
+    filename = "export_project_test.zip"
+    response = client.post(route, json={"snapshot": snapshot, "filename": filename})
+    assert response.status_code == 200
+    assert response.headers.get("new-file-name") == filename
+    assert response.mimetype == "application/octet-binary"
+    response.direct_passthrough = False
+    data = response.get_data()
+    with zipfile.ZipFile(io.BytesIO(data), "r") as zf:
+        names = zf.namelist()
+        assert "snapshot.json" in names
+        parsed = json.loads(zf.read("snapshot.json").decode("utf-8"))
+        assert parsed == snapshot
+        assert "1/project.db" in names
+    response.close()
+    upload_folder = client.application.config["UPLOAD_FOLDER"]
+    export_path = os.path.join(upload_folder, filename)
+    if os.path.exists(export_path):
+        os.remove(export_path)

tests/test_utils_functions.py

@@ -6,6 +6,9 @@
 import flask
 import shutil
 import uuid
+import glob
+import zipfile
+import io
 
 # Local application imports
 from opengeodeweb_microservice.database.data import Data
@@ -206,3 +209,45 @@ def test_generate_native_viewable_and_light_viewable_from_file(client):
     assert isinstance(result["object_type"], str)
     assert isinstance(result["binary_light_viewable"], str)
     assert isinstance(result["input_file"], str)
+
+
+def test_send_file_multiple_returns_zip(client, tmp_path):
+    app = client.application
+    with app.app_context():
+        app.config["UPLOAD_FOLDER"] = str(tmp_path)
+        file_paths = []
+        for i, content in [(1, b"hello 1"), (2, b"hello 2")]:
+            file_path = tmp_path / f"tmp_send_file_{i}.txt"
+            file_path.write_bytes(content)
+            file_paths.append(str(file_path))
+        with app.test_request_context():
+            response = utils_functions.send_file(app.config["UPLOAD_FOLDER"], file_paths, "bundle")
+            assert response.status_code == 200
+            assert response.mimetype == "application/zip"
+            assert response.headers.get("new-file-name") == "bundle.zip"
+
+            response.direct_passthrough = False
+            zip_bytes = response.get_data()
+            with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zip_file:
+                zip_entries = zip_file.namelist()
+                assert "tmp_send_file_1.txt" in zip_entries
+                assert "tmp_send_file_2.txt" in zip_entries
+            response.close()
+
+
+def test_send_file_single_returns_octet_binary(client, tmp_path):
+    app = client.application
+    with app.app_context():
+        app.config["UPLOAD_FOLDER"] = str(tmp_path)
+        file_path = tmp_path / "tmp_send_file_1.txt"
+        file_path.write_bytes(b"hello 1")
+        with app.test_request_context():
+            response = utils_functions.send_file(app.config["UPLOAD_FOLDER"], [str(file_path)], "tmp_send_file_1.txt")
+            assert response.status_code == 200
+            assert response.mimetype == "application/octet-binary"
+            assert response.headers.get("new-file-name") == "tmp_send_file_1.txt"
+
+            response.direct_passthrough = False
+            file_bytes = response.get_data()
+            assert file_bytes == b"hello 1"
+            response.close()