在 flask-sqlalchemy 中使用具有相同模型的多个数据库

Posted 2023-02-24

【中文标题】在 flask-sqlalchemy 中使用具有相同模型的多个数据库

【英文标题】：Using multiple database with same model in flask-sqlalchemy

【发布时间】：2018-01-22 08:16:13

【问题描述】：

所以我有多个数据库。 DB_US 和 DB_UK。它们具有完全相同的结构，但它们保存不同国家/地区的数据。我有一个带有 flask_restful 的 api。端点接受一个国家来检索这个特定国家的数据。

我的问题是我不知道如何根据请求选择数据库。我已经阅读了 bind 方法，但是您必须为每个模型指定一个 __bind_key，并且我的所有模型都在两个数据库中。

我尝试在@before_request 中初始化数据库，但是在处理完第一个请求后，您无法调用init_db 方法（或任何设置方法）。

这是一个特殊情况，因为数据库是相同的，所以将它放在同一个数据库中似乎是有意义的，但它是更大系统的一部分，需要这样。

如何根据每个请求选择数据库？

【参考方案1】：

我遇到了同样的问题。我在这里描述过Using multiple POSTGRES databases and schemas with the same Flask-SQLAlchemy model

您的案例稍微容易一些，因为您不必处理架构差异。 您可以按照 stamaimer 的建议进行操作，但问题是为每个国家/地区创建了新的用户模型（如果您有超过 2 个，它就会变得混乱）。

为了解决这个问题，我停止使用 flask-sqlAlchemy db 实例化方法，让内容提供者 init 方法处理要选择的 db：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker()

class ContentProvider():

    db = None
    connection = None
    session = None

    def __init__(self, center):
        if center == UK:
            self.db = create_engine('sqlite://UK', echo=True, pool_threadlocal=True)
            self.connection = self.db.connect()
        elif center == US:
            self.db = create_engine('sqlite://US', echo=True, pool_threadlocal=True)
            self.connection = self.db.connect()
        self.session = Session(bind=self.connection)

    def index(country):
        user = models[country]["User"](username=country + str(time.time()))
        self.session.add(user)
        self.session.commit()
        return country, "added"

【参考方案2】：

也许您可以定义一些基本数据模型并为单独的数据库继承它们。例如：

class User(db.Model):

    __abstract__ = 1

    # fields define

class UserUK(User):

    __bind_key__ = 'uk'

class UserUS(User):

    __bind_key__ = 'us'

您可以创建一个model = "uk": "User": UserUK, "us": "User": UserUS 的字典，以便从请求中按国家/地区选择合适的型号。例如：model[country]["User"]。

完整演示：

import time
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:////path/2/default.db"
app.config["SQLALCHEMY_BINDS"] = "uk": "sqlite:////path/2/uk.db", "us": "sqlite:////path/2/us.db"

db = SQLAlchemy(app)


class UserBase(db.Model):

    __abstract__ = 1

    id = db.Column(db.Integer, primary_key=1)

    username = db.Column(db.String(16))


class UserUK(UserBase):

    __bind_key__ = "uk"


class UserUS(UserBase):

    __bind_key__ = "us"


models = "uk": "User": UserUK, "us": "User": UserUS


@app.route("/<country>")
def index(country):

    user = models[country]["User"](username=country + str(time.time()))

    db.session.add(user)

    db.session.commit()

    return country, "added"


if __name__ == "__main__":

    app.run(debug=1)