preface
In general, the easiest way to solve SQLAlchemy connection to multiple libraries is to create two or more db. sessions that are not associated with each other, and then modle configures different db. sessions to connect. In this case, the normal configuration of relationship will work.
If so, there is no need to look at the following configuration, which uses SQLAlchemy_Binds to configure multiple databases and uses 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:[email protected]:3306/%s?charset=utf8' % db_name
SQLALCHEMY_DATABASE_URL_WRITE = 'mysql://toto1:[email protected]: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
"""
Reference documents:
* https://docs.sqlalchemy.org/en/13/orm/relationship_api.html # worthy of scrutiny * https://www.cnblogs.com/srd945/p/9851227.html * https://www.osgeo.cn/sqlalchemy/orm/relationship_api.html # ditto, Chinese extend_existing: (False) When the table already exists in the metadata, if a column with the same name exists in the column_list, the column in the column_list replaces the existing column in the metadata. * useexisting is deprecated, and the new version uses extend_existingconclusion
It’s easy to make mistakes, read the official documentation, and try to create modles in a simple, consistent way, without creating foreign keys in the database layer.
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)digression
User access to the multi-connection database needs to be configured in advance, and I ran into a snag: I connected to a database with IP 111.111.111.111, and even using the MySQL command enabled me to access the data perfectly, but using SQLAlchemy_Binder I got an error when I accessed it:
OperationalError: (_mysql_exceptions.OperationalError) (1142, SELECT command denied to user' toto'@'125.121.34.123' FOR TABLE 'user' 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)]125.121.34.123 Don’t be attracted by this IP. The last time I had this problem was with SQLALCHEMY_DATABASE_URL There is a connection error, resulting in different IPs seen before and after. The difference between 111.111.111.111 in this connection and 125.121.34.123 in the error report is because the IP is not in the same network segment. 125.121.34.123 is the post assignment of 111.111.111.111.111.111 through multi-segment routing The TCP/IP)
Mysql [email protected]:test> SELECT test2.user.username from 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 [email protected]:test> SELECT * from user where sess_id = '125.121.34.123' 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 | +--------------------------+-----------------------+