Using Flask-SQLAlchemy instead of SQLAlchemy for the database

Author: MaxNumerique

.gitignore

@@ -9,4 +9,5 @@ __pycache__
 .vscode
 uploads
 node_modules
-schemas.json
+schemas.json
+*.db

app.py

@@ -10,6 +10,7 @@
 from src.opengeodeweb_back.routes.models import blueprint_models
 from src.opengeodeweb_back.utils_functions import handle_exception
 from src.opengeodeweb_back import app_config
+from src.opengeodeweb_back.database import init_db
 
 
 """ Global config """
@@ -58,4 +59,5 @@ def return_error():
 # ''' Main '''
 if __name__ == "__main__":
     print(f"Python is running in {FLASK_DEBUG} mode")
+    init_db(app)
     app.run(debug=FLASK_DEBUG, host=DEFAULT_HOST, port=PORT, ssl_context=SSL)

src/opengeodeweb_back/app_config.py

@@ -15,6 +15,7 @@ class Config(object):
     REQUEST_COUNTER = 0
     LAST_REQUEST_TIME = time.time()
     LAST_PING_TIME = time.time()
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
 
 
 class ProdConfig(Config):
@@ -23,6 +24,7 @@ class ProdConfig(Config):
     MINUTES_BEFORE_TIMEOUT = "1"
     SECONDS_BETWEEN_SHUTDOWNS = "10"
     DATA_FOLDER_PATH = "/data/"
+    SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(DATA_FOLDER_PATH, 'db.sqlite3')}"
 
 
 class DevConfig(Config):
@@ -31,3 +33,4 @@ class DevConfig(Config):
     MINUTES_BEFORE_TIMEOUT = "1"
     SECONDS_BETWEEN_SHUTDOWNS = "10"
     DATA_FOLDER_PATH = "./data/"
+    SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(DATA_FOLDER_PATH, 'db.sqlite3')}"

src/opengeodeweb_back/database.py

@@ -2,9 +2,15 @@
 
 db = SQLAlchemy()
 
+initialized = False
 
 def init_db(app):
+    global initialized
+    if initialized:
+        return db 
+    print('DB', app.config.get("SQLALCHEMY_DATABASE_URI"))
     db.init_app(app)
     with app.app_context():
         db.create_all()
+    initialized = True
     return db

src/opengeodeweb_back/models.py

@@ -15,9 +15,7 @@ class Data(db.Model):
     name = db.Column(String, nullable=False)
     native_file_name = db.Column(String, nullable=False)
     viewable_file_name = db.Column(String, nullable=False)
-    light_viewable = db.Column(
-        String, nullable=True
-    )  # Renommé pour correspondre au code
+    light_viewable = db.Column(String, nullable=True)
     geode_object = db.Column(String, nullable=False)
     input_files = db.Column(JSON, nullable=True)
     created_at = db.Column(DateTime, default=datetime.now)

tests/conftest.py

@@ -1,12 +1,12 @@
 # Standard library imports
 import time
 import shutil
-
-# Third party imports
+import os
 import pytest
 
-# Local application imports
+
 from app import app
+from src.opengeodeweb_back.database import init_db
 
 TEST_ID = "1"
 
@@ -25,6 +25,14 @@ def client():
     app.config["UPLOAD_FOLDER"] = "./tests/data/"
     app.config["REQUEST_COUNTER"] = 0
     app.config["LAST_REQUEST_TIME"] = time.time()
+    BASE_DIR = os.path.abspath(os.path.dirname(__file__))
+    db_path = os.path.join(BASE_DIR, "data", "project.db") 
+    app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
+
+    print("Current working directory:", os.getcwd())
+    print("Directory contents:", os.listdir("."))
+
+    init_db(app)
     client = app.test_client()
     client.headers = {"Content-type": "application/json", "Accept": "application/json"}
     yield client

tests/test_utils_functions.py

@@ -7,6 +7,8 @@
 import shutil
 
 # Local application imports
+from src.opengeodeweb_back.database import db
+from src.opengeodeweb_back.models import Data
 from src.opengeodeweb_back import geode_functions, utils_functions
 
 
@@ -97,15 +99,24 @@ def test_save_all_viewables_and_return_info(client):
             geode_object, data, generated_id, data_path, additional_files
         )
 
-    assert isinstance(result, dict)
-    assert result["name"] == data.name()
-    assert result["native_file_name"].startswith("native.")
-    assert result["viewable_file_name"].endswith(".vtm")
-    assert re.match(r"[0-9a-f]{32}", result["id"])
-    assert isinstance(result["object_type"], str)
-    assert isinstance(result["binary_light_viewable"], str)
-    assert result["geode_object"] == geode_object
-    assert result["input_files"] == additional_files
+        assert isinstance(result, dict)
+        assert result["name"] == data.name()
+        assert result["native_file_name"].startswith("native.")
+        assert result["viewable_file_name"].endswith(".vtm")
+        assert re.match(r"[0-9a-f]{32}", result["id"])
+        assert isinstance(result["object_type"], str)
+        assert isinstance(result["binary_light_viewable"], str)
+        assert result["geode_object"] == geode_object
+        assert result["input_files"] == additional_files
+
+        db_entry = Data.query.get(generated_id)
+        assert db_entry is not None
+        assert db_entry.name == data.name()
+        assert db_entry.native_file_name == os.path.basename(result["native_file_name"])
+        assert db_entry.viewable_file_name == os.path.basename(result["viewable_file_name"])
+        assert db_entry.light_viewable == os.path.basename(db_entry.light_viewable)
+        assert db_entry.geode_object == geode_object
+        assert db_entry.input_files == additional_files
 
 
 def test_generate_native_viewable_and_light_viewable_from_object(client):