用于 Flask 管理的单独数据库

Question

谁能告诉我，如何为 Flask 管理和 Flask 安全创建一个单独的数据库？ 这就是我使用 PostgreSQL 数据库加载一些表并执行 CRUD 的方式：

import flask_admin
from flask import Flask
from sqlalchemy import create_engine


app =Flask(__name__)
engine = create_engine{'postgresql://name:password@localhost/tablesdatabase')

我希望为flask-admin创建一个单独的数据库:(这就是我正在尝试的)

admin = flask_admin.Admin(app)

app.config['SQLALCHEMY_BINDS'] = {'admindb' : 'postgresql://name:password@localhost/adminedatabase'}
admin.add_view(ModelView) ? // how can i impelement this ? with a seperate datbase ?

Answer 1

没有官方支持，但您可以自定义 Flask-SQLalchemy 会话以使用不同的连接，这里是使用

master

slave

连接的示例，您可以轻松添加任意数量的连接

from functools import partial

from sqlalchemy import orm
from flask import current_app
from flask_sqlalchemy import SQLAlchemy, get_state


class RoutingSession(orm.Session):
    def __init__(self, db, autocommit=False, autoflush=True, **options):
        self.app = db.get_app()
        self.db = db
        self._bind_name = None
        orm.Session.__init__(
            self, autocommit=autocommit, autoflush=autoflush,
            bind=db.engine,
            binds=db.get_binds(self.app),
            **options,
        )

    def get_bind(self, mapper=None, clause=None):
        try:
            state = get_state(self.app)
        except (AssertionError, AttributeError, TypeError) as err:
            current_app.logger.info(
                'cant get configuration. default bind. Error:' + err)
            return orm.Session.get_bind(self, mapper, clause)

        # If there are no binds configured, use default SQLALCHEMY_DATABASE_URI
        if not state or not self.app.config['SQLALCHEMY_BINDS']:
            return orm.Session.get_bind(self, mapper, clause)

        # if want to user exact bind
        if self._bind_name:
            return state.db.get_engine(self.app, bind=self._bind_name)
        else:
            # if no bind is used connect to default
            return orm.Session.get_bind(self, mapper, clause)

    def using_bind(self, name):
        bind_session = RoutingSession(self.db)
        vars(bind_session).update(vars(self))
        bind_session._bind_name = name
        return bind_session


class RouteSQLAlchemy(SQLAlchemy):
    def __init__(self, *args, **kwargs):
        SQLAlchemy.__init__(self, *args, **kwargs)
        self.session.using_bind = lambda s: self.session().using_bind(s)

    def create_scoped_session(self, options=None):
        if options is None:
            options = {}
        scopefunc = options.pop('scopefunc', None)
        return orm.scoped_session(
            partial(RoutingSession, self, **options),
            scopefunc=scopefunc,
        )

默认的会话是

master

，当你想从slave中选择时可以直接调用它，这里是例子：

在您的应用程序中：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql:///master'
app.config['SQLALCHEMY_BINDS'] = {
    'slave': 'postgresql:///slave'
}

db = RouteSQLAlchemy(app)

从大师中选择

session.query(User).filter_by(id=1).first() 

从从机中选择

session.using_bind('slave').query(User).filter_by(id=1).first() 

Answer 2

您可以使用绑定参数定义2个数据库 像这样

# Configure the main database
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://name:password@localhost/tablesdatabase'
db = SQLAlchemy(app)

# Configure the admin database
app.config['SQLALCHEMY_BINDS'] = {'admindb' : 'postgresql://name:password@localhost/adminedatabase'}
admin_db = SQLAlchemy(app, bind='admindb')

或者您可以在同一文件中创建一个新表

import hashlib
from flask import Flask, request, render_template
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# Configure SQLAlchemy
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///kids.db'
db = SQLAlchemy(app)

class Admin(db.Model):
    Name = db.Column(db.Text, nullable=False)
    Email = db.Column(db.Text)
    Password = db.Column(db.Text, nullable=False)
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)



def get_input_md5(input_str):
    result = hashlib.md5(input_str.encode())
    return result.hexdigest()

def check_password(input_password, hashed_password):
    return get_input_md5(input_password) == hashed_password

@app.route("/", methods=["GET", "POST"])
def index():
    if request.method == "POST":
        input_password = request.form["password"]

        # Get the hashed password from the database
        admin = Admin.query.filter_by(Name='Ethan M.').first()
        hashed_password = admin.Password

        # Compare the hashed input with the hashed password in the database
        if check_password(input_password, hashed_password):
            return "Password matches!"
        else:
            return "Password does not match."

    return render_template("check.html")

然后我们访问您可以指定的第一个

class AdminModel(admin_db.Model):
    __tablename__ = 'admin_table'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    # Add other columns as needed

# Initialize Flask-Admin with the admin database
admin = flask_admin.Admin(app, base_template='my_master.html', template_mode='bootstrap4')