chore(deps): upgrade SQLAlchemy from 1.4.41 to 2.0.45

This PR upgrades SQLAlchemy to version 2.0.45, addressing breaking changes:

- Update DeclarativeBase: Replace deprecated declarative_base() with new
  class-based DeclarativeBase pattern
- Fix raw SQL execution: Wrap string SQL with text() in mod_health and tests
- Update DeclEnumType: Use String(50) as impl instead of Enum to avoid
  SQLAlchemy 2.0's stricter enum value validation in TypeDecorator
- Fix engine creation: Remove deprecated convert_unicode parameter and
  use StaticPool for SQLite in-memory databases to ensure connection sharing
- Update type hints: Use generic Dialect type instead of SQLite-specific

All 402 tests pass with the upgraded SQLAlchemy version.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: cfsmp3

database.py

@@ -4,15 +4,14 @@
 import re
 import traceback
 from abc import ABCMeta
-from typing import Any, Dict, Iterator, Tuple, Type, Union
+from typing import Any, Dict, Iterator, Optional, Tuple, Type, Union
 
 from sqlalchemy import create_engine
-from sqlalchemy.dialects.sqlite.pysqlite import SQLiteDialect_pysqlite
+from sqlalchemy.engine import Dialect
 from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
-from sqlalchemy.orm import scoped_session, sessionmaker
-from sqlalchemy.sql.schema import Column, Table
-from sqlalchemy.sql.sqltypes import Enum, SchemaType, TypeDecorator
+from sqlalchemy.orm import DeclarativeBase, DeclarativeMeta, scoped_session, sessionmaker
+from sqlalchemy.pool import StaticPool
+from sqlalchemy.sql.sqltypes import String, TypeDecorator
 
 from exceptions import EnumParsingException, FailedToSpawnDBSession
 
@@ -23,8 +22,13 @@ class DeclarativeABCMeta(DeclarativeMeta, ABCMeta):
     pass
 
 
-Base = declarative_base(metaclass=DeclarativeMeta)
-Base.query = None
+class Base(DeclarativeBase):
+    """Base class for all models."""
+
+    pass
+
+
+Base.query = None  # type: ignore[attr-defined]
 db_engine = None
 
 
@@ -43,9 +47,18 @@ def create_session(db_string: str, drop_tables: bool = False) -> scoped_session:
     global db_engine, Base
 
     try:
-        # In testing, we want to maintain same memory variable
-        if db_engine is None or 'TESTING' not in os.environ or os.environ['TESTING'] == 'False':
-            db_engine = create_engine(db_string, convert_unicode=True)
+        # Only create engine if it doesn't exist
+        # For SQLite in-memory, we must reuse the engine to share the database
+        if db_engine is None:
+            # For SQLite in-memory databases, use StaticPool to share connection
+            if db_string == 'sqlite:///:memory:':
+                db_engine = create_engine(
+                    db_string,
+                    connect_args={"check_same_thread": False},
+                    poolclass=StaticPool
+                )
+            else:
+                db_engine = create_engine(db_string)
         db_session = scoped_session(sessionmaker(bind=db_engine))
         Base.query = db_session.query_property()
 
@@ -162,32 +175,27 @@ def db_type(cls) -> DeclEnumType:
         return DeclEnumType(cls)
 
 
-class DeclEnumType(SchemaType, TypeDecorator):
+class DeclEnumType(TypeDecorator):
     """Declarative enumeration type."""
 
     cache_ok = True
+    impl = String(50)
 
     def __init__(self, enum: Any) -> None:
         self.enum = enum
-        self.impl = Enum(
-            *enum.values(),
-            name="ck{0}".format(re.sub('([A-Z])', lambda m: "_" + m.group(1).lower(), enum.__name__))
-        )
-
-    def _set_table(self, table: Column, column: Table) -> None:
-        self.impl._set_table(table, column)
+        super().__init__()
 
-    def copy(self) -> DeclEnumType:
+    def copy(self, **kwargs: Any) -> DeclEnumType:
         """Get enumeration type of self."""
         return DeclEnumType(self.enum)
 
-    def process_bind_param(self, value: EnumSymbol, dialect: SQLiteDialect_pysqlite) -> str:
+    def process_bind_param(self, value: Optional[Any], dialect: Dialect) -> Optional[str]:
         """Get process bind parameter."""
         if value is None:
             return None
         return value.value
 
-    def process_result_value(self, value: str, dialect: SQLiteDialect_pysqlite) -> EnumSymbol:
+    def process_result_value(self, value: Optional[Any], dialect: Dialect) -> Optional[EnumSymbol]:
         """Get process result value."""
         if value is None:
             return None

mod_health/controllers.py

@@ -6,6 +6,7 @@
 from typing import Any, Dict, Optional, Tuple
 
 from flask import Blueprint, current_app, jsonify
+from sqlalchemy import text
 
 mod_health = Blueprint('health', __name__)
 
@@ -20,7 +21,7 @@ def check_database() -> Dict[str, Any]:
     try:
         from database import create_session
         db = create_session(current_app.config['DATABASE_URI'])
-        db.execute('SELECT 1')
+        db.execute(text('SELECT 1'))
         # remove() returns the scoped session's connection to the pool
         db.remove()
         return {'status': 'ok'}

requirements.txt

@@ -1,4 +1,4 @@
-sqlalchemy==1.4.41
+sqlalchemy==2.0.45
 flask==3.1.2
 passlib==1.7.4
 pymysql==1.1.2

tests/base.py

@@ -7,6 +7,7 @@
 
 from flask import g
 from flask_testing import TestCase
+from sqlalchemy import text
 from werkzeug.datastructures import Headers
 
 from database import create_session
@@ -250,7 +251,7 @@ def setUp(self):
         g.db = create_session(
             self.app.config['DATABASE_URI'], drop_tables=True)
         # enable Foreign keys for unit tests
-        g.db.execute('pragma foreign_keys=on')
+        g.db.execute(text('pragma foreign_keys=on'))
 
         general_data = [
             GeneralData('last_commit', "1978060bf7d2edd119736ba3ba88341f3bec3323"),