前言

一般来说,解决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 |

+--------------------------+-----------------------+