Flask-SQLAlchemy 中多表鏈接查詢（不使用外鍵）

SQLAlchemy 是一個功能強大的 ORM 。 Flask-SQLAlchemy 是一個 Flask 插件，它讓我們在 Flask 框架中使用 SQLAlchemy 變得更容易。

本篇介紹我在使用 Flask-SQLAlchemy 2.1 時進行聯表查詢的一些經驗。

表定義

這里有兩個表，account 表保存帳號 ID 和昵稱，bind 表保存 account 之間的綁定關系。

1 2 3 4 5 6 7 8 9 10 11 12

# 省略了外鍵定義，請自行腦補 create table account ( gameuid int auto_increment primary key, nickname varchar(34) not null ); create table bind ( bindid int auto_increment primary key, fromid int not null, toid int not null );

對應的 Model 如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

class Account(db.Model): __tablename__ = 'account' gameuid = db.Column(db.INT, primary_key=True, nullable=False, autoincrement=True) nickname = db.Column(db.VARCHAR(64), nullable=False, unique=True) def __repr__(self): return '<Account %r>' % (self.gameuid) class Bind(db.Model): __tablename__ = 'bind' bindid = db.Column(db.BIGINT, primary_key=True, autoincrement=True) # 綁定者和被綁定者的 gameuid,foreignKey也可以不用設置！！！ fromid = db.Column(db.BIGINT, db.ForeignKey('account.gameuid'), nullable=False) toid = db.Column(db.BIGINT, db.ForeignKey('account.gameuid'), nullable=False) def __repr__(self): return '<Bind %r.%r>' % (self.fromid, self.toid)

關聯查詢

先來看一個簡單的例子：查詢 gameuid 1000 賬號下綁定的所有帳號。

1 2 3 4 5

>>> db.session.query(Bind.bindid, Bind.fromid, Bind.toid, Account.gameuid, Account.nickname). \ filter(Bind.toid == 1000). \ filter(Bind.fromid == Account.gameuid). \ all() [(2, 10001, 1000, 10001, '玩家10001'), (3, 10002, 1000, 10002, '玩家10002'), (4, 10003, 1000, 10003, '玩家10003'), (5, 10004, 1000, 10004, '玩家10004'), (6, 10005, 1000, 10005, '玩家10005'), (7, 10006, 1000, 10006, '玩家10006'), (8, 10007, 1000, 10007, '玩家10007'), (9, 10008, 1000, 10008, '玩家10008'), (10, 10009, 1000, 10009, '玩家10009'), (53, 10000, 1000, 10000, '玩家10000'), (54, 11000, 1000, 11000, '玩家11000')]

看一看生成的 SQL 語句：

1 2 3 4 5 6

>>> print(db.session.query(Bind.bindid, Bind.fromid, Bind.toid, Account.gameuid, Account.nickname). \ filter(Bind.toid == 1000). \ filter(Bind.fromid == Account.gameuid)) SELECT bind.bindid AS bind_bindid, bind.fromid AS bind_fromid, bind.toid AS bind_toid, account.gameuid AS account_gameuid, account.nickname AS account_nickname FROM bind, account WHERE bind.toid = %(toid_1)s AND bind.fromid = account.gameuid

這里的聯表查詢使用的是 WHERE 語句。如果希望使用 JOIN 語句，可以這樣寫：

1 2 3 4 5 6

>>> print(db.session.query(Bind.bindid, Account.gameuid, Account.nickname). \ join(Account, Account.gameuid==Bind.fromid). \ filter(Bind.toid == 1000)) SELECT bind.bindid AS bind_bindid, bind.fromid AS bind_fromid, account.gameuid AS account_gameuid, account.nickname AS account_nickname FROM bind INNER JOIN account ON account.gameuid = bind.fromid WHERE bind.toid = %(toid_1)s

