使用用户输入动态创建烧瓶 sqlalchemy 数据库连接答案

【问题标题】：Dynamically create flask sqlalchemy database connection using user inputs使用用户输入动态创建烧瓶 sqlalchemy 数据库连接
【发布时间】：2020-11-30 22:49:51
【问题描述】：

我想知道是否可以使用用户传入的输入来创建 SqlAlchemy 数据库 URI。例如，具有输入文本字段，用户必须在其中输入创建数据库 uri 所需的参数。只有这样才能建立连接，然后用户才能访问数据库中的数据。

【问题讨论】：

标签： python-3.x flask flask-sqlalchemy

【解决方案1】：

其实是可以的。但是你需要在没有flask-sqlalchemy的情况下管理数据库连接。

您需要数据库引擎的存储和更新它的路径。数据库引擎是在这条路线中创建的，并通过执行简单的请求来测试它们是否真正起作用。数据库 URI 被保存到用户会话中以供以后使用：

from flask import Flask, Response, render_template, request, session
from sqlalchemy import create_engine
from sqlalchemy.exc import OperationalError

ENGINES = {}

app = Flask(__name__)

@app.route("/set_database_uri", methods=("GET", "POST"))
def set_database_uri():
    if request.method == "POST":
        database_uri = request.form["database_uri"]
        if database_uri not in ENGINES:
            engine = create_engine(database_uri)
            try:
                engine.execute("SELECT 1")
            except OperationalError:
                return Response("Invalid database URI", 400)
            ENGINES[database_uri] = engine
        session["database_uri"] = database_uri
    return render_template("database_uri.html")

这些引擎可用于生成 SQLAlchemy 会话以进行所需的数据库操作：

from flask import redirect, session, url_for
from sqlalchemy.orm import Session

@app.route("/records")
def get_records():
    if "database_uri" not in session:
        return redirect(url_for("set_database_uri"))

    db_session = Session(session["database_uri"])
    records = db_session.query(Record).all()
    db_session.close()

    return render_template("records.html", records=records)

虽然技术上可行，但不建议使用这种方法。这可能会导致严重的安全漏洞，因为潜在用户可以创建与应用程序主机上任何可用数据库的数据库连接。

但是，如果您正在制作某种数据库管理界面，它可能会很有用。

【讨论】：

【解决方案2】：

我刚遇到同样的问题，我就这样解决了。

配置部分：

from contextlib import contextmanager

from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

from greenwater_app import app


app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db/database1.db'
app.config['SQLALCHEMY_BINDS'] = {
    'database2': 'sqlite:///db/database2.db'
}
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

migrate = Migrate(app, db, compare_type=True)


@contextmanager
def session_scope(bind='default'):
    """Provide a transactional scope around a series of operations."""

    if bind != 'default':
        s = db.create_scoped_session(options={'bind': db.get_engine(db.get_app(), bind)})
    else:
        s = db.session

    s.expire_on_commit = False
    try:
        yield s
        s.commit()
    except:
        s.rollback()
        raise
    finally:
        s.close()

然后，我们可以在类定义中添加“bind_key”作为“database2”来使用“database2”。像这样。 https://flask-sqlalchemy.palletsprojects.com/en/2.x/binds/

class User(db.Model, BaseModelMixin):
    __bind_key__ = "database2"
    __tablename__ = "user"

如果我们想参加特别会议，我们可以这样做。

sql = 'select * from user'
with session_scope(bind='database2') as s:
    res = s.execute(sql).fetchall()
    print([dict(r) for r in res])

with session_scope() as s:
    res = s.execute(sql).fetchall()
    print([dict(r) for r in res])

session_scope(bind='database2') 将为 database2 获取会话。

【讨论】：