SQLAlchemy join subqueries 範例

Posted on  Mar 11, 2020  in  Python 模組/套件推薦  by  Amo Chen  ‐ 2 min read

子查詢(subquery)是使用 SQL 查詢經常會使用的技巧之一,不過各家 ORM(Object-relational mapping) 並沒有共通的實作標準,因此 subquery 的使用方法不一定會相同,這大概是使用 ORM 所必須面對的問題。

本篇紀錄 SQLAlchemy 中如何使用 subquery 以及 subquery 的 JOIN 範例。

本文環境

  • python 3.6.5
  • SQLAlchemy 1.3.15
$ pip install sqlalchemy

本文所使用的 models

以下是本文所定義的 models ,後續的範例也都會使用以下 2 個 models 作為示範:

import hashlib
import sqlalchemy.orm
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref
from sqlalchemy.orm import sessionmaker
from sqlalchemy import desc
from sqlalchemy.orm import aliased
from sqlalchemy import func


Base = declarative_base()


class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    username = Column(String)
    password = Column(String)

    def __init__(self, name, username, password):
        self.name = name
        self.username = username
        self.password = hashlib.sha1(password).hexdigest()


class Address(Base):
    __tablename__ = 'user_address'

    id = Column(Integer, primary_key=True)
    address = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('user.id'))
    user = relationship('User', backref=backref('address', order_by=id))

    def __init__(self, address):
        self.address = address
        
        
engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
user_1 = User('user1', 'username1', 'password_1'.encode('utf-8'))
user_2 = User('user2', 'username2', 'password_2'.encode('utf-8'))
user_1.address = [Address('臺灣凱達格蘭大道')]
user_2.address = [Address('美國紐約時代廣場'), Address('美國華盛頓DC')]
session.add(user_1)
session.add(user_2)
session.commit()

礙於篇幅,上述程式就不再多做解釋。

subquery()

事實上 subquery() 會回傳 Alias 類別的實例:

>>> subq = session.query(User.id).filter(User.id == 1).subquery()
>>> type(subq)
sqlalchemy.sql.selectable.Alias

subquery() 讓我們能夠將子查詢其作為參數放入其他查詢之中。例如將 subq 放進另一個查詢(query)中:

>>> q = session.query(Address).filter(Address.user_id.in_(subq))
>>> print(q)
SELECT user_address.id AS user_address_id, user_address.address AS user_address_address, user_address.user_id AS user_address_user_id 
FROM user_address 
WHERE user_address.user_id IN (SELECT user.id 
FROM user 
WHERE user.id = ?)

可以清楚看到 subq 成為上述範例中的 (SELECT user.id FROM user WHERE user.id = ?) ,也就是我們想要使用的 subquery 。

另外我們可以列出 Alias 中可以使用的 columns :

>>> subq.columns.keys()
['id']

然後也能夠透過 subq.c.<column_name> 取得該欄位:

>>> subq.c.id
Column('id', Integer(), table=<%(4375859600 anon)s>, primary_key=True, nullable=False)

這使得我們能夠在 filter() 中取得 subquery 的欄位名稱作為查詢條件,例如將 in_ 改為 ==

>>> q = session.query(Address).filter(Address.user_id == subq.c.id)
>>> print(q)
SELECT user_address.id AS user_address_id, user_address.address AS user_address_address, user_address.user_id AS user_address_user_id 
FROM user_address, (SELECT user.id AS id 
FROM user 
WHERE user.id = ?) AS anon_1 
WHERE user_address.user_id = anon_1.id

上述範例中可以清楚看到 WHERE user_address.user_id = anon_1.id 就是我們 filter 中所指定的篩選條件,而 anon_1 就是 SQLAlchemy 自動為我們產生的子查詢別名,如果想要指定的話,可以再給 subquery 指定一個別名(alias):

>>> q = session.query(Address).filter(Address.user_id == subq.alias('my_temp_alias').c.id)
>>> print(q)
SELECT user_address.id AS user_address_id, user_address.address AS user_address_address, user_address.user_id AS user_address_user_id 
FROM user_address, (SELECT user.id AS id 
FROM user 
WHERE user.id = ?) AS my_temp_alias 
WHERE user_address.user_id = my_temp_alias.id

上述結果顯示 AS my_temp_alias ,是我們透過 alias() 方法另外指定的別名。

學會這 2 個基礎之後,我們就能進一步學習 subquery 的 JOIN 。

join

JOIN subquery 的方法很簡單,只要將 subquery 放進 join() 方法中作為參數,並取得 columns 中的欄位寫成 ON 所需要的條件即可:

>>> q = session.query(subq.c.id, Address.address).join(subq, subq.c.id == Address.user_id)
>>> print(q)
SELECT anon_1.id AS anon_1_id, user_address.address AS user_address_address 
FROM user_address JOIN (SELECT user.id AS id 
FROM user 
WHERE user.id = ?) AS anon_1 ON anon_1.id = user_address.user_id

上述範例可以看到 JOIN (SELECT user.id AS id FROM user WHERE user.id = ?) AS anon_1 ON anon_1.id = user_address.user_id 就是我們所 JOIN 的 subquery 。

以上,就是 SQLAlchemy 中使用 subquery 的方法,以及 JOIN subquery 的方法。

References

https://docs.sqlalchemy.org/

對抗久坐職業傷害

研究指出每天增加 2 小時坐著的時間,會增加大腸癌、心臟疾病、肺癌的風險,也造成肩頸、腰背疼痛等常見問題。

然而對抗這些問題,卻只需要工作時定期休息跟伸展身體即可!

你想輕鬆改變現狀嗎?試試看我們的 PomodoRoll 番茄鐘吧! PomodoRoll 番茄鐘會根據你所設定的專注時間,定期建議你 1 項辦公族適用的伸展運動,幫助你打敗久坐所帶來的傷害!

贊助我們的創作

看完這篇文章了嗎? 休息一下,喝杯咖啡吧!

如果你覺得 MyApollo 有讓你獲得實用的資訊,希望能看到更多的技術分享,邀請你贊助我們一杯咖啡,讓我們有更多的動力與精力繼續提供高品質的文章,感謝你的支持!