建表
連接一個數(shù)據(jù)庫
>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite:///:memory:', echo=True)
Dialect 定義特定數(shù)據(jù)庫的DBAPI行為纠炮，能為不同的數(shù)據(jù)庫轉(zhuǎn)換SQL語法。
Pool 是連接池灯蝴，當(dāng)生成的session實(shí)例操作數(shù)據(jù)庫（或者多個實(shí)例并發(fā)簽出）的時候把session放進(jìn)這個池統(tǒng)一管理恢口，如果session實(shí)例綁定了bind=engine，則該session實(shí)例在使用時默認(rèn)放到由這個engine管理的連接池中穷躁。
把表映射到自己的類
聲明ORM基類
>>> from sqlalchemy.ext.declarative import declarative_base
>>> Base = declarative_base()#這是一個關(guān)系映射基類
User 將是我們映射此表的類耕肩。在類中，我們定義了要映射到的表的詳細(xì)信息问潭，主要是表名以及列的名稱和數(shù)據(jù)類型：
>>> from sqlalchemy import Column, Integer, String
>>> class User(Base):
... __tablename__ = 'users'
...
... id = Column(Integer, primary_key=True)
... name = Column(String)
... fullname = Column(String)
... nickname = Column(String)
...
... def __repr__(self):
... return "<User(name='%s', fullname='%s', nickname='%s')>" % (
... self.name, self.fullname, self.nickname)
查看剛才創(chuàng)建的表結(jié)構(gòu)
>>> User.__table__
Table('users', MetaData(bind=None),
Column('id', Integer(), table=<users>, primary_key=True, nullable=False),
Column('name', String(), table=<users>),
Column('fullname', String(), table=<users>),
Column('nickname', String(), table=<users>), schema=None)
這個表結(jié)構(gòu)被登記在Base.metadata.tables中猿诸，如果有多個子類繼承了這個Base，那么這個Base.metadata就記錄了所有剛剛定義的子類（表結(jié)構(gòu)）狡忙。
在數(shù)據(jù)庫中創(chuàng)建表
現(xiàn)在用 MetaData 向數(shù)據(jù)庫發(fā)出create table語句：
>>> Base.metadata.create_all(engine)
如果Base.metadata存在多個子類梳虽，則會全部創(chuàng)建。
向表中插入數(shù)據(jù)
每一個映射類對應(yīng)一張表灾茁，我們要插入的每一條記錄都是映射類的實(shí)例窜觉，要插入數(shù)據(jù)，就是向這張表中添加映射類的實(shí)例北专。
創(chuàng)建會話
會話（session）是一個與數(shù)據(jù)庫對話的控制器禀挫，能向Engine申請連接池（Pool）資源。
session常用操作：
session.query():查詢拓颓，從數(shù)據(jù)庫中定位數(shù)據(jù)
session.flush()：預(yù)提交语婴，提交到數(shù)據(jù)庫文件，還未寫入數(shù)據(jù)庫文件中
session.commit()：提交了一個事務(wù)录粱，把緩存的數(shù)據(jù)直接寫入數(shù)據(jù)庫
session.rollback()：回滾
session.close()：關(guān)閉
關(guān)于報(bào)錯 “Can’t reconnect until invalid transaction is rolled back”
session事務(wù)失敗腻格，占用連接池資源，無法提交新的事務(wù)啥繁。需要rollback()菜职，或者close()來打破這種狀態(tài)（也可以通過設(shè)置engine的autocommit為True來取消事務(wù)提交，比較野蠻旗闽，不推薦）酬核。
創(chuàng)建會話的方式一（為每個Session工廠分配engine）
>>> from sqlalchemy.orm import sessionmaker
>>> Session = sessionmaker(bind=engine)# 生成Session工廠，與數(shù)據(jù)庫連接綁定
>>> session = Session()
創(chuàng)建會話的方式二（為每個session分配engine）
>>> Session = sessionmaker()
>>> session = Session(bind=engine)# 生成session實(shí)例适室，與數(shù)據(jù)庫連接綁定
創(chuàng)建會話的方式三（延遲綁定）
>>> Session = sessionmaker()
>>> Session.configure(bind=engine)
>>> session = Session()
創(chuàng)建會話的方式四（不綁定）
>>> Session = sessionmaker()
>>> session = Session()
注意：如果是Session = sessionmaker(bind=engine)嫡意，由這個Session實(shí)例化的session只能操作engine能訪問的視圖或表，關(guān)聯(lián)Engine是為了確保這個session可以使用該engine的連接資源捣辆。
如果不用Session.configure(bind=engine)綁定engine蔬螟，則這個會話實(shí)例可以操作任意的engine，由于我們在下面的程序中反射表的時候用了綁定到engine的metadata汽畴，使用上面這個沒有綁定engine的session查詢的時候會找到表對應(yīng)的metadata綁定的engine來進(jìn)行坎怪，但是這個session無法操作metadata沒有綁定到engine的表。
創(chuàng)建會話的方式五（線程安全）
>>> from sqlalchemy.orm import scoped_session
>>> from sqlalchemy.orm import sessionmaker
>>> session = scoped_session(sessionmaker(bind=engine))
更多關(guān)于session的問題廓握，請看我的另一篇文章：http://www.reibang.com/p/2524784c89c6
添加和更新對象
向之前創(chuàng)建的表中添加數(shù)據(jù)：
>>> ed_user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
>>> session.add(ed_user)
此時搅窿，這個session處于Pending狀態(tài)，還沒有發(fā)出SQL隙券，對象還沒有在數(shù)據(jù)庫中表示為一行是尔。這個session將使用一個刷新的方法殉了，在需要時立即發(fā)出SQL來持久化數(shù)據(jù)。
下面恩溅，我們創(chuàng)建一個新的query對象隔箍，查看User表中的名為‘ed’的對象：
>>> our_user = session.query(User).filter_by(name='ed').first()
>>> our_user
<User(name='ed', fullname='Ed Jones', nickname='edsnickname')>
這時候會話已經(jīng)識別出返回的行與在其內(nèi)部對象映射中表示的行相同，所以我們實(shí)際上得到了與我們剛剛添加的相同的實(shí)例脚乡。
>>> ed_user is our_user
True
這里使用的ORM概念稱為標(biāo)識映射蜒滩，它確保會話中對特定行進(jìn)行的所有操作都對同一組數(shù)據(jù)進(jìn)行操作。一旦會話中出現(xiàn)具有特定主鍵的對象奶稠，該會話上的所有SQL查詢將始終返回該特定主鍵的相同Python對象俯艰；如果試圖在會話中放置具有相同主鍵的第二個已經(jīng)持久化的對象，也會引發(fā)錯誤锌订。
我們可以使用add_all()一次性添加更多的User對象：
>>> session.add_all([
... User(name='wendy', fullname='Wendy Williams', nickname='windy'),
... User(name='mary', fullname='Mary Contrary', nickname='mary'),
... User(name='fred', fullname='Fred Flintstone', nickname='freddy')])
更改一個人的昵稱：
>>> ed_user.nickname = 'eddie'
這個更改出現(xiàn)在session.dirty屬性中：
>>> session.dirty
IdentitySet([<User(name='ed', fullname='Ed Jones', nickname='eddie')>])
三個新的用戶對象正在等待處理的掛起（pending）態(tài)：
>>> session.new
IdentitySet([<User(name='wendy', fullname='Wendy Williams', nickname='windy')>,
<User(name='mary', fullname='Mary Contrary', nickname='mary')>,
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>])
我們告訴session竹握，我們希望向數(shù)據(jù)庫發(fā)出所有剩余的更改，并提交正在進(jìn)行中的事務(wù)辆飘。我們通過session.commit()來做這件事啦辐。會話發(fā)出針對“ed”上的昵稱更改的UPDATE語句，以及我們添加的三個新用戶對象的INSERT語句：
>>> session.commit()
commit()將剩余的更改刷新到數(shù)據(jù)庫蜈项，并提交事務(wù)芹关。會話引用的連接資源現(xiàn)在返回到連接池。此會話的后續(xù)操作將在一個新事務(wù)中發(fā)生紧卒，該事務(wù)將在首次需要時再次重新獲取連接資源侥衬。
如果我們看看Ed的id屬性，之前它是None，現(xiàn)在它有一個值:
>>> ed_user.id
1
User的對象在這個寫入數(shù)據(jù)庫的過程中經(jīng)歷了以下三種狀態(tài)
transient, pending, persistent
一個對象在一個會話中可以擁有的狀態(tài)如下：
- Transient：不在會話中且未保存到數(shù)據(jù)庫中的實(shí)例轴总；也就是說贬媒，它沒有數(shù)據(jù)庫標(biāo)識。這樣的對象與ORM的唯一關(guān)系是肘习，它的類有一個與之關(guān)聯(lián)的mapper()。
- Pending：當(dāng)您使用Session.add()臨時實(shí)例時坡倔，它會變成掛起的漂佩。它實(shí)際上還沒有被刷新到數(shù)據(jù)庫中，但下一次刷新時就會刷新罪塔。
- Persistent：在會話中出現(xiàn)并在數(shù)據(jù)庫中有記錄的實(shí)例投蝉。您可以通過刷新使掛起的實(shí)例變?yōu)槌志脤?shí)例，或者通過查詢現(xiàn)有實(shí)例的數(shù)據(jù)庫(或者將持久實(shí)例從其他會話移動到本地會話)來獲得持久實(shí)例征堪。
- Deleted：一個實(shí)例瘩缆，在一次刷新中已被刪除，但事務(wù)尚未完成佃蚜。這種狀態(tài)下的對象本質(zhì)上與“掛起”狀態(tài)相反;當(dāng)提交會話事務(wù)時庸娱，對象將移動到分離狀態(tài)⌒乘悖或者熟尉，當(dāng)會話的事務(wù)回滾時，已刪除的對象將移動回持久狀態(tài)洲脂。
- Detached：與數(shù)據(jù)庫中的記錄相對應(yīng)或以前相對應(yīng)的實(shí)例斤儿，但當(dāng)前不在任何會話中。分離的對象將包含一個數(shù)據(jù)庫標(biāo)識標(biāo)記恐锦，但是由于它與會話沒有關(guān)聯(lián)往果，因此無法知道這個數(shù)據(jù)庫標(biāo)識是否實(shí)際存在于目標(biāo)數(shù)據(jù)庫中。分離的對象正常使用是安全的一铅，除了它們沒有能力加載未加載的屬性或之前標(biāo)記為“過期”的屬性陕贮。
深入了解“https://docs.sqlalchemy.org/en/13/orm/session_events.html#session-lifecycle-events”
任何映射對象的實(shí)際狀態(tài)可以在任何時候使用inspect()查看：
>>> from sqlalchemy import inspect
>>> insp = inspect(ed_user)
>>> insp.persistent
True
回滾
先亂改一波數(shù)據(jù)
>>> ed_user.name = 'Edwardo'
>>> fake_user = User(name='fakeuser', fullname='Invalid', nickname='12345')
>>> session.add(fake_user)
查詢會話時，可以看到它們被刷新到當(dāng)前事務(wù)中：
>>> session.query(User).filter(User.name.in_(['Edwardo', 'fakeuser'])).all()
[<User(name='Edwardo', fullname='Ed Jones', nickname='eddie')>, <User(name='fakeuser', fullname='Invalid', nickname='12345')>]
回滾一下馅闽，我們可以看到 ed_user 的name變回 ed 飘蚯，fake_user 已不在會話中：
>>> session.rollback()
>>> ed_user.name
u'ed'
>>> fake_user in session
False
再次查詢發(fā)現(xiàn)已經(jīng)恢復(fù)
>>> session.query(User).filter(User.name.in_(['ed', 'fakeuser'])).all()
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>]
查詢
>>> for instance in session.query(User).order_by(User.id):
... print(instance.name, instance.fullname)
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flintstone
>>> for name, fullname in session.query(User.name, User.fullname):
... print(name, fullname)
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flintstone
>>> for row in session.query(User, User.name).all():
... print(row.User, row.name)
<User(name='ed', fullname='Ed Jones', nickname='eddie')> ed
<User(name='wendy', fullname='Wendy Williams', nickname='windy')> wendy
<User(name='mary', fullname='Mary Contrary', nickname='mary')> mary
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')> fred
>>> for row in session.query(User.name.label('name_label')).all():
... print(row.name_label)
ed
wendy
mary
fred
>>> from sqlalchemy.orm import aliased
>>> user_alias = aliased(User, name='user_alias')
>>> for row in session.query(user_alias, user_alias.name).all():
... print(row.user_alias)
<User(name='ed', fullname='Ed Jones', nickname='eddie')>
<User(name='wendy', fullname='Wendy Williams', nickname='windy')>
<User(name='mary', fullname='Mary Contrary', nickname='mary')>
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>
>>> for u in session.query(User).order_by(User.id)[1:3]:
... print(u)
<User(name='wendy', fullname='Wendy Williams', nickname='windy')>
<User(name='mary', fullname='Mary Contrary', nickname='mary')>
>>> for name, in session.query(User.name).\
... filter_by(fullname='Ed Jones'):
... print(name)
ed
>>> for name, in session.query(User.name).\
... filter(User.fullname=='Ed Jones'):
... print(name)
ed
>>> for user in session.query(User).\
... filter(User.name=='ed').\
... filter(User.fullname=='Ed Jones'):
... print(user)
<User(name='ed', fullname='Ed Jones', nickname='eddie')>
更多的filter操作：
query.filter(User.name == 'ed')
query.filter(User.name != 'ed')
query.filter(User.name.like('%ed%'))
query.filter(User.name.ilike('%ed%'))
query.filter(User.name.in_(['ed', 'wendy', 'jack']))
query.filter(User.name.in_(
session.query(User.name).filter(User.name.like('%ed%'))
))
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
query.filter(User.name == None)
query.filter(User.name.is_(None))
query.filter(User.name != None)
query.filter(User.name.isnot(None))
# and在filter中的三種方式
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')
from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))
query.filter(User.name.match('wendy'))
返回列表和標(biāo)量
>>> query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)
>>> query.all()
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>,
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>]
>>> query.first()
<User(name='ed', fullname='Ed Jones', nickname='eddie')>
>>> user = query.one()#不是只有一行就報(bào)錯（0行）
Traceback (most recent call last):
...
MultipleResultsFound: Multiple rows were found for one()
>>> user = query.filter(User.id == 99).one()#不是只有一行就報(bào)錯（2行以上）
Traceback (most recent call last):
...
NoResultFound: No row was found for one()
>>> user = query.filter(User.id == 99).one_or_none()#超過一行就報(bào)錯，一行或零行不報(bào)錯
Traceback (most recent call last):
...
NoResultFound: No row was found for one()
>>> query = session.query(User.id).filter(User.name == 'ed').order_by(User.id)
>>> query.scalar()#scalar() 調(diào)用 one() 方法福也，并在成功時返回行的第一列：
1
使用文本SQL
>>> from sqlalchemy import text
>>> for user in session.query(User).\
... filter(text("id<224")).\
... order_by(text("id")).all():
... print(user.name)
ed
wendy
mary
fred
使用冒號指定綁定參數(shù)局骤，用 params() 方法指定值：
>>> session.query(User).filter(text("id<:value and name=:name")).\
... params(value=224, name='fred').order_by(User.id).one()
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>
把表示完整語句的text()傳遞給 from_statement()：
>>> session.query(User).from_statement(
... text("SELECT * FROM users where name=:name")).\
... params(name='ed').all()
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>]
text() 允許我們將文本SQL按位置鏈接到ORM映射的列；具體的操作是將列作為位置參數(shù)傳遞給 TextClause.columns()：
>>> stmt = text("SELECT name, id, fullname, nickname "
... "FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id, User.fullname, User.nickname)
>>> session.query(User).from_statement(stmt).params(name='ed').all()
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>]
或者
>>> stmt = text("SELECT name, id FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id)
>>> session.query(User.id, User.name).\
... from_statement(stmt).params(name='ed').all()
[(1, u'ed')]
計(jì)數(shù)
>>> session.query(User).filter(User.name.like('%ed')).count()
2
>>> from sqlalchemy import func
>>> session.query(func.count(User.name), User.name).group_by(User.name).all()
[(1, u'ed'), (1, u'fred'), (1, u'mary'), (1, u'wendy')]
>>> session.query(func.count('*')).select_from(User).scalar()#SELECT count(*) FROM table
4
>>> session.query(func.count(User.id)).scalar()#也可以這樣
4
建立關(guān)聯(lián)
User表中的每一用戶可以存儲任意數(shù)量的與其用戶名相關(guān)聯(lián)的電子郵件地址暴凑。從User表到存儲電子郵件地址（我們稱之為Address）的新表存在一對多關(guān)聯(lián)峦甩。我們定義這個表及其映射類Address:
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy.orm import relationship
>>> class Address(Base):
... __tablename__ = 'addresses'
... id = Column(Integer, primary_key=True)
... email_address = Column(String, nullable=False)
... user_id = Column(Integer, ForeignKey('users.id'))
...
... user = relationship("User", back_populates="addresses")
...
... def __repr__(self):
... return "<Address(email_address='%s')>" % self.email_address
>>> User.addresses = relationship(
... "Address", order_by=Address.id, back_populates="user")
上面的類引入了ForeignKey，它是一個應(yīng)用于列的指令，指示應(yīng)該將該列中的值限制在指定的另一張表的對應(yīng)列中可能出現(xiàn)的值的范圍內(nèi)凯傲。這是關(guān)系數(shù)據(jù)庫的核心特性犬辰，也是將原本不連接的表集合轉(zhuǎn)換為具有豐富的重疊關(guān)系的粘合劑。上面的外鍵表示Address中的user_id列應(yīng)該被約束為User表中的id列冰单，即它的主鍵幌缝。
第二個指令稱為relationship()，它告訴ORM诫欠，Address類本身應(yīng)該使用屬性Address. user鏈接到User類涵卵。relationship()使用兩個表之間的外鍵關(guān)系來確定這個鏈接的性質(zhì)，從而確定該Address與User的關(guān)系是多對一荒叼。
附加的relationship()放在User類下新建的User.address屬性上轿偎。在這兩個relationship()中，back_populates指定為互相引用的屬性名被廓；通過這樣做坏晦，每個relationship()可以用back_populates傳遞關(guān)聯(lián)：一端Address.user引用一個user實(shí)例，另一端則是User.addresses引用Address實(shí)例的列表嫁乘，我們可以通過User表的back_populates反向查出所有它在addresses表里的關(guān)聯(lián)項(xiàng)昆婿。
我們需要在數(shù)據(jù)庫中創(chuàng)建地址表，所以我們將從我們的元數(shù)據(jù)發(fā)出另一個創(chuàng)建亦渗，它將跳過已經(jīng)創(chuàng)建的表:
>>> Base.metadata.create_all(engine)
現(xiàn)在挖诸，當(dāng)我們創(chuàng)建一個User時，將出現(xiàn)一個空白addresses集合法精。在這里可以使用各種集合類型多律，比如集合和字典，但是在默認(rèn)情況下搂蜓，集合是一個Python列表狼荞。
>>> jack = User(name='jack', fullname='Jack Bean', nickname='gjffdd')
>>> jack.addresses
[]
我們可以自由地在User對象上添加Address對象：
>>> jack.addresses = [
... Address(email_address='jack@google.com'),
... Address(email_address='j25@yahoo.com')]
當(dāng)使用雙向關(guān)系時，在一個方向上添加的元素在另一個方向上自動變得可見帮碰。此行為基于on-change事件屬性發(fā)生相味，并在Python中進(jìn)行計(jì)算，不使用任何SQL：
>>> jack.addresses[1]
<Address(email_address='j25@yahoo.com')>
>>> jack.addresses[1].user
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
我們現(xiàn)在就向數(shù)據(jù)庫添加并提交Jack Bean殉挽。jack和對應(yīng)addresses集合中的兩個Address成員都被一次性添加到會話中丰涉，這個添加關(guān)聯(lián)數(shù)據(jù)的過程叫級聯(lián)（cascading）：
>>> session.add(jack)
>>> session.commit()
查詢 Jack，只返回 Jack斯碌。沒有關(guān)于Jack的addresses的SQL：
>>> jack = session.query(User).\
... filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
讓我們看看addresses集合一死。注意觀察SQL：
>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
當(dāng)我們訪問addresses集合時，SQL指令才被發(fā)出傻唾。這是一個延遲加載（lazy loading）關(guān)系的例子⊥洞龋現(xiàn)在已經(jīng)加載了addresses集合承耿，其行為與普通列表一樣。我們將簡要介紹優(yōu)化這個集合的加載的方法伪煤。
基于外鍵關(guān)聯(lián)的多表聯(lián)合查詢
要在User和Address之間構(gòu)造一個簡單的隱式連接加袋，可以使用Query.filter()將它們的相關(guān)列等同起來。下面我們用這個方法加載用戶和地址實(shí)體
>>> for u, a in session.query(User, Address).\
... filter(User.id==Address.user_id).\
... filter(Address.email_address=='jack@google.com').\
... all():
... print(u)
... print(a)
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
<Address(email_address='jack@google.com')>
另一方面抱既，實(shí)際的SQL JOIN語法最容易使用Query.join()方法實(shí)現(xiàn)：
>>> session.query(User).join(Address).\
... filter(Address.email_address=='jack@google.com').\
... all()
[<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>]
Query.join()知道如何在用戶和地址之間進(jìn)行連接职烧，因?yàn)樗鼈冎g只有一個外鍵。如果沒有外鍵防泵，或者有多個外鍵阳堕，那么Query.join()在使用以下一種形式時工作得更好：
query.join(Address, User.id==Address.user_id) # 顯式關(guān)系
query.join(User.addresses) # 右連接指定關(guān)系
query.join(Address, User.addresses) # 顯式關(guān)系指定
正如您所期望的，使用Query.outerjoin()函數(shù)對外連接使用了相同的思想：
query.outerjoin(User.addresses) # LEFT OUTER JOIN
join()方法通常從實(shí)體列表中最左邊的項(xiàng)進(jìn)行連接择克，如果省略了ON子句，或者ON子句是一個普通的SQL表達(dá)式前普。要控制連接列表中的第一個實(shí)體肚邢，可以使用Query.select_from()方法：
query = session.query(User, Address).select_from(Address).join(User)
使用別名
在跨多個表進(jìn)行查詢時，如果需要多次引用同一個表拭卿，SQL通常要求該表使用另一個名稱作為別名骡湖，這樣就可以根據(jù)該表的其他情況對其進(jìn)行區(qū)分。使用aliased()構(gòu)造支持這一點(diǎn)峻厚。在使用aliased()連接關(guān)系時响蕴，可以使用特殊屬性方法PropComparator.of_type()來更改關(guān)系連接的目標(biāo)，以引用給定的aliased()對象惠桃。下面浦夷，我們兩次連接到Address實(shí)體，以定位具有兩個不同電子郵件地址的用戶：
>>> from sqlalchemy.orm import aliased
>>> adalias1 = aliased(Address)
>>> adalias2 = aliased(Address)
>>> for username, email1, email2 in \
... session.query(User.name, adalias1.email_address, adalias2.email_address).\
... join(User.addresses.of_type(adalias1)).\
... join(User.addresses.of_type(adalias2)).\
... filter(adalias1.email_address=='jack@google.com').\
... filter(adalias2.email_address=='j25@yahoo.com'):
... print(username, email1, email2)
jack jack@google.com j25@yahoo.com
除了使用PropComparator.of_type()方法之外辜王，還經(jīng)撑可以看到Query.join()方法通過單獨(dú)指示一個特定的目標(biāo)來連接：
# 等同于 query.join(User.addresses.of_type(adalias1))
q = query.join(adalias1, User.addresses)
使用子查詢
Query可以用作生成子查詢的語句。假設(shè)我們想加載User對象以及每個用戶有多少Address記錄的計(jì)數(shù)呐馆。生成這樣的SQL的最佳方法是獲得按用戶id分組的地址計(jì)數(shù)肥缔，并連接到父地址。在這種情況下汹来，我們使用左外連接续膳，這樣我們可以為那些沒有任何地址的用戶返回行，例如：
SELECT users.*, adr_count.address_count FROM users LEFT OUTER JOIN
(SELECT user_id, count(*) AS address_count
FROM addresses GROUP BY user_id) AS adr_count
ON users.id=adr_count.user_id
把這段SQL換成我們的query：
>>> from sqlalchemy.sql import func
>>> stmt = session.query(Address.user_id, func.count('*').\
... label('address_count')).\
... group_by(Address.user_id).subquery()
func生成SQL函數(shù)收班，Query上的subquery()方法生成一個SQL表達(dá)式構(gòu)造坟岔，表示嵌入在別名中的SELECT語句(實(shí)際上是Query.statement.alias()的簡寫形式)。
一旦有了語句闺阱，它的行為就像一個表結(jié)構(gòu)炮车，就像我們在本教程開始時為用戶創(chuàng)建的表結(jié)構(gòu)一樣。語句的列可以通過一個名為c的屬性訪問：
>>> for u, count in session.query(User, stmt.c.address_count).\
... outerjoin(stmt, User.id==stmt.c.user_id).order_by(User.id):
... print(u, count)
<User(name='ed', fullname='Ed Jones', nickname='eddie')> None
<User(name='wendy', fullname='Wendy Williams', nickname='windy')> None
<User(name='mary', fullname='Mary Contrary', nickname='mary')> None
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')> None
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')> 2
從子查詢中選擇對象
上面，我們只是從子查詢中選擇了一個包含列的結(jié)果
如果我們希望我們的子查詢映射到一個對象呢瘦穆？
可以使用aliased()將映射類的“別名”關(guān)聯(lián)到子查詢
>>> stmt = session.query(Address).\
... filter(Address.email_address != 'j25@yahoo.com').\
... subquery()
>>> adalias = aliased(Address, stmt)
>>> for user, address in session.query(User, adalias).\
... join(adalias, User.addresses):#這樣就返回Address的對象纪隙，而不是列了
... print(user)
... print(address)
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
<Address(email_address='jack@google.com')>
如果不加aliased()是這樣的，stmt被拆成列：
>>> stmt = session.query(Address).\
... filter(Address.email_address != 'j25@yahoo.com').\
... subquery()
>>> for user in session.query(User, stmt).\
... join(stmt, User.addresses):
... print(user)
(<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>, 1, 'jack@google.com', 5)
判斷數(shù)據(jù)的存在性
SQL中的EXISTS關(guān)鍵字是一個布爾運(yùn)算符扛或，如果給定的表達(dá)式包含任何行绵咱，它將返回True。它可以在許多場景中代替join熙兔，并且對于定位相關(guān)表中沒有對應(yīng)行的行也很有用悲伶。
存在一個顯式的結(jié)構(gòu)，它看起來像這樣：
>>> from sqlalchemy.sql import exists
>>> stmt = exists().where(Address.user_id==User.id)
>>> for name, in session.query(User.name).filter(stmt):
... print(name)
jack
上面的語句換成可以沿著User.addresses的關(guān)聯(lián)使用Comparator.any()：
>>> for name, in session.query(User.name).\
... filter(User.addresses.any()):
... print(name)
jack
也可以給它個判斷依據(jù)住涉，來限制它返回的行數(shù)：
>>> for name, in session.query(User.name).\
... filter(User.addresses.any(Address.email_address.like('%google%'))):
... print(name)
jack
Comparator.has() 跟 Comparator.any()一樣都是用于多對一的關(guān)系：
>>> session.query(Address).\
... filter(~Address.user.has(User.name=='jack')).all()
[]
各種基于關(guān)系的操作：
#Comparator.__eq__() (多對一麸锉，等于):
query.filter(Address.user == someuser)
#Comparator.__ne__() (多對一，不等于):
query.filter(Address.user != someuser)
#IS NULL (多對一):
query.filter(Address.user == None)
#Comparator.contains() (用于一對多的集合):
query.filter(User.addresses.contains(someaddress))
#Comparator.any() (用于集合):
query.filter(User.addresses.any(Address.email_address == 'bar'))
# 也接受關(guān)鍵字參數(shù)
query.filter(User.addresses.any(email_address='bar'))
#Comparator.has() (Address.user.has(User.name=='ed')):
query.filter(Address.user.has(name='ed'))
#Query.with_parent() (用于任何關(guān)聯(lián)):
session.query(Address).with_parent(someuser, 'addresses')
預(yù)加載
回想一下前面我們在訪問User.addresses時演示的延遲加載操作舆声，在訪問User的address集合時才發(fā)出SQL花沉。如果您想要減少查詢的次數(shù)(在很多情況下是顯著減少)，我們可以對查詢操作應(yīng)用即時加載媳握。
SQLAlchemy提供了三種即時加載類型碱屁，其中兩種是自動加載的，第三種是自定義標(biāo)準(zhǔn)蛾找。這三種方法通常都是通過稱為query options的函數(shù)調(diào)用的娩脾，該函數(shù)通過query.options()方法向查詢提供關(guān)于我們希望如何加載各種屬性的附加指令。
Selectin Load
在這種情況下打毛，我們想要User.addresses應(yīng)盡快加載柿赊。加載一組對象及其相關(guān)的集合的一個很好的選擇是selectinload()，它發(fā)出第二個SELECT語句幻枉，該語句完全加載與剛剛加載的結(jié)果相關(guān)聯(lián)的集合闹瞧。名稱“selectin”源于SELECT語句使用IN子句來同時定位多個對象的相關(guān)行：
>>> from sqlalchemy.orm import selectinload
>>> jack = session.query(User).\
... options(selectinload(User.addresses)).\
... filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
Joined Load
這種加載方式會發(fā)出連接，默認(rèn)情況下是左外連接展辞，以便在一次操作中預(yù)先加載主對象及其相關(guān)對象（注意：這個不是JOIN）
>>> from sqlalchemy.orm import joinedload
>>> jack = session.query(User).\
... options(joinedload(User.addresses)).\
... filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
Explicit Join + Eagerload
第三種快速加載方式是奥邮，當(dāng)我們顯式地構(gòu)造一個JOIN以定位主行，并且希望將額外的表應(yīng)用于主對象上的相關(guān)對象或集合時罗珍。該特性是通過contains_eager()函數(shù)提供的洽腺，通常用于在需要對同一對象進(jìn)行篩選的查詢中預(yù)加載多對一對象。下面的例子做到了預(yù)先加載每一條Address表的記錄和與之關(guān)聯(lián)的User對象（例子中只篩選出名為“jack”的User）覆旱，使用contains_eager()將“User”表中的列預(yù)加載進(jìn)Address.user蘸朋。
>>> from sqlalchemy.orm import contains_eager
>>> jacks_addresses = session.query(Address).\
... join(Address.user).\
... filter(User.name=='jack').\
... options(contains_eager(Address.user)).\
... all()
>>> jacks_addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
>>> jacks_addresses[0].user
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
刪除操作
讓我們試著刪除jack，看看結(jié)果如何扣唱。我們會在會話中將對象標(biāo)記為已刪除藕坯，然后我們會發(fā)出一個計(jì)數(shù)查詢用以確認(rèn)团南。
>>> session.delete(jack)
sql
>>> session.query(User).filter_by(name='jack').count()
0
到目前為止，一切順利炼彪。那Jack的Address對象呢?
>>> session.query(Address).filter(
... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
2
它還在吐根！SQLAlchemy并沒有級聯(lián)刪除它。
配置級聯(lián)刪除
我們將在User.addresses的關(guān)聯(lián)上配置級聯(lián)選項(xiàng)辐马。雖然SQLAlchemy允許您在任何時間點(diǎn)向映射添加新的屬性和關(guān)系拷橘，但是在這種情況下，需要刪除現(xiàn)有的關(guān)系喜爷，所以我們需要完全刪除映射并重新開始—先關(guān)閉會話：
>>> session.close()
ROLLBACK
創(chuàng)建一個新的declarative_base()：
>>> Base = declarative_base()
接下來冗疮，我們將聲明User類，添加addresses關(guān)聯(lián)檩帐，包括級聯(lián)配置(我們也將不使用構(gòu)造函數(shù))：
>>> class User(Base):
... __tablename__ = 'users'
...
... id = Column(Integer, primary_key=True)
... name = Column(String)
... fullname = Column(String)
... nickname = Column(String)
...
... addresses = relationship("Address", back_populates='user',
... cascade="all, delete, delete-orphan")
...
... def __repr__(self):
... return "<User(name='%s', fullname='%s', nickname='%s')>" % (
... self.name, self.fullname, self.nickname)
然后我們重建Address類术幔，注意，在本例中我們已經(jīng)通過User類創(chuàng)建了Address.user關(guān)聯(lián)：
>>> class Address(Base):
... __tablename__ = 'addresses'
... id = Column(Integer, primary_key=True)
... email_address = Column(String, nullable=False)
... user_id = Column(Integer, ForeignKey('users.id'))
... user = relationship("User", back_populates="addresses")
...
... def __repr__(self):
... return "<Address(email_address='%s')>" % self.email_address
下面使用Query.get()加載user jack（它按主鍵加載）湃密，從對應(yīng)的addresses集合中移除一個地址將會導(dǎo)致該地址也被刪除：
# 按主鍵加載jack
>>> jack = session.query(User).get(5)
# 刪除一個地址(延遲加載被觸發(fā))
>>> del jack.addresses[1]
# 只剩下一個地址
>>> session.query(Address).filter(
... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
1
刪除Jack將同時刪除Jack和與用戶關(guān)聯(lián)的剩余Address：
>>> session.delete(jack)
>>> session.query(User).filter_by(name='jack').count()
0
>>> session.query(Address).filter(
... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
0
建立多對多的關(guān)聯(lián)
對于普通的多對多特愿，我們需要創(chuàng)建一個未映射的表結(jié)構(gòu)作為關(guān)聯(lián)表。這看起來像如下：
>>> from sqlalchemy import Table, Text
>>> # 關(guān)聯(lián)表
>>> post_keywords = Table('post_keywords', Base.metadata,
... Column('post_id', ForeignKey('posts.id'), primary_key=True),
... Column('keyword_id', ForeignKey('keywords.id'), primary_key=True)
... )
在上面勾缭，我們可以看到直接聲明一個表與聲明一個映射類有一點(diǎn)不同。Table是一個構(gòu)造函數(shù)目养，因此每個列參數(shù)之間用逗號分隔俩由。列對象的名稱也是顯式給定的，而不是從已分配的屬性名稱中獲取的癌蚁。
接下來幻梯，我們定義BlogPost和Keyword結(jié)構(gòu)，互相relationship()努释，每個都將post_keywords表引用為一個關(guān)聯(lián)表:
>>> class BlogPost(Base):
... __tablename__ = 'posts'
...
... id = Column(Integer, primary_key=True)
... user_id = Column(Integer, ForeignKey('users.id'))
... headline = Column(String(255), nullable=False)
... body = Column(Text)
...
... # many to many BlogPost<->Keyword
... keywords = relationship('Keyword',
... secondary=post_keywords,
... back_populates='posts')
...
... def __init__(self, headline, body, author):
... self.author = author
... self.headline = headline
... self.body = body
...
... def __repr__(self):
... return "BlogPost(%r, %r, %r)" % (self.headline, self.body, self.author)
>>> class Keyword(Base):
... __tablename__ = 'keywords'
...
... id = Column(Integer, primary_key=True)
... keyword = Column(String(50), nullable=False, unique=True)
... posts = relationship('BlogPost',
... secondary=post_keywords,
... back_populates='keywords')
...
... def __init__(self, keyword):
... self.keyword = keyword
我們還希望我們的BlogPost類有一個author字段碘梢。我們將把它添加為另一個雙向關(guān)系，因?yàn)閱蝹€用戶可能有很多博客文章伐蒂。當(dāng)我們訪問User.posts時煞躬，我們希望能夠在不加載整個集合的前提下進(jìn)一步filter結(jié)果。為此逸邦，我們使用了一個名為lazy='dynamic'的relationship()加載策略：
>>> BlogPost.author = relationship(User, back_populates="posts")
>>> User.posts = relationship(BlogPost, back_populates="author", lazy="dynamic")
創(chuàng)建新表
>>> Base.metadata.create_all(engine)
用法和我們之前做的沒有太大區(qū)別恩沛。讓我們給Wendy一些博客文章：
>>> wendy = session.query(User).\
... filter_by(name='wendy').\
... one()
>>> post = BlogPost("Wendy's Blog Post", "This is a test", wendy)
>>> session.add(post)
我們將關(guān)鍵字唯一地存儲在數(shù)據(jù)庫中，但我們知道我們還沒有任何關(guān)鍵字缕减，所以我們先創(chuàng)建它們
>>> post.keywords.append(Keyword('wendy'))
>>> post.keywords.append(Keyword('firstpost'))
我們現(xiàn)在可以用“firstpost”這個關(guān)鍵字查找所有的博客文章雷客。我們將使用any操作符來定位“其中的任何關(guān)鍵詞含有字符串' firstpost '的博客文章”：
>>> session.query(BlogPost).\
... filter(BlogPost.keywords.any(keyword='firstpost')).\
... all()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', nickname='windy')>)]
如果我們想要查找用戶wendy擁有的文章，我們可以告訴查詢將范圍縮小到User對象作為父對象:
>>> session.query(BlogPost).\
... filter(BlogPost.author==wendy).\
... filter(BlogPost.keywords.any(keyword='firstpost')).\
... all()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', nickname='windy')>)]
或者我們可以使用Wendy自己的posts關(guān)系桥狡，這是一種“動態(tài)”關(guān)系搅裙，直接從wendy.posts查詢：
>>> wendy.posts.\
... filter(BlogPost.keywords.any(keyword='firstpost')).\
... all()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', nickname='windy')>)]
（完）
參考資料：[SQLAlchemy 1.3 Documentation]https://docs.sqlalchemy.org/en/13/orm/tutorial.html