可以看出，現在生成的 SQL 語句已經使用 JOIN 語句了。但上面的語意有點奇怪，既然已經在 query 中使用了 Bind 和 Account，后面再 join 一次 Account 總覺得有點多余。那么 SQLAlchemy 如何選擇 JOIN 的時候誰先誰后呢？看看這個錯誤就知道了：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

>>> db.session.query(Bind.bindid, Bind.fromid, Account.gameuid, Account.nickname). \ join(Bind, Account.gameuid==Bind.fromid). \ filter(Bind.toid == 1000) >>> Traceback (most recent call last): File "<console>", line 1, in <module> File "/Users/zrong/.pyvenv/api/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 1971, in join from_joinpoint=from_joinpoint) File "<string>", line 2, in _join File "/Users/zrong/.pyvenv/api/lib/python3.6/site-packages/sqlalchemy/orm/base.py", line 201, in generate fn(self, *args[1:], **kw) File "/Users/zrong/.pyvenv/api/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 2115, in _join outerjoin, full, create_aliases, prop) File "/Users/zrong/.pyvenv/api/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 2171, in _join_left_to_right l_info.selectable) sqlalchemy.exc.InvalidRequestError: Can't join table/selectable 'bind' to itself

這個錯誤顯然說明，query 中參數的順序很重要，第一個參數所代表的 table 就是 JOIN 時放在前面的那個 table。因此，此處 JOIN 的目標應該是 Account， 而不應該是 Bind 自身。

分頁支持

上面的例子已經解決了大多數需求了。我們再來看看分頁。在 Flask-SQLAlchemy 中封裝了一個 paginate方法，可以方便地將查詢記錄進行分頁：

1 2 3 4 5 6 7

>>> db.session.query(Bind.bindid, Bind.fromid, Account.gameuid, Account.nickname). \ join(Bind, Account.gameuid==Bind.fromid). \ filter(Bind.toid == 1000). \ paginate(1, 10) Traceback (most recent call last): File "<console>", line 1, in <module> AttributeError: 'Query' object has no attribute 'paginate'

報錯的原因是 db.session.query 默認返回的是 orm.Query 對象，這個對象並不包含 paginate 方法。要解決這個問題，需要修改 Flask-SQLAlchemy 的源碼。

找到 SQLAlchemy 對象的 __init__ 定義，在其中加入 session_options['query_cls'] = BaseQuery 即可：flask-sqlalchemy 2.3.2 版本支持的，不用修改源碼了！！！

1 2 3 4 5 6 7 8 9 10 11

def __init__(self, app=None, use_native_unicode=True, session_options=None, metadata=None): if session_options is None: session_options = {} session_options.setdefault('scopefunc', connection_stack.__ident_func__) self.use_native_unicode = use_native_unicode self.app = app # 使用 BaseQuery，這樣可以讓使用 db.session.query 等方法創建的 Query 對象支持 BaseQuery 的方法 session_options['query_cls'] = BaseQuery

另一種關聯查詢語法

在 Flask-SQLAlchemy 提供的 Model 對象中，可以使用 Model.query 這樣的語法來直接得到一個查詢對象，這是由於 Flask-SQLAlchemy 中存在一個 _QueryProperty 類，每次調用 Model.__get__ 時，會自動生成一個基於當前 session 的 query 對象：

1 2 3 4 5 6 7 8 9 10 11 12

class _QueryProperty(object): def __init__(self, sa): self.sa = sa def __get__(self, obj, type): try: mapper = orm.class_mapper(type) if mapper: return type.query_class(mapper, session=self.sa.session()) except UnmappedClassError: return None

使用 Model.query 得到的這個 query 對象可以直接進行 JOIN 操作，得到的結果是 Model 對象。這樣就方便多了：

1 2	>>> Account.query.join(Bind, Bind.fromid == Account.gameuid).filter(Bind.toid == 1000).all() [<Account 10001>, <Account 10002>, <Account 10003>, <Account 10004>, <Account 10005>, <Account 10006>, <Account 10007>, <Account 10008>, <Account 10009>, <Account 10000>, <Account 11000>]

