sqlalchemy 配置多链接读写库后的relationship设置

前言

通常来讲,解决sqlalchemy 链接多个库的最简单的方式是新建两个或多个db.session 相互没有关联,而后modle配置不一样的db.session来链接,这样的话,relationship正常配置就行,不用特殊配置.

若是这样解决的话,也就不用看下面的配置了,下面是使用SQLALCHEMY_BINDS配置多个多个数据库并使用relationship.(我的建议最好不用relationship,很容易发生各类报错...)

# -*- coding:utf-8 -*-
__doc__ = "使用SQLALCHEMY_BINDS 就必须双份Model各自配置__bind_key__ ,同名库的读写分离与relationship配置的示例"

import flask
from flask_sqlalchemy import SQLAlchemy  # Flask-SQLAlchemy 2.3.2
from datetime import datetime
from sqlalchemy.orm import backref, foreign  # SQLAlchemy 1.3.1

db_name = 'db_name'
SQLALCHEMY_DATABASE_URL_READ = 'mysql://toto:toto123@111.111.111.111:3306/%s?charset=utf8' % db_name
SQLALCHEMY_DATABASE_URL_WRITE = 'mysql://toto1:toto123@222.222.222.222:3306/%s?charset=utf8' % db_name

app = flask.Flask(__name__)
app.config['DEBUG'] = True
app.config['SQLALCHEMY_BINDS'] = {
    'read_db': SQLALCHEMY_DATABASE_URI_READ,
    'write_db': SQLALCHEMY_DATABASE_URL_WRITE,
}

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True
db = SQLAlchemy(app)


class RDriver(db.Model):
    __bind_key__ = 'read_db'
    __tablename__ = 'driver'
    # __table_args__ = {'schema': db_name}  # __tablename__ 相同的,读库写库至少有一个要加,此示例是读库都不加,写库必须加

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    fk_user_id = db.Column(db.Integer, db.ForeignKey("user.id"))
    create_at = db.Column(db.TIMESTAMP, default=datetime.now)


class RUser(db.Model):
    __bind_key__ = 'read_db'
    __tablename__ = 'user'
    # __table_args__ = {'schema': db_name}

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(32), index=True, unique=True)
    email = db.Column(db.String(32))
    create_time = db.Column(db.TIMESTAMP, default=datetime.now)
    update_time = db.Column(db.TIMESTAMP, default=datetime.now)
    # 以下的五种方式都是能够的
    # driver_fk = db.relationship("RDriver", foreign_keys='RDriver.fk_user_id')
    # driver_fk = db.relationship("RDriver", primaryjoin=lambda: RDriver.fk_user_id == RUser.id, viewonly=True)
    # driver_fk = db.relationship("RDriver", primaryjoin=RDriver.fk_user_id == id)
    fk_driver = db.relationship("RDriver", primaryjoin='RDriver.fk_user_id == RUser.id')
    # driver_fk = db.relationship("RDriver", backref=db.backref('user', lazy=True),
    #                             primaryjoin=lambda: RDriver.fk_user_id == RUser.id, viewonly=True)


####################################################上面为同名读库,下面为同名写库###########################################
class WDriver(db.Model):
    __bind_key__ = 'write_db'
    __tablename__ = 'driver'
    __table_args__ = {'schema': db_name, 'extend_existing': True}  # 这个配置很关键,同名写库都要配置
    # __table_args__ = {'extend_existing': True}
    # 这样配置的话会报错: sqlalchemy.exc.ArgumentError: Could not locate any relevant foreign
    # key columns for primary join condition 'driver.fk_user_id = "user".id' on relationship RUser.fk_driver.
    # Ensure that referencing columns are associated with a ForeignKey or ForeignKeyConstraint,
    # or are annotated in the join condition with the foreign() annotation.

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    fk_user_id = db.Column(db.Integer, db.ForeignKey("%s.user.id" % db_name))  # db_name.user.id很关键,必须指定用的数据库名
    create_at = db.Column(db.TIMESTAMP, default=datetime.now)


class WUser(db.Model):
    __bind_key__ = 'write_db'
    __tablename__ = 'user'
    __table_args__ = {'schema': db_name, 'extend_existing': True}  # 这个配置很关键,同名写库都要配置
    # __table_args__ = {'extend_existing': True}  # 这个配置很关键,同名写库都要配置

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(32), index=True, unique=True)
    email = db.Column(db.String(32))
    create_time = db.Column(db.TIMESTAMP, default=datetime.now)
    update_time = db.Column(db.TIMESTAMP, default=datetime.now)
    # 如下五种方式都是能够的
    # fk_driver = db.relationship("WDriver", foreign_keys='WDriver.fk_user_id', uselist=False)
    # fk_driver = db.relationship("WDriver", primaryjoin=lambda: WDriver.fk_user_id == WUser.id)
    fk_driver = db.relationship("WDriver", primaryjoin=WDriver.fk_user_id == id)
    # fk_driver = db.relationship("WDriver", primaryjoin='WDriver.fk_user_id == WUser.id')
    # fk_driver = db.relationship("WDriver", backref=db.backref('test.user', lazy=True),
    #                             primaryjoin=lambda: WDriver.fk_user_id == WUser.id)


