sql alchemy replacing flask sql alchemy

Author: MaxNumerique

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "OpenGeodeWeb-Microservice"
-version = "0.0.0"
+version = "1.0.0"
 dynamic = ["dependencies"]
 authors = [{ name = "Geode-solutions", email = "team-web@geode-solutions.com" }]
 description = "Database model and ORM layer for OpenGeodeWeb ecosystem"

requirements.txt

@@ -1,3 +1,2 @@
 SQLAlchemy==2.0.43
-Flask-SQLAlchemy==3.1.1
 fastjsonschema==2.21.1

src/opengeodeweb_microservice/database/connection.py

@@ -1,29 +1,29 @@
 """Database connection management"""
 
-from typing import Optional
-from sqlalchemy.orm import scoped_session
-from flask import Flask
-from flask_sqlalchemy import SQLAlchemy
-from flask_sqlalchemy.session import Session
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, scoped_session
 from .base import Base
 
 DATABASE_FILENAME = "project.db"
-database: Optional[SQLAlchemy] = None
 
+engine = None
+session_factory = None
+scoped_session_registry = None
 
-def init_database(app: Flask, db_filename: str = DATABASE_FILENAME) -> SQLAlchemy:
-    global database
-    if database is None:
-        database = SQLAlchemy(model_class=Base)
-    database.init_app(app)
-    with app.app_context():
-        database.create_all()
-    return database
 
+def init_database(db_path: str = DATABASE_FILENAME) -> None:
+    global engine, session_factory, scoped_session_registry
 
-def get_database() -> Optional[SQLAlchemy]:
-    return database
+    if engine is None:
+        engine = create_engine(
+            f"sqlite:///{db_path}", connect_args={"check_same_thread": False}
+        )
+        session_factory = sessionmaker(bind=engine)
+        scoped_session_registry = scoped_session(session_factory)
+        Base.metadata.create_all(engine)
 
 
-def get_session() -> Optional[scoped_session[Session]]:
-    return database.session if database else None
+def get_session():
+    if scoped_session_registry is None:
+        raise RuntimeError()
+    return scoped_session_registry

src/opengeodeweb_microservice/database/data.py

@@ -25,8 +25,8 @@ def create(
         input_file: str | None = None,
         additional_files: list[str] | None = None,
     ) -> "Data":
-        input_file = input_file if input_file is not None else ""
-        additional_files = additional_files if additional_files is not None else []
+        input_file = input_file or ""
+        additional_files = additional_files or []
 
         data_entry = Data(
             geode_object=geode_object,
@@ -38,15 +38,12 @@ def create(
         )
 
         session = get_session()
-        if session:
-            session.add(data_entry)
-            session.flush()
-            session.commit()
+        session.add(data_entry)
+        session.flush()
+        session.commit()
         return data_entry
 
     @staticmethod
     def get(data_id: str) -> "Data | None":
         session = get_session()
-        if session:
-            return session.get(Data, data_id)
-        return None
+        return session.get(Data, data_id)

tests/conftest.py

@@ -1,56 +1,40 @@
 import os
 import pytest
-from flask import Flask
 from src.opengeodeweb_microservice.database.connection import init_database, get_session
 from src.opengeodeweb_microservice.database.data import Data
 
 
-@pytest.fixture(scope="session")
-def app():
-    app = Flask(__name__)
-    app.config.update(
-        {
-            "TESTING": True,
-            "SQLALCHEMY_DATABASE_URI": f"sqlite:///{os.path.join(os.path.dirname(__file__), 'test_project.db')}",
-            "SQLALCHEMY_TRACK_MODIFICATIONS": False,
-        }
-    )
-    with app.app_context():
-        init_database(app, "test_project.db")
-        yield app
-        _cleanup_database()
+DB_PATH = os.path.join(os.path.dirname(__file__), "test_project.db")
 
 
-def _cleanup_database():
+@pytest.fixture(scope="session", autouse=True)
+def setup_database():
+    init_database(DB_PATH)
+    yield
+    _cleanup_database(DB_PATH)
+
+
+def _cleanup_database(db_path: str):
     try:
         session = get_session()
-        if session:
-            session.close()
+        session.close()
     except Exception:
         pass
-    db_path = os.path.join(os.path.dirname(__file__), "test_project.db")
+
     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):
+@pytest.fixture(autouse=True)
+def clean_database():
     session = get_session()
-    if session:
-        session.query(Data).delete()
-        session.commit()
+    session.query(Data).delete()
+    session.commit()
     yield
-    if session:
-        try:
-            session.rollback()
-        except Exception:
-            pass
+    try:
+        session.rollback()
+    except Exception:
+        pass

tests/test_connection.py

@@ -1,23 +1,21 @@
-from src.opengeodeweb_microservice.database.connection import get_session
 from src.opengeodeweb_microservice.database.data import Data
 
 
-def test_data_crud_operations():
+def test_data_crud_operations(clean_database):
     data = Data.create(geode_object="test_object", input_file="test.txt")
     assert data.id is not None
-    session = get_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():
+def test_data_with_additional_files(clean_database):
     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 is not None
     assert retrieved.additional_files == files