前言
一般来说,解决sqlalchemy 连接多个库的最简单的方式是新建两个或多个db.session 相互没有关联,modle配置不同的db.session来连接,这样的话,relationship正常配置就行,不用特殊配置.如果这样解决的话,也就不用看下面的配置了
# -*- coding:utf-8 -*-import flaskfrom flask_sqlalchemy import SQLAlchemy # Flask-SQLAlchemy 2.3.2from datetime import datetimefrom sqlalchemy.orm import backref, foreign # SQLAlchemy 1.3.1app = flask.Flask(__name__)app.config['DEBUG'] = Trueapp.config['SQLALCHEMY_BINDS'] = { 'read_db': 'mysql://reader:test@127.0.0.1:3306/test?charset=utf8', 'write_db': 'mysql://writer:test@127.0.0.2:3306/test?charset=utf8'}app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Falseapp.config['SQLALCHEMY_ECHO'] = Falsedb = SQLAlchemy(app)class RDriver(db.Model): __bind_key__ = 'read_db' __tablename__ = 'driver' # __table_args__ = {'schema': 'test'} # 不可以加上 id = db.Column(db.Integer, primary_key=True, autoincrement=True) fk_user_id = db.Column(db.Integer, db.ForeignKey("user.id")) driver_name = db.Column(db.String(7)) create_time = db.Column(db.TIMESTAMP, default=datetime.now)class RUser(db.Model): __bind_key__ = 'read_db' __tablename__ = 'user' # __table_args__ = {'schema': 'test'} id = db.Column(db.Integer, primary_key=True, autoincrement=True) user_name = db.Column(db.String(32), index=True, unique=True) user_password = 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': 'test', 'extend_existing': True} # 这个配置很关键 id = db.Column(db.Integer, primary_key=True, autoincrement=True) fk_user_id = db.Column(db.Integer, db.ForeignKey("test.user.id")) # test.user.id很关键 plate = db.Column(db.String(7)) create_at = db.Column(db.TIMESTAMP, default=datetime.now)class WUser(db.Model): __bind_key__ = 'write_db' __tablename__ = 'user' __table_args__ = {'schema': 'test', 'extend_existing': True} # 这个配置很关键 id = db.Column(db.Integer, primary_key=True, autoincrement=True) hash = db.Column(db.String(256), nullable=False) user_no = db.Column(db.String(32), index=True, unique=True) # 用户工号 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)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)参考文档:
* 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)