NullPool for one connection - one request

custom apply_driver_hacks: see pallets-eco/flask-sqlalchemy#803

apps/__init__.py

@@ -1,12 +1,14 @@
 from flask import Flask
 from flask_migrate import Migrate
-from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy.pool import NullPool
 
 from config import Config
 
+from .engine import SQLAlchemy
+
 app = Flask(__name__)
 app.config.from_object(Config)
-db = SQLAlchemy(app)
+db = SQLAlchemy(app, engine_options={"poolclass": NullPool})
 migrate = Migrate(app, db)
 
 

apps/engine.py

@@ -1,27 +1,56 @@
-from datetime import datetime
+import os
 
-from sqlalchemy.orm.query import Query
+from flask_sqlalchemy import SQLAlchemy as FlaskSQLAlchemy
+from flask_sqlalchemy import _sa_url_query_setdefault, _sa_url_set
+from sqlalchemy.pool import NullPool
 
-from . import db
 
+class SQLAlchemy(FlaskSQLAlchemy):
+    def apply_driver_hacks(self, app, sa_url, options):
+        if sa_url.drivername.startswith("mysql"):
+            sa_url = _sa_url_query_setdefault(sa_url, charset="utf8")
 
-class Model(db.Model):
-    __abstract__ = True
-    query: Query
+            if sa_url.drivername != "mysql+gaerdbms" and isinstance(
+                options.get("poolclass"), NullPool
+            ):
+                options.setdefault("pool_size", 10)
+                options.setdefault("pool_recycle", 7200)
+        elif sa_url.drivername == "sqlite":
+            pool_size = options.get("pool_size")
+            detected_in_memory = False
+            if sa_url.database in (None, "", ":memory:"):
+                detected_in_memory = True
+                from sqlalchemy.pool import StaticPool
 
+                options["poolclass"] = StaticPool
+                if "connect_args" not in options:
+                    options["connect_args"] = {}
+                options["connect_args"]["check_same_thread"] = False
 
-class BaseModel(Model):
-    __abstract__ = True
+                # we go to memory and the pool size was explicitly set
+                # to 0 which is fail.  Let the user know that
+                if pool_size == 0:
+                    raise RuntimeError(
+                        "SQLite in memory database with an "
+                        "empty queue not possible due to data "
+                        "loss."
+                    )
+            # if pool size is None or explicitly set to 0 we assume the
+            # user did not want a queue for this sqlite connection and
+            # hook in the null pool.
+            elif not pool_size:
+                options["poolclass"] = NullPool
 
-    id = db.Column(db.Integer, primary_key=True)
-    create_at = db.Column(
-        db.DateTime,
-        default=datetime.now(),
-        comment="Дата создания",
-    )
-    update_at = db.Column(
-        db.DateTime,
-        default=datetime.now(),
-        onupdate=datetime.now(),
-        comment="Дата обновления",
-    )
+            # if it's not an in memory database we make the path absolute.
+            if not detected_in_memory:
+                sa_url = _sa_url_set(
+                    sa_url, database=os.path.join(app.root_path, sa_url.database)
+                )
+
+        unu = app.config["SQLALCHEMY_NATIVE_UNICODE"]
+        if unu is None:
+            unu = self.use_native_unicode
+        if not unu:
+            options["use_native_unicode"] = False
+
+        return sa_url, options

apps/gkgn/models.py

@@ -3,7 +3,7 @@
 from sqlalchemy.orm import relationship
 
 from apps import db
-from apps.engine import BaseModel, Model
+from apps.models import BaseModel, Model
 
 __all__ = ("Settlement", "ATE", "Type", "Object")
 

apps/models.py

@@ -0,0 +1,27 @@
+from datetime import datetime
+
+from sqlalchemy.orm.query import Query
+
+from . import db
+
+
+class Model(db.Model):
+    __abstract__ = True
+    query: Query
+
+
+class BaseModel(Model):
+    __abstract__ = True
+
+    id = db.Column(db.Integer, primary_key=True)
+    create_at = db.Column(
+        db.DateTime,
+        default=datetime.now(),
+        comment="Дата создания",
+    )
+    update_at = db.Column(
+        db.DateTime,
+        default=datetime.now(),
+        onupdate=datetime.now(),
+        comment="Дата обновления",
+    )

import_gkgn.py

@@ -5,7 +5,9 @@
 import sqlalchemy as sa
 from sqlalchemy import select, update
 from sqlalchemy.orm import Session, declarative_base, relationship, sessionmaker
+
 from config import Config
+
 from apps.gkgn.constants import LevelEnum
 
 engine = sa.create_engine(Config.SQLALCHEMY_DATABASE_URI)