sqlalchemy2

docker-compose.yml

@@ -20,8 +20,6 @@ x-flask-defaults: &flask-defaults
   environment:
     # Set this variable in .env to start the app with a different config file (default: config.yaml)
     CONFIG_FILE:
-    # TODO: This should be removed at some point and the app should be made SQLAlchemy 2.0 compatible!
-    SQLALCHEMY_SILENCE_UBER_WARNING: 1
     OCPDB_POSTGRES_DB: ocpdb
     OCPDB_POSTGRES_USER: ocpdb
     OCPDB_POSTGRES_PASSWORD: admin
@@ -71,7 +69,7 @@ services:
       retries: 20
 
   postgre:
-    image: postgis/postgis
+    image: postgis/postgis:15-3.5-alpine
     volumes:
       - postgres:/var/lib/postgresql/data/
     environment:

requirements-dev.txt

@@ -1,4 +1,4 @@
 pytest~=8.3.5
-pytest-cov~=6.0.0
+pytest-cov~=6.1.1
 requests-mock~=1.12.1
 ruff~=0.11.5

requirements.txt

@@ -1,12 +1,11 @@
 gunicorn~=23.0.0
 Flask~=3.1.0
 Flask-Failsafe~=0.2.0
-# Flask-SQLAlchemy 3.1 requires SQLAlchemy 2.0
-Flask-SQLAlchemy~=3.0.5
+Flask-SQLAlchemy~=3.1.1
 Flask-Celery-Helper~=1.1.0
-Flask-Migrate~=4.0.7
+Flask-Migrate~=4.1.0
 Flask-CORS~=5.0.1
-SQLAlchemy~=1.4.54
+SQLAlchemy~=2.0.40
 requests~=2.32.3
 alembic~=1.15.2
 lxml~=5.3.2

tests/integration/conftest.py

@@ -16,38 +16,51 @@
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
+import re
+from typing import Generator
+
 import pytest
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, text
 
+from tests.integration.helpers import empty_all_tables
 from webapp import launch
 from webapp.common.flask_app import App
 from webapp.common.sqlalchemy import SQLAlchemy
 from webapp.extensions import db as flask_sqlalchemy
 
 
