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                   |
+--------------------------+-----------------------+