refacto

Author: MaxNumerique

src/opengeodeweb_microservice/app_config.py

@@ -1,6 +1,8 @@
 import os
-from .database.session import DATABASE_FILENAME
+from .database.connection import DATABASE_FILENAME
 
+class Config:
+    pass
 
 class ProdConfig(Config):
     DATA_FOLDER_PATH = "/data"

src/opengeodeweb_microservice/database/connection.py

@@ -3,26 +3,28 @@
 from typing import Optional
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
+from ..microservice.base import Base
 
-from src.opengeodeweb_microservice.microservice.base import Base
+DATABASE_FILENAME = "project.db"
 
+db: Optional[SQLAlchemy] = None
 
-class DatabaseConnection:
+def init_database(app: Flask, db_filename: str = DATABASE_FILENAME) -> SQLAlchemy:
+    global db
+    if db is None:
+        db = SQLAlchemy(model_class=Base)
+    db.init_app(app)
+    with app.app_context():
+        db.create_all()
+    return db
 
-    def __init__(self, database_filename: str = "project.db"):
-        self.database_filename = database_filename
-        self.database: Optional[SQLAlchemy] = None
+def get_database() -> Optional[SQLAlchemy]:
+    return db
 
-    def init_app(self, app: Flask) -> SQLAlchemy:
-        if self.database is None:
-            self.database = SQLAlchemy(model_class=Base)
+def get_session():
+    if db is None:
+        return None
+    return db.session
 
-        self.database.init_app(app)
-
-        with app.app_context():
-            self.database.create_all()
-
-        return self.database
-
-    def get_database(self) -> Optional[SQLAlchemy]:
-        return self.database
+def get_database_connection():
+    return get_database()

src/opengeodeweb_microservice/database/session.py

@@ -1,10 +1,9 @@
 from sqlalchemy import String, JSON
 from sqlalchemy.orm import Mapped, mapped_column
-from ..database.session import get_session
+from ..database.connection import get_session
 from .base import Base
 import uuid
 
-
 class Data(Base):
     __tablename__ = "datas"
 
@@ -38,11 +37,14 @@ def create(
         )
 
         session = get_session()
-        session.session.add(data_entry)
-        session.session.flush()
+        if session:
+            session.add(data_entry)
+            session.flush()
         return data_entry
 
     @staticmethod
     def get(data_id: str) -> "Data | None":
         session = get_session()
-        return session.session.get(Data, data_id)
+        if session:
+            return session.get(Data, data_id)
+        return None

src/opengeodeweb_microservice/microservice/data.py

@@ -1,10 +1,9 @@
 import os
 import pytest
 from flask import Flask
-from src.opengeodeweb_microservice.database.session import init_database, get_session
+from src.opengeodeweb_microservice.database.connection import init_database, get_session
 from src.opengeodeweb_microservice.microservice.data import Data
 
-
 @pytest.fixture(scope="session")
 def app():
     app = Flask(__name__)
@@ -19,41 +18,35 @@ def app():
         _cleanup_database_connections()
     _remove_test_database(db_path)
 
-
 def _cleanup_database_connections():
     try:
         session = get_session()
-        if session and hasattr(session, "session"):
-            session.session.close()
-        if session and hasattr(session, "engine"):
-            session.engine.dispose()
+        if session:
+            session.close()
     except Exception:
         pass
 
-
 def _remove_test_database(db_path: str):
     if os.path.exists(db_path):
         try:
             os.remove(db_path)
         except PermissionError:
             pass
 
-
 @pytest.fixture
 def app_context(app):
     with app.app_context():
         yield
 
-
 @pytest.fixture
 def clean_database(app_context):
     session = get_session()
     if session:
-        session.session.query(Data).delete()
-        session.session.commit()
+        session.query(Data).delete()
+        session.commit()
     yield
     if session:
         try:
-            session.session.rollback()
+            session.rollback()
         except Exception:
             pass

tests/conftest.py

@@ -1,35 +1,11 @@
-import pytest
-from src.opengeodeweb_microservice.database.session import (
+from src.opengeodeweb_microservice.database.connection import (
     get_session,
     get_database_connection,
 )
 from src.opengeodeweb_microservice.microservice.data import Data
 
-
 def test_database_connection_basic(app_context):
     session = get_session()
     assert session is not None
-    assert session.session is not None
     connection = get_database_connection()
     assert connection is not None
-
-
-def test_data_crud_operations(app_context, clean_database):
-    data = Data.create(geode_object="test_object", input_file="test.txt")
-    assert data.id is not None
-    session = get_session()
-    session.session.commit()
-    retrieved = Data.get(data.id)
-    assert retrieved is not None
-    assert retrieved.geode_object == "test_object"
-    non_existent = Data.get("fake_id")
-    assert non_existent is None
-
-
-def test_data_with_additional_files(app_context, clean_database):
-    files = ["file1.txt", "file2.txt"]
-    data = Data.create(geode_object="test_files", additional_files=files)
-    session = get_session()
-    session.session.commit()
-    retrieved = Data.get(data.id)
-    assert retrieved.additional_files == files

tests/test_connection.py

@@ -1,28 +1,21 @@
-import pytest
-from src.opengeodeweb_microservice.database.session import get_session
+from src.opengeodeweb_microservice.database.connection import get_session
 from src.opengeodeweb_microservice.microservice.data import Data
 
-
-def test_database_session_available(app_context):
-    session = get_session()
-    assert session is not None
-    assert session.session is not None
-
-
-def test_data_crud_operations(app_context, clean_database):
-    data = Data.create(
-        geode_object="test_object",
-        input_file="test.txt",
-        additional_files=["file1.txt", "file2.txt"],
-    )
+def test_data_crud_operations():
+    data = Data.create(geode_object="test_object", input_file="test.txt")
     assert data.id is not None
-    assert data.geode_object == "test_object"
-    assert data.additional_files == ["file1.txt", "file2.txt"]
     session = get_session()
-    session.session.commit()
+    session.commit()
     retrieved = Data.get(data.id)
     assert retrieved is not None
     assert retrieved.geode_object == "test_object"
-    assert retrieved.additional_files == ["file1.txt", "file2.txt"]
     non_existent = Data.get("fake_id")
     assert non_existent is None
+
+def test_data_with_additional_files():
+    files = ["file1.txt", "file2.txt"]
+    data = Data.create(geode_object="test_files", additional_files=files)
+    session = get_session()
+    session.commit()
+    retrieved = Data.get(data.id)
+    assert retrieved.additional_files == files