-@pytest.fixture
-def app() -> App:
-    test_app = launch()
-    test_app.config.update(
-        TESTING=True,
-        DEBUG=True,
+@pytest.fixture(scope='session')
+def flask_app() -> Generator[App, None, None]:
+    app = launch(
+        config_overrides={
+            'TESTING': True,
+            'DEBUG': True,
+        }
     )
-    with test_app.app_context():
-        yield test_app
-
 
-@pytest.fixture
-def db(app: App) -> SQLAlchemy:
     # Create the database and the database tables
-
     # db_path should be 'mysql+pymysql://root:root@mysql' if
-    # SQLALCHEMY_DATABASE_URI: 'mysql+pymysql://root:root@mysql/ocpdb' is set in test_config.yaml
-    db_path: str = app.config.get('SQLALCHEMY_DATABASE_URI')[:-6]
+    # SQLALCHEMY_DATABASE_URI: 'mysql+pymysql://root:root@mysql/backend?charset=utf8mb4' is set in test_config.yaml
+    db_path: str = re.sub(r'/[^/]+$', '', app.config.get('SQLALCHEMY_DATABASE_URI'))
 
     engine = create_engine(db_path)
-    connection = engine.connect()
-    connection.execute('DROP DATABASE IF EXISTS ocpdb;')
-    connection.execute('CREATE DATABASE IF NOT EXISTS ocpdb;')
-    flask_sqlalchemy.create_all()
+
+    # We use DROP + CREATE here because it's faster and more reliable in case of foreign keys
+    with engine.connect() as connection:
+        connection.execute(text('DROP DATABASE IF EXISTS `post-salad-backend`;'))
+        connection.execute(text('CREATE DATABASE IF NOT EXISTS `post-salad-backend`;'))
+
+    with app.app_context():
+        flask_sqlalchemy.create_all()
+
+        yield app  # type: ignore
+
+
+@pytest.fixture
+def db(flask_app: App) -> Generator[SQLAlchemy, None, None]:
+    """
+    Yields the database as a function-scoped fixture with freshly emptied tables.
+    """
+    empty_all_tables(db=flask_sqlalchemy)
 
     yield flask_sqlalchemy

tests/integration/helpers.py

@@ -0,0 +1,34 @@
+"""
+Open ChargePoint DataBase OCPDB
+Copyright (C) 2025 binary butterfly GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from sqlalchemy import text
+
+from webapp.common.sqlalchemy import SQLAlchemy
+
+
+def empty_all_tables(db: SQLAlchemy) -> None:
+    """
+    empty all tables in the database
+    (this is much faster than completely deleting the database and creating a new one)
+    """
+    db.session.close()
+    with db.engine.connect() as connection:
+        connection.execute(text('SET FOREIGN_KEY_CHECKS=0;'))
+        for table_name in db.metadata.tables.keys():
+            connection.execute(text(f'TRUNCATE `{table_name}`;'))
+        connection.execute(text('SET FOREIGN_KEY_CHECKS=1;'))

webapp/app.py

@@ -34,19 +34,19 @@
 __all__ = ['launch']
 
 
-def launch() -> App:
+def launch(config_overrides: dict | None = None) -> App:
     app = App(BaseConfig.PROJECT_NAME)
-    configure_app(app)
+    configure_app(app, config_overrides)
     configure_extensions(app)
     configure_blueprints(app)
     configure_error_handlers(app)
     configure_periodic_tasks()
     return app
 
 
-def configure_app(app: App) -> None:
+def configure_app(app: App, config_overrides: dict | None = None) -> None:
     config_loader = ConfigLoader()
-    config_loader.configure_app(app)
+    config_loader.configure_app(app, config_overrides)
 
 
 def configure_extensions(app: App) -> None:

webapp/common/config/config_loader.py

@@ -27,7 +27,7 @@
 
 class ConfigLoader:
     @staticmethod
-    def configure_app(app: Flask) -> None:
+    def configure_app(app: Flask, config_overrides: None = None) -> None:
         """
         Initializes the app config with default values and loads the actual config from a YAML file.
         """
@@ -65,6 +65,9 @@ def configure_app(app: Flask) -> None:
             for key, server in app.config['REMOTE_SERVERS'].items()
         }
 
+        if config_overrides is not None:
+            app.config.update(config_overrides)
+
         # Ensure that important config values are set
         config_check = [key for key in app.config['ENFORCE_CONFIG_VALUES'] if key not in app.config]
         if len(config_check) > 0:

webapp/common/sqlalchemy/__init__.py

@@ -18,4 +18,3 @@
 
 from .query import Query
 from .sqlalchemy import SQLAlchemy
-from .typing import Mapped

webapp/dependencies.py

@@ -21,7 +21,7 @@
 
 from butterfly_pubsub.sync import PubSubClient
 from flask import current_app
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import scoped_session
 
 from webapp.common.celery import CeleryHelper
 from webapp.common.config import ConfigHelper
@@ -125,7 +125,7 @@ def get_server_auth_helper(self) -> 'ServerAuthHelper':
 
     # Database
     @cache_dependency
-    def get_db_session(self) -> Session:
+    def get_db_session(self) -> scoped_session:
         # Late import (don't initialize all the extensions unless needed)
         from webapp.extensions import db
 

webapp/models/base.py

@@ -17,31 +17,34 @@
 """
 
 from datetime import datetime, timezone
-from typing import List, Optional
 
+from sqlalchemy import BigInteger
+from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy.types import UserDefinedType
 from sqlalchemy_utc import UtcDateTime
 
-from webapp.common.sqlalchemy import Mapped
 from webapp.extensions import db
 
 
-class BaseModel:
+class BaseModel(db.Model):
+    __abstract__ = True
     __table_args__ = {
         'mysql_charset': 'utf8mb4',
         'mysql_collate': 'utf8mb4_unicode_ci',
     }
 
-    id: Mapped[int] = db.Column(db.BigInteger, primary_key=True)
-    created: Mapped[datetime] = db.Column(UtcDateTime(), nullable=False, default=lambda: datetime.now(tz=timezone.utc))
-    modified: Mapped[datetime] = db.Column(
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, nullable=False)
+    created: Mapped[datetime] = mapped_column(
+        UtcDateTime(), nullable=False, default=lambda: datetime.now(tz=timezone.utc)
+    )
+    modified: Mapped[datetime] = mapped_column(
         UtcDateTime(),
         nullable=False,
         default=datetime.now(tz=timezone.utc),
         onupdate=datetime.now(tz=timezone.utc),
     )
 
-    def to_dict(self, fields: Optional[List[str]] = None, ignore: Optional[List[str]] = None) -> dict:
+    def to_dict(self, fields: list[str] | None = None, ignore: list[str] | None = None) -> dict:
         result = {}
         for field in self.metadata.tables[self.__tablename__].c.keys():
             if fields is not None and field not in fields:

webapp/models/business.py

@@ -18,24 +18,24 @@
 
 from typing import TYPE_CHECKING, Optional
 
-from webapp.common.sqlalchemy import Mapped
-from webapp.extensions import db
+from sqlalchemy import BigInteger, ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import BaseModel
 
 if TYPE_CHECKING:
     from .image import Image
 
 
-class Business(db.Model, BaseModel):
+class Business(BaseModel):
     __tablename__ = 'business'
 
-    logo: Mapped[Optional['Image']] = db.relationship('Image', uselist=False)
+    logo: Mapped[Optional['Image']] = relationship('Image', uselist=False)
 
-    logo_id: Mapped[int | None] = db.Column(db.BigInteger, db.ForeignKey('image.id', use_alter=True), nullable=True)
+    logo_id: Mapped[int | None] = mapped_column(BigInteger, ForeignKey('image.id', use_alter=True), nullable=True)
 
-    name: Mapped[str] = db.Column(db.String(255), index=True, nullable=False)
-    website: Mapped[str | None] = db.Column(db.String(255), nullable=True)
+    name: Mapped[str] = mapped_column(String(255), index=True, nullable=False)
+    website: Mapped[str | None] = mapped_column(String(255), nullable=True)
 
     def to_dict(self, *args, ignore: list[str] | None = None, **kwargs) -> dict:
         ignore = ignore or []

webapp/models/connector.py

@@ -21,11 +21,11 @@
 from math import sqrt
 from typing import TYPE_CHECKING
 
+from sqlalchemy import BigInteger, ForeignKey, Integer, String
+from sqlalchemy import Enum as SqlalchemyEnum
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy_utc import UtcDateTime
 
-from webapp.common.sqlalchemy import Mapped
-from webapp.extensions import db
-
 from .base import BaseModel
 
 if TYPE_CHECKING:
@@ -106,23 +106,23 @@ class PowerType(Enum):
     DC = 'DC'
 
 
-class Connector(db.Model, BaseModel):
+class Connector(BaseModel):
     __tablename__ = 'connector'
 
-    evse: Mapped['Evse'] = db.relationship('Evse', back_populates='connectors')
-    evse_id: Mapped[int] = db.Column(db.BigInteger, db.ForeignKey('evse.id', use_alter=True), nullable=False)
+    evse: Mapped['Evse'] = relationship('Evse', back_populates='connectors')
+    evse_id: Mapped[int] = mapped_column(BigInteger, ForeignKey('evse.id', use_alter=True), nullable=False)
 
-    uid: Mapped[str] = db.Column(db.String(64), nullable=False, index=True)  # OCPI: id
-    standard: Mapped[ConnectorType | None] = db.Column(db.Enum(ConnectorType), nullable=True)
-    format: Mapped[ConnectorFormat | None] = db.Column(db.Enum(ConnectorFormat), nullable=True)
+    uid: Mapped[str] = mapped_column(String(64), nullable=False, index=True)  # OCPI: id
+    standard: Mapped[ConnectorType | None] = mapped_column(SqlalchemyEnum(ConnectorType), nullable=True)
+    format: Mapped[ConnectorFormat | None] = mapped_column(SqlalchemyEnum(ConnectorFormat), nullable=True)
     # OCHP: chargePointType, OCPI: power_type
-    power_type: Mapped[PowerType | None] = db.Column(db.Enum(PowerType), nullable=True)
-    max_voltage: Mapped[int | None] = db.Column(db.Integer, nullable=True)  # OCHP: nominalVoltage, OCPI: max_voltage
-    max_amperage: Mapped[int | None] = db.Column(db.Integer, nullable=True)  # OCPI: max_amperage
+    power_type: Mapped[PowerType | None] = mapped_column(SqlalchemyEnum(PowerType), nullable=True)
+    max_voltage: Mapped[int | None] = mapped_column(Integer, nullable=True)  # OCHP: nominalVoltage, OCPI: max_voltage
+    max_amperage: Mapped[int | None] = mapped_column(Integer, nullable=True)  # OCPI: max_amperage
     # OCHP: maximumPower, OCPI: max_electric_power
-    max_electric_power: Mapped[int | None] = db.Column(db.Integer, nullable=True)
-    last_updated: Mapped[datetime | None] = db.Column(UtcDateTime(), nullable=True)
-    terms_and_conditions: Mapped[str | None] = db.Column(db.String(255), nullable=True)  # OCPI: terms_and_conditions
+    max_electric_power: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    last_updated: Mapped[datetime | None] = mapped_column(UtcDateTime(), nullable=True)
+    terms_and_conditions: Mapped[str | None] = mapped_column(String(255), nullable=True)  # OCPI: terms_and_conditions
 
     # tariff_ids TODO