new test session.commit to see if it is relevant

used non legacy SQLAlchemy methods
mypy typing

Author: MaxNumerique

src/opengeodeweb_back/data.py

@@ -1,6 +1,7 @@
 from sqlalchemy import String, JSON
 from .database import database
 import uuid
+from typing import List
 
 
 class Data(database.Model):
@@ -20,12 +21,23 @@ class Data(database.Model):
     additional_files = database.Column(JSON, nullable=True)
 
     @staticmethod
-    def create(name, geode_object, input_file, additional_files=[]):
+    def create(
+        name: str,
+        geode_object: str,
+        input_file: str, 
+        additional_files: List[str]
+    ) -> 'Data':
+        if additional_files is None:
+            additional_files = []
+            
         data_entry = Data(
             name=name,
             geode_object=geode_object,
             input_file=input_file,
             additional_files=additional_files,
+            native_file_name="",
+            viewable_file_name="",
+            light_viewable="",
         )
 
         database.session.add(data_entry)

src/opengeodeweb_back/database.py

@@ -1,11 +1,12 @@
+from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 
 DATABASE_FILENAME = "project.db"
 
 database = SQLAlchemy()
 
 
-def initialize_database(app):
+def initialize_database(app: Flask) -> SQLAlchemy:
     database.init_app(app)
     with app.app_context():
         database.create_all()

src/opengeodeweb_back/utils_functions.py

@@ -3,6 +3,7 @@
 import threading
 import time
 import zipfile
+from typing import List, Dict, Any
 
 # Third party imports
 import flask
@@ -154,8 +155,11 @@ def create_data_folder_from_id(data_id: str) -> str:
 
 
 def save_all_viewables_and_return_info(
-    geode_object, data, input_file, additional_files=None
-):
+    geode_object: str, 
+    data: Any, 
+    input_file: List[str], 
+    additional_files: List[str] = []
+) -> Dict[str, Any]:
     data_entry = Data.create(
         name=data.name(),
         geode_object=geode_object,
@@ -191,18 +195,27 @@ def save_all_viewables_and_return_info(
         "object_type": geode_functions.get_object_type(geode_object),
         "binary_light_viewable": binary_light_viewable.decode("utf-8"),
         "geode_object": data_entry.geode_object,
-        "input_files": data_entry.additional_files,
+        "input_files": data_entry.input_file,
+        "additional_files": data_entry.additional_files,
     }
 
 
-def generate_native_viewable_and_light_viewable_from_object(geode_object, data):
-    return save_all_viewables_and_return_info(geode_object, data)
+def generate_native_viewable_and_light_viewable_from_object(
+    geode_object: str, 
+    data: Any
+) -> Dict[str, Any]:
+    return save_all_viewables_and_return_info(geode_object, data, input_file=[])
 
 
-def generate_native_viewable_and_light_viewable_from_file(geode_object, input_filename):
-    temp_data_entry = Data.create_and_get_id(
+def generate_native_viewable_and_light_viewable_from_file(
+    geode_object: str, 
+    input_filename: str
+) -> Dict[str, Any]:
+    temp_data_entry = Data.create(
         name="temp",
-        geode_object=geode_object
+        geode_object=geode_object,
+        input_file=[input_filename],
+        additional_files=[],
     )
 
     data_path = create_data_folder_from_id(temp_data_entry.id)
@@ -213,7 +226,7 @@ def generate_native_viewable_and_light_viewable_from_file(geode_object, input_fi
     )
     shutil.copy2(full_input_filename, copied_full_path)
 
-    additional_files_copied = []
+    additional_files_copied: List[str] = []
     additional = geode_functions.additional_files(geode_object, full_input_filename)
     for additional_file in additional.mandatory_files + additional.optional_files:
         if additional_file.is_missing:
@@ -236,5 +249,6 @@ def generate_native_viewable_and_light_viewable_from_file(geode_object, input_fi
     return save_all_viewables_and_return_info(
         geode_object,
         data,
+        input_file=[input_filename],
         additional_files=additional_files_copied,
     )

tests/test_utils_functions.py

@@ -119,7 +119,7 @@ def test_save_all_viewables_and_return_info(client):
         assert result["geode_object"] == geode_object
         assert result["input_files"] == input_file  
 
-        db_entry = Data.query.get(result["id"])
+        db_entry = database.session.get(Data, result["id"])
         assert db_entry is not None
         assert db_entry.name == data.name()
         assert db_entry.native_file_name == result["native_file_name"]
@@ -132,6 +132,25 @@ def test_save_all_viewables_and_return_info(client):
         assert os.path.exists(expected_data_path)
 
 
+def test_save_all_viewables_commits_to_db_properly(client):
+    app = client.application
+    with app.app_context():
+        geode_object = "BRep"
+        data = geode_functions.load(geode_object, "./tests/data/test.og_brep")
+        input_file = ["test.og_brep"]
+        result = utils_functions.save_all_viewables_and_return_info(
+            geode_object, data, input_file
+        )
+        data_id = result["id"]
+        db_entry_before = database.session.get(Data,data_id)
+        assert db_entry_before is not None
+        assert db_entry_before.native_file_name == result["native_file_name"]
+        database.session.rollback()
+        db_entry_after = database.session.get(Data, data_id)
+        assert db_entry_after is not None, "database.session.commit() was not called - entry missing after rollback"
+        assert db_entry_after.native_file_name == result["native_file_name"]
+
+
 def test_generate_native_viewable_and_light_viewable_from_object(client):
     app = client.application
     with app.app_context():