r_user_obj = RUser.query.filter_by().first()
print("r_user_obj:", r_user_obj)
print("r_user_obj.driver_fk:", r_user_obj.fk_driver)

""" 打印出来的sql
SELECT user.id AS user_id, user.username AS user_username, user.email AS user_email, user.create_time AS user_create_time, user.update_time AS user_update_time
FROM user
 LIMIT 1

SELECT driver.id AS driver_id, driver.fk_user_id AS driver_fk_user_id, driver.create_at AS driver_create_at
FROM driver
WHERE driver.fk_user_id = %s
"""

w_user_obj = WUser.query.filter_by(id=2188).first()
print("w_user_obj:", w_user_obj)
print("w_user_obj.driver_fk:", w_user_obj.fk_driver)

""" 打印出来的sql,能够看出多了数据库名
SELECT db_name.user.id AS db_name_user_id, db_name.user.username AS db_name_user_username, db_name.user.email AS db_name_user_email, db_name.user.create_time AS db_name_user_create_time, db_name.user.update_time AS db_name_user_update_time
FROM db_name.user
WHERE db_name.user.id = %s
 LIMIT %s

 SELECT db_name.driver.id AS db_name_driver_id, db_name.driver.fk_user_id AS db_name_driver_fk_user_id, db_name.driver.create_at AS db_name_driver_create_at
FROM db_name.driver
WHERE db_name.driver.fk_user_id = %s

"""

参考文档:

* https://docs.sqlalchemy.org/en/13/orm/relationship_api.html # 值得细看
* https://www.osgeo.cn/sqlalchemy/orm/relationship_api.html # 同上,中文
* https://www.cnblogs.com/srd945/p/9851227.html
* extend_existing: (False)当表已经存在于元数据中时，若是元数据中存在与column_list中的列同名的列，column_list中同名的列会替换掉元数据中已经有的列
* useexisting已被废弃, 新版本使用extend_existing

总结

关系配置参数真的不少,以下,很容易就会出错,须要多读读官方文档,还有就是创建modle时候尽可能简洁,风格统一,不要在数据库层创建外键.

sqlalchemy.orm.relationship(argument, secondary=None, primaryjoin=None, secondaryjoin=None, foreign_keys=None,
uselist=None, order_by=False, backref=None, back_populates=None, post_update=False, cascade=False, extension=None,
viewonly=False, lazy='select', collection_class=None, passive_deletes=False, passive_updates=True, remote_side=None,
enable_typechecks=True, join_depth=None, comparator_factory=None, single_parent=False, innerjoin=False,
distinct_target_key=None, doc=None, active_history=False, cascade_backrefs=True, load_on_pending=False,
bake_queries=True, _local_remote_pairs=None, query_class=None, info=None, omit_join=None)

题外话

配置多链接数据库用户访问权限须要提早配置好,我就遇到个坑:
我链接了ip为111.111.111.111的数据库,使用mysql命令好都能很好的正常访问到数据,可是使用SQLALCHEMY_BINDS访问后就报错 :

OperationalError: (_mysql_exceptions.OperationalError) (1142, "SELECT command denied to user 'toto'@'125.121.34.123' for table 'user'") [SQL: u'SELECT test2.user.com_name AS test2_user_com_name, test2.user.com_no AS test2_user_com_no \nFROM test2.user \nWHERE test2.user.id = %s \n LIMIT %s'] [parameters: (3385, 1)]

其实仍是本身没有检查清楚,toto用户我只给了test库的权限,而test2库的权限没给,才会形成这样的报错(125.121.34.123别被这个ip吸引了注意力,我上次遇到这样的问题是配置中SQLALCHEMY_DATABASE_URL的链接有点错误,致使先后看到的ip不同,此次链接中的111.111.111.111和报错中的125.121.34.123不同是由于ip不在同一网段,125.121.34.123就是111.111.111.111通过多段路由链接过来的后分配到的ip)

mysql toto@111.111.111.111:test> SELECT test2.user.username AS test_user_username, test2.user.id AS test_ship
                                    pers_id FROM test2.user WHERE test2.user.id = 3385  LIMIT 1;
(1142, "SELECT command denied to user 'toto'@'125.121.34.123' for table 'user'")
mysql toto@111.111.111.111:test> SELECT test.user.username AS test_user_username, test.user.id AS
                                    test_user_id FROM test.user WHERE test.user.id = 3385  LIMIT 1;
+--------------------------+-----------------------+
| test_user_username       | test_user_id          |
+--------------------------+-----------------------+
| XXXXXXX                  | 123                   |
+--------------------------+-----------------------+