本文將介紹sqlalchemy的高級用法。
首先創(chuàng)建數(shù)據(jù)庫，在這里一個user對應(yīng)多個address，因此需要在address上增加user_id這個外鍵（一對多）。
#!/usr/bin/env python# encoding: utf-8from sqlalchemy import create_enginefrom sqlalchemy import Columnfrom sqlalchemy import Integerfrom sqlalchemy import Stringfrom sqlalchemy import ForeignKeyfrom sqlalchemy.orm import backreffrom sqlalchemy.orm import sessionmakerfrom sqlalchemy.orm import relationship, backreffrom sqlalchemy.ext.declarative import declarative_baseBase = 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) 接下來，調(diào)用user和address來添加數(shù)據(jù)，
>>> jack = User(name='jack')>>> jack.addressTraceback (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_12015-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_12015-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_12015-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>>> 此時，查看數(shù)據(jù)庫，可以得到剛才插入的數(shù)據(jù)，
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的話，可以直接聯(lián)表查詢，
>>> 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 = %s2015-08-19 14:02:02,878 INFO sqlalchemy.engine.base.Engine ('test@test.com',)[('jack', 'test@test.com')]在sqlalchemy中提供了Queqy.join()函數(shù)，
>>> 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 %s2015-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().name2015-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 %s2015-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().addresses2015-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 %s2015-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.id2015-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 conditionquery.join(User.addresses) # specify relationship from left to rightquery.join(Address, User.addresses) # same, with explicit targetquery.join('addresses') >>> from sqlalchemy.orm import aliased>>> adalias1 = aliased(Address)假設(shè)我們需要這樣一個查詢，
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()# 聯(lián)接子句，注意子句中需要使用c來調(diào)用字段內(nèi)容>>> 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_countFROM users LEFT OUTER JOIN (SELECT addresses.user_id AS user_id, count(%s) AS address_countFROM addresses GROUP BY addresses.user_id) AS anon_1 ON users.id = anon_1.user_id2015-08-19 14:42:53,425 INFO sqlalchemy.engine.base.Engine ('*',)[('jack', 2L)]>>>query.filter(User.addresses.contains(someaddress))>>> session.delete(jack)>>> session.query(User).filter_by(name='jack').count()0在上面的例子中，刪除了user－jack，但是address中的數(shù)據(jù)并沒有刪除。
cascade字段用來
addresses = relationship("Address", backref='user', cascade="all, delete, delete-orphan")