轉換成 SQL 是這樣的：

1 2 3

SELECT account.gameuid AS account_gameuid, account.nickname AS account_nickname FROM account INNER JOIN bind ON bind.fromid = account.gameuid WHERE bind.toid = %(toid_1)s

可以看出，這樣的查詢結果和使用 db.session.query 並沒有什么不同。由於返回的是 Model 對象，使用上可能還更加方便了。

篩選字段

如何使用 Model.query.join 語法得到部分字段呢？這里可以使用 SQLAlchemy 提供的 with_eitities 方法：

1 2 3 4 5

>>> Account.query.join(Bind, Bind.fromid == Account.gameuid). \ filter(Bind.toid == 1000). \ with_entities(Bind.bindid, Account.nickname).all() [(2, '玩家10001'), (3, '玩家10002'), (4, '玩家10003'), (5, '玩家10004'), (6, '玩家10005'), (7, '玩家10006'), (8, '玩家10007'), (9, '玩家10008'), (10, '玩家10009'), (53, '玩家10000'), (54, '玩家11000')] >>>

注意，列表中的項 (2, '玩家10001') 並不是標准的 Python tuple。你如果查看它的類型，會發現一個奇怪的名稱： <class 'sqlalchemy.util._collections.result'> 。它是一個 AbstractKeyedTuple 對象，擁有一個 keys() 方法，這樣可以很容易將其轉換成 dict ：

1 2 3 4 5

>>> results = Account.query.join(Bind, Bind.fromid == Account.gameuid). \ filter(Bind.toid == 1000). \ with_entities(Bind.bindid, Account.nickname).all() >>> [dict(zip(result.keys(), result)) for result in results] [{'bindid': 2, 'nickname': '玩家10001'}, {'bindid': 3, 'nickname': '玩家10002'}, {'bindid': 4, 'nickname': '玩家10003'}, {'bindid': 5, 'nickname': '玩家10004'}, {'bindid': 6, 'nickname': '玩家10005'}, {'bindid': 7, 'nickname': '玩家10006'}, {'bindid': 8, 'nickname': '玩家10007'}, {'bindid': 9, 'nickname': '玩家10008'}, {'bindid': 10, 'nickname': '玩家10009'}, {'bindid': 53, 'nickname': '玩家10000'}, {'bindid': 54, 'nickname': '玩家11000'}]

想了解 AbstractKeyedTuple ，可以看看這篇文檔 New KeyedTuple implementation dramatically faster 。

獲得多個 Model 的記錄

除了篩選字段外，還可以用另一個方法獲取多個 Model 的記錄。那就是，返回兩個 Model 的所有字段：

1 2

>>> db.session.query(Account, Bind).join(Bind, Account.gameuid==Bind.fromid).filter(Bind.toid==1000).all() [(<Account 10001>, <Bind 10001, 1000>), (<Account 10002>, <Bind 10002, 1000>), (<Account 10004>, <Bind 10004, 1000>), (<Account 10005>, <Bind 10005, 1000>), (<Account 10006>, <Bind 10006, 1000>), (<Account 10007>, <Bind 10007, 1000>), (<Account 10008>, <Bind 10008, 1000>), (<Account 10009>, <Bind 10009, 1000>), (<Account 10000>, <Bind 10000, 1000>), (<Account 11000>, <Bind 11000, 1000>)]

使用上面的語法直接返回 Account 和 Bind 對象，可以進行更加靈活的操作。

多表查詢

要聯結超過 2 張以上的表，可以直接在 join 得到的結果之后鏈式調用 join 。也可以在 filter 的結果后面鏈式調用 join 。join 和 filter 返回的都是 query 對象，因此可以無限鏈式調用下去。

寫完查詢后，應該打印生成的 SQL 語句查看一下有沒有性能問題。

https://blog.zengrong.net/post/2656.html