sqlalchemy(二)高級用法
本文將介紹sqlalchemy的高級用法。
外鍵以及relationship
首先創建數據庫，在這里一個user對應多個address，因此需要在address上增加user_id這個外鍵（一對多）。
#!/usr/bin/env python
# encoding: utf-8
from sqlalchemy import create_engine
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import ForeignKey
from sqlalchemy.orm import backref
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(32))
addresses = relationship("Address", order_by="Address.id", backref="user")
class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email_address = Column(String(32), nullable=False)
user_id = Column(Integer, ForeignKey('users.id'))
#user = relationship("User", backref=backref('addresses', order_by=id))
engine = create_engine('mysql://root:root@localhost:3306/test', echo=True)
#Base.metadata.create_all(engine)
接下來，調用user和address來添加數據，
>>> jack = User(name='jack')
>>> jack.address
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: 'User' object has no attribute 'address'
>>> jack.addresses
[]
>>> jack.addresses = [Address(email_address='test@test.com'), Address(email_address='test1@test1.com')]
>>> jack.addresses
[<demo.Address object at 0x7f2536564f90>, <demo.Address object at 0x7f2535dc71d0>]
>>> session.add(jack)
>>> session.commit()
2015-08-19 13:45:36,237 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'
2015-08-19 13:45:36,237 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,238 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()
2015-08-19 13:45:36,238 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8' and `Collation` = 'utf8_bin'
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_1
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_1
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8) COLLATE utf8_bin AS anon_1
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,241 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-08-19 13:45:36,242 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name) VALUES (%s)
2015-08-19 13:45:36,242 INFO sqlalchemy.engine.base.Engine ('jack',)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine INSERT INTO addresses (email_address, user_id) VALUES (%s, %s)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine ('test@test.com', 1L)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine INSERT INTO addresses (email_address, user_id) VALUES (%s, %s)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine ('test1@test1.com', 1L)
2015-08-19 13:45:36,244 INFO sqlalchemy.engine.base.Engine COMMIT
>>>
此時，查看數據庫，可以得到剛才插入的數據，
mysql> select * from users;
+----+------+
| id | name |
+----+------+
| 1 | jack |
+----+------+
1 row in set (0.00 sec)
mysql> select * from addresses;
+----+-----------------+---------+
| id | email_address | user_id |
+----+-----------------+---------+
| 1 | test@test.com | 1 |
| 2 | test1@test1.com | 1 |
+----+-----------------+---------+
2 rows in set (0.00 sec)
join查詢
如果不使用join的話，可以直接聯表查詢，
>>> session.query(User.name, Address.email_address).filter(User.id==Address.user_id).filter(Address.email_address=='test@test.com').all()
2015-08-19 14:02:02,877 INFO sqlalchemy.engine.base.Engine SELECT users.name AS users_name, addresses.email_address AS addresses_email_address
FROM users, addresses
WHERE users.id = addresses.user_id AND addresses.email_address = %s
2015-08-19 14:02:02,878 INFO sqlalchemy.engine.base.Engine ('test@test.com',)
[('jack', 'test@test.com')]
在sqlalchemy中提供了Queqy.join()函數，
>>> session.query(User).join(Address).filter(Address.email_address=='test@test.com').first()
2015-08-19 14:06:56,624 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name
FROM users INNER JOIN addresses ON users.id = addresses.user_id
WHERE addresses.email_address = %s
LIMIT %s
2015-08-19 14:06:56,624 INFO sqlalchemy.engine.base.Engine ('test@test.com', 1)
<demo.User object at 0x7f9a74139a10>
>>> session.query(User).join(Address).filter(Address.email_address=='test@test.com').first().name
2015-08-19 14:07:04,224 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name
FROM users INNER JOIN addresses ON users.id = addresses.user_id
WHERE addresses.email_address = %s
LIMIT %s
2015-08-19 14:07:04,224 INFO sqlalchemy.engine.base.Engine ('test@test.com', 1)
'jack'
>>> session.query(User).join(Address).filter(Address.email_address=='test@test.com').first().addresses
2015-08-19 14:07:06,534 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name
FROM users INNER JOIN addresses ON users.id = addresses.user_id
WHERE addresses.email_address = %s
LIMIT %s
2015-08-19 14:07:06,534 INFO sqlalchemy.engine.base.Engine ('test@test.com', 1)
2015-08-19 14:07:06,535 INFO sqlalchemy.engine.base.Engine SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id
FROM addresses
WHERE %s = addresses.user_id ORDER BY addresses.id
2015-08-19 14:07:06,535 INFO sqlalchemy.engine.base.Engine (1L,)
[<demo.Address object at 0x7f9a74139350>, <demo.Address object at 0x7f9a741390d0>]
>>>
注意，上面的用法的前提是存在外鍵的情況下，如果沒有外鍵，那么可以使用,
query.join(Address, User.id==Address.user_id) # explicit condition
query.join(User.addresses) # specify relationship from left to right
query.join(Address, User.addresses) # same, with explicit target
query.join('addresses')
表的別名
>>> from sqlalchemy.orm import aliased
>>> adalias1 = aliased(Address)
子查詢
假設我們需要這樣一個查詢，
mysql> 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;
+----+------+---------------+
| id | name | address_count |
+----+------+---------------+
| 1 | jack | 2 |
+----+------+---------------+
1 row in set (0.00 sec)
# 生成子句，等同於（select user_id ... group_by user_id）
>>> sbq = session.query(Address.user_id, func.count('*').label('address_count')).group_by(Address.user_id).subquery()
# 聯接子句，注意子句中需要使用c來調用字段內容
>>> session.query(User.name, sbq.c.address_count).outerjoin(sbq, User.id==sbq.c.user_id).all()
2015-08-19 14:42:53,425 INFO sqlalchemy.engine.base.Engine SELECT users.name AS users_name, anon_1.address_count AS anon_1_address_count
FROM users LEFT OUTER JOIN (SELECT addresses.user_id AS user_id, count(%s) AS address_count
FROM addresses GROUP BY addresses.user_id) AS anon_1 ON users.id = anon_1.user_id
2015-08-19 14:42:53,425 INFO sqlalchemy.engine.base.Engine ('*',)
[('jack', 2L)]
>>>
包含contains
query.filter(User.addresses.contains(someaddress))
數據刪除delete
>>> session.delete(jack)
>>> session.query(User).filter_by(name='jack').count()
0
外鍵配置
在上面的例子中，刪除了user－jack，但是address中的數據並沒有刪除。
cascade字段用來
addresses = relationship("Address", backref='user',
cascade="all, delete, delete-orphan")