Load SQLite Extension with SQLAlchemy (in Flask)

flask-sqlalchemy-sqlite.md

Load SQLite Extension with SQLAlchemy (in Flask)

Files

app/db.py

from flask_sqlalchemy import SQLAlchemy

from sqlalchemy import event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.engine import Engine


db: SQLAlchemy = SQLAlchemy()


def create_session():
    return sessionmaker(bind=db.get_engine())()


def get_engine() -> Engine:
    return db.get_engine()


def init_db() -> None:

    @event.listens_for(get_engine(), "connect")
    def receive_connect(connection, _) -> None:
        connection.enable_load_extension(True)
        connection.execute("SELECT load_extension('mod_spatialite');")
        connection.enable_load_extension(False)

resources/spatialite.py

from flask_restful import Resource

from sqlalchemy import text

from app.db import create_session


class Version(Resource):

    @classmethod
    def get(cls):

        session = create_session()
    
        try:

            result = session.execute(text("SELECT spatialite_version() as version;"))
            row = result.fetchone()
            return {"spatialite_version": row["version"]}, 200

        finally:

            if session:
                session.close()

config.py

SQLALCHEMY_DATABASE_URI = "sqlite:///project.db"

# python -c "import secrets; print(secrets.token_hex(16))"
SECRET_KEY = "a684a727ec8f3d4ffa733ecaeed5b1c4"

run.py

from flask import Flask, render_template

from flask_restful import Api

from resources.spatialite import Version


def run() -> None:

    flask_app = Flask(__name__)
    flask_app.config.from_config("config")

    api = Api(app)
    api.add_resource(Version, "/version")

    # delayed import
    from app.db import db, init_db

    @flask_app.before_first_request
    def create_tables() -> None:
        db.create_all()
        init_db()

    db.init_app(flask_app)


if __name__ == "__main__":
    run()

Troubleshooting

Database is locked

Receiving following error:

sqlalchemy.exc.OperationalError: (OperationalError) database is locked

Execute PRAGMA busy_timeout = 30000; (ms), and/or:

SQLALCHEMY_ENGINE_OPTIONS = {
    "connect_args": {
        "timeout": 30  # in seconds
    }
}

Note: It may be a better idea to consider an alternative database if possible.