SQLAlchemy的经典用法是使用它的ORM (Object Relational Mapper)。

• 它可以定义一个Python class来关联到一个database table，该类的实例关联到对应table的一行记录。
• 它可以通过修改Python class实例的属性从而修改关联的table记录

连接数据库使用create_engine()函数：sqlalchemy.create_engine(*args, **kwargs)。它返回一个Engine实例。

通常create_engine()函数的第一个参数是个url字符串，表示采用的数据库。格式为：'dialect[+driver]://user:password@host/dbname[?key=value..]'。其中：

• dialect是数据库的名字，可以为：'mysql'、'oracle'、'postgresql'、'sqlite'等等
• driver是DBAPI（驱动程序）的名字，可以为：'psycopg2'、'pyodbc'、'cx_oracle'等等

典型的url：

• postgresql:
  • default：create_engine('postgresql://scott:tiger@localhost/mydatabase')
  • psycopg2：create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')
  • pg8000：create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')
• mysql:
  • default：create_engine('mysql://scott:tiger@localhost/foo')
  • mysql-python：create_engine('mysql+mysqldb://scott:tiger@localhost/foo')
  • MySQL-connector-python：create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')
  • OurSQL:create_engine('mysql+oursql://scott:tiger@localhost/foo')
• oracle:
  • default：create_engine('oracle://scott:[email protected]:1521/sidname')
  • cx_oracle:create_engine('oracle+cx_oracle://scott:tiger@tnsname')
• SQL Server:
  • pyodbc:create_engine('mssql+pyodbc://scott:tiger@mydsn')
  • pymssql:create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')
• SQLite:create_engine('sqlite:///foo.db')（注意斜线和反斜线的转义处理）
  • 如果你想使用:memory:，则指定一个空URL即可，create_engine('sqlite://')

create_engine()的关键字参数有很多。他们有的是用于指定Engine的参数，有的是用于指定Engine底层的Dialect或者Pool的参数。常用的关键字参数有：

• case_sensitive：默认为True。如果为False则结果的column names是 case nsensitive，如row[’SomeColumn’]和row[’somecolumn’]相同
• echo：默认为False。如果为True则Engine会对执行的每一条statement进行log（默认是输出到sys.stdout）
• encoding：字符串编码，默认为'utf-8'
• max_overflow: the number of connections to allow in connection pool “overflow”, that is connections that can be opened above and beyond the pool_size setting.默认为10.只针对QueuePool有效
• pool：已提供的连接池，默认为None
• pool_recycle：连接池中的连接每隔多少秒（循环时间）就放回连接池中。默认为-1，意思是永不循环。
• pool_size：连接池中打开的连接的数量，默认为5。它可以用于QueuePool和SingletonThreadPool.在QueuePool中，0意味着没有限制。
• pool_timeout：从连接池中获取连接的超时时间（秒），默认为30。只用于QueuePool

create_engine()仅仅创建Engine实例，它并没有连接到数据库。只有当执行Engine.execute()或者Engine.connect()方法时，Engine才建立了数据库的连接。

一个Mapped Class映射了一张数据库的表。所有的Mapped Class都继承自一个基类，称之为declarative base class，该基类由sqlalchemy.ext.declarative.declarative_base()函数返回。

一个Mapped Class继承自declarative base class，并且有以下属性：

• 一个__tablename__属性，指定了该类所映射的数据库的表名

• 一个或者多个Column实例属性，该属性对应着所映射的数据库表的列

  它们和__tablename__均为Python描述符

• 其他的属性或者方法，这些属性或者方法与数据库表没有映射关系，即普通的Python属性或方法

  

sqlalchemy.schema.Column代表了database table中的一列。其初始化方法为： __init__(*args, **kwargs)，重要的关键字参数如下：

• name：列名。是个字符串。该参数是第一个位置参数，或者是关键字参数。如果该参数值字符串不包含大写字母，则列名视为case insensitive；否则列名视为case sensitive
• type_:列的类型，是一个TypeEngine子类的实例。当然你也可以传入一个TypeEngine子类（此时用它的初始化函数生成一个默认的实例）。该参数是第二个位置参数，或者是关键字参数。
  • 如果忽略该参数或者设为None，则默认为特殊的类型NullType
  • 如果该列使用了ForeignKey或者ForeignKeyConstraint，则该列的类型为外键引用的列的类型
  • 所有类型都在sqlalchemy.types中
• *args：其他的位置参数，包括Constraint、ForeignKey、ColumnDefault、Sequence实例等。
• autoincrement：列是否自增。如果为True，则该列（通常是存放整数）会随着INSERT操作自增，其值由DBAPI cursor.lastrowid属性返回。该属性只能用于满足下面条件的列：列存放的是整数，且列是primary key的组成部分，且不是外键
• default：一个标量，或者一个Python callable或者ColumnElement，用于该列的默认值。当插入记录且该列没有提供value时，使用这个default来插入
• index：如果为True则表示该列是indexed。如果你想对多格列的组合执行index，则需要使用Index显式构造
• nullable：如果为True，则表示database table中该列允许为NULL；否则为NOT NULL
• onupdate：一个标量，或者一个Python callable或者ClauseElement，用于该列的默认值。当update记录且该列没有提供value时，使用这个default来更新
• primary_key：如果为True则标记该列为主键。如果有多个列设置了primary_key=True，则这些列一起组合成主键
• server_default：一个FetchedValue实例，或者一个字符串，或者一个text()，表示该列DDL DEFAULT value
• quote：一个布尔值，表示是否对列名包裹上引号。如果是None则为默认行为：名字至少包含一个大写字母时，包裹上引号。
• unique：如果为True则表示该列有一个unique constraint。如果你想对多个列的组合指定unique，则需要通过UniqueConstraint或者Index显式构造
• system：如果为True则表示该列是database自动生成的列，因此该列并不会出现在CREATE TABLE statement中

当定义好了Mapped class之后，该类存储了database table的大量信息。这些信息称之为table metadata。我们可以通过该类的.__table__属性查看，该属性是一个Table实例。

sqlalchemy.schema.MetaData是Table实例的集合（可以在MetaData.tables字典查看），它持有许多Table实例并且它可以绑定到Engine或者Connection。你可以通过它来执行许多数据库操作，如创建数据表等。

通常你可以通过declarative base class的.metadata属性来获取一个MetaData实例。

Mapped Class实例提供了一个默认的构造方法，它自动接收关键字参数，这些关键字就是mapped column的identifier。你可以自定义一个.__init__()方法，此时自定义的方法会覆盖默认的行为。

如果某些mapped column并未赋值，则SQLAlchemy会自动生成default value；对于那些已经赋值的mapped column，SQLAlchemy会自动跟踪这些赋值，这是为了将最终的value插入或者更新到数据库中

ORM系统是在Session中处理数据库操作的。通常在create_engine()的时候我们就会通过工厂方法定义一个Session class:

• Session=sqlalchemy.orm.sessionmaker(bind=engine)：绑定了engine
• Session=sqlalchemy.orm.sessionmaker()：未绑定engine
  • 之后可以通过Session.configure(bind=engine)再绑定engine

通常创建Session的实例，用该实例来操作数据库

通过Session.add()方法可以添加mapped class实例。但是注意此时该实例是pending状态，还没有执行数据库插入操作。当下列行为之一发生时，才进行真正的数据库插入操作：

• 显式调用Session.flush()操作
• 查询数据库mapped class对应的表。此时先flush所有pending的实例执行数据库插入和更新，再执行查询操作
• 调用Session.commit()操作

你也可以通过Session.add_all()方法来添加一个mapped class实例的列表从而添加多个实例

在ORM思想中，一旦一个mapped class实例被添加到Session中，假设它的主键为pk1。则所有针对主键pk1的数据库查询操作都会返回该实例，而不是创建新的实例来返回。另外，如果你向该Session中添加另一个实例，而该实例的主键也是pk1则会抛出异常。

Session会跟踪那些被添加到它的那些mapped class实例。如果某个实例被修改过，则该修改并不是马上执行数据库的update操作，而是仅仅记录下来该实例被修改过。可以通过查看Session.dirty属性查看那些被修改过的mapped class实例。

一旦执行了Session.commit()方法，那么Session中的那些实例就会推送至数据库中：

• 如果数据库中没有该主键的记录，则执行插入操作
• 如果数据库中有该主键的记录，且该mapped class实例为dirty，则执行更新操作
• 执行commit transaction操作

commit之后，该Session引用的连接资源会放回至连接池中。如果后面继续用该Session来操作数据库（如add()），则会开启一个新的transaction并且重新从连接池中取出连接

commit之后，那么之后在一个新的transaction中访问数据时，他会刷新数据（从数据库中获取）从而保持mapped class实例的最新的状态。

由于Session是工作于transaction，你可以通过Session.rollback()执行回滚操作。

通过Session.query()方法能够创建一个Query对象。这个方法可以使用各种类型的参数，如mapped class或者mapped column descriptor（如User和User.name）。通常我们在迭代环境中使用Query对象。

• for instance in session.query(User).order_by(User.id)：迭代产生的是User类的实例

• for instance in session.query(User.name, User.fullname).order_by(User.id)：迭代产生的是一个元组

  只要query()参数为多个mapped class或者mapped column descriptor，或者它们的组合时，迭代就产生一个named tuple。这个元组可以当作普通的Python对象，其各个属性就是mapped column descriptor以及mapped class（属性名就是mapped column descriptor名字以及mapped class的类名）

  

• 迭代Query对象产生命名元组时，你可以自定义这个命名元组的名字，通过在query(User.name.label('my_name'))，利用label()，则该命名元组有个my_name属性，对应的是User的name属性。

• 你可以通过sqlalchemy.orm.aliased来重命名一个mapped class。如user_alias = aliased(User, name='user_alias')，此后user_alias就可以代表User（对应于SQL语句中的AS表达式）

  

你可以对Query对象进行Python array slice来实现SQL的LIMIT和OFFSET，分片操作返回的是查询的结果数组。你也可以通过Query.order_by()方法来实现ORDER BY，其中order_by()方法的参数为mapped column descriptor（如User.id），该方法返回一个新的Query对象。

Query对象的大多数方法都会返回一个新的Query对象

如果你想对查询结果进行更复杂的筛选，则可以使用Query.filter()方法。该方法的参数可以对mapped column descriptor（如User.id）进行复杂的运算（比如相等比较，不等比较等）。 常用的filter operator有：

• 相等比较equals，如query.filter(User.name == 'ed')

• 不等比较equals，如query.filter(User.name != 'ed')

• LIKE，如query.filter(User.name.like('%ed%'))

• IN，如query.filter(User.name.in_(['ed','wendy','jack']))

  或者用query.filter(User.name.in_(session.query(User.name).filter(User.name.like('%ed%'))))

• Not IN，如query.filter(~User.name.in_(['ed','wendy','jack']))

• IS NULL，如query.filter(User.name == None)

  或者用query.filter(User.name.is_(None))

• IS NOT NULL,如query.filter(User.name != None)

  或者用query.filter(User.name.isnot(None))

• AND，有三种形式：

  • 利用and_函数：query.filter(sqlalchemy.and_(User.name == 'ed', User.fullname == 'Ed Jones'))（注意不是Python的and操作符）
  • 利用filter的多表达式：query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
  • 利用filter链：query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')

• OR：利用or_函数：query.filter(sqlalchemy.or_(User.name == 'ed', User.fullname == 'Ed Jones'))（注意不是Python的or操作符）

• MATCH：query.filter(User.name.match('wendy'))。该过滤器的行为会因数据库的不同而呈现不同的行为，其中SQLite不支持这种过滤。对于大多数数据库，该过滤器相当于MATCH或者CONTAINS

  

有一些Query方法会立即执行SQL查询并且返回查询结果。

• Query.all()方法返回查询记录，结果是一个列表

• Query.first()，它执行的是limit one，返回一个标量值，该标量是第一条查询记录

• Query.one()，它首先获取所有的记录：

  • 如果有超过一条记录，则抛出MultipleResultsFound异常
  • 如果一条记录也没有，则抛出NoResultFound异常
  • 如果只有一条记录，则返回一个标量值，该标量就是查询到的记录

• Query.one_or_none()，它首先获取所有的记录：

  • 如果有超过一条记录，则抛出MultipleResultsFound异常
  • 如果一条记录也没有，则返回None
  • 如果只有一条记录，则返回一个标量值，该标量就是查询到的记录

• Query.scalar()，它返回记录的第一条数据作为结果返回

  • 如果有超过一条记录，则抛出MultipleResultsFound异常
  • 如果一条记录也没有，则返回None
  • 如果只有一条记录，则返回一个标量值，该标量就是查询到的记录

  

前面介绍的Query.filter()方法和Query.order_by()方法都用的是mapped column descriptor（如User.id），你也可以在这些地方使用字符串，这看起来更像是SQL。

通过sqlalchemy.text()函数中传入字符串字面量可以实现该做法。
session.query(User).filter(text("id<224")).order_by(text("id"))等价于 session.query(User).filter(User.id<224).order_by(User.id)

当然224这个数字你也可以通过绑定参数来动态传入。这是通过Query.params()方法绑定的。
session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(User.id)。参数是以:key字符串的形式实现的。在params(key=value)绑定的。

如果你想通过字符串指定完整的SQL查询语句，可以使用Query.from_statement()方法。
session.query(User).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed')

Query.count()方法执行的是SQL的COUNT，它用于确定能返回多少条查询记录。在SQLAlchemy内部，它首先执行一个子查询，然后在这个子查询的结果集之上在执行COUNT查询。这种实现方法比较低效，更高效的方法是SELECT count(*) FROM table。我们可以通过sqlalchemy.func.count()来实现这种高效的做法：

• session.query(func.count(User.name)).all():返回的是一个列表，列表元素为元组，元组内容对应于func.count()的参数。

• session.query(func.count(User.name)).scalar()：返回的是标量，指定有多少个User.name

• session.query(func.count('*')).select_from(User).scalar():返回的是标量，指定有多少条记录

  假设有100条记录，可能只有10个User.name，因为User.name可以重复

  必须指定.select_from(User)，否则Query压根不知道要查询哪个表

• session.query(func.count(User.id)).scalar():因为id是主键，因此主键有多少条，记录就有多少条

  

在mapped class的定义过程中，如果有外键，我们可以将该列指定为ForeignKey，如：

class Address(Base):
	__tablename__='addresses'
	...
	user_id=Column(Integer,ForeignKey('users.id'))
	...

其中ForeignKey的参数为字符串，指定了外键所在的表（User.__tablename__指定的）以及列名（id）

我们可以通过sqlalchemy.orm.relationship()定义两个mapped class之间的关系：一对多、多对一、多对多、一对一。该函数返回一个RelationshipProperty。你可以在mapped class类内定义关系，也可以在类外定义关系。如：

class Address(Base):
	__tablename__='addresses'
	...
	user_id=Column(Integer,ForeignKey('users.id'))
	user=relationship("User",back_populates='addresses')
	...
User.addresses=relationship("Address",back_populates='user',order_by=Address.id)

• 第一个位置参数为关联的mapped class类名

• back_populates参数：该参数指出了对方的关联名。如Address的关联名为user。那么你通过User.addresses获取的Address的user属性就指向本User。

• 关联结合外键可以自动确定是一对多关系/多对一关系。如例子中多个Address对应一个User。因此User.addresses是一个Python列表，而Address.user是一个User实例

  

默认情况下，User.addresses返回的是一个Python列表（你可以配置它为一个字典，或者一个Python set）。你可以修改User.addresses，在二元关系中，你针对某一端的实例修改之后，另一端的实例也会自动被修改从而匹配二元关系。

当我们查询User表的时候，并不会同时查询Address表；只有当访问User.addresses属性时，才会有SQL查询发生在Address表

通常我们可以使用Query.filter()来执行联合查询。它会隐式的执行 JOIN。如

session.query(User, Address).filter(User.id==Address.user_id).\
filter(Address.email_address=='[email protected]').all()

你也可以显式的执行 JOIN 操作：如：session.query(User).join(Address).filter(Address.email_address=='[email protected]').all()，这里隐含着一个条件，即User和 Address之间只有一个 foreign key。如果没有外键，或者有多个外键，则你必须采用下列方法：

• query(User).join(Address,User.id==Address.user_id)：显式指定条件
• query(User).join(User.addresses)：specify relationship from left to right
• query(User).join(Address,User.addresses)：同上，但是显式指定关系
• query(User).join('addresses')：同上，但是用字符串代替

当然你也可以用 query(User).outerjoin(User.addresses)来指定外连接

当使用 join 时，你必须指定多个表。但是有时候是表自己跟自己 join ，此时你必须使用别名。如：

adalias1 = aliased(Address)
adalias2 = aliased(Address)
for username, email1, email2 in \
	session.query(User.name, adalias1.email_address, adalias2.email_address).\
	join(adalias1, User.addresses).\
	join(adalias2, User.addresses).\
	filter(adalias1.email_address=='[email protected]').\
	filter(adalias2.email_address=='[email protected]'):
 		print(username, email1, email2)

如果我们想知道每个 User有几个Address，则我们可以使用子查询。方法为：

from sqlalchemy.sql import func
stmt = session.query(Address.user_id, func.count('*').\
 	label('address_count')). group_by(Address.user_id).subquery()
session.query(User, stmt.c.address_count).\
  outerjoin(stmt, User.id==stmt.c.user_id).order_by(User.id).all()

• func生成SQL function
• Query.subquery()方法生成一个 SELECT statement
• stmt.c是子查询的列属性。通过它可以引用子查询的各列
• 这里之所以用外连接，是因为可能有某些User没有 Address

如果子查询的结果是一条mapped class记录，则你要用别名机制将它放入父查询中：

stmt = session.query(Address).\
	filter(Address.email_address != '[email protected]').subquery()
adalias = aliased(Address, stmt)
session.query(User, adalias).join(adalias, User.addresses).all()

通常你可以在Query.filter()中使用 exists 条件：

from sqlalchemy.sql import exists
session.query(User.name).filter(exists().where(Address.user_id==User.id)).all()

也有一些操作会自动使用 EXISTS 语句。上述的代码可以替代为：

session.query(User.name).filter(User.addresses.any()).all():

这里 any()可以提供筛选条件，如session.query(User.name).filter(User.addresses.any(Address.email_address.like('%google%'))) 如果是多对一关系，你可以使用has()，如：session.query(Address).filter(~Address.user.has(User.name=='jack')).all()

下面是常用的过滤：

• query.filter(Address.user == someuser)：多对一关系的相等比较
• query.filter(Address.user != someuser)：多对一关系的不相等比较
• query.filter(Address.user == None)：IS NULL, 多对一关系的比较
• query.filter(User.addresses.contains(someaddress))：一对多关系的包含过滤
• query.filter(User.addresses.any(Address.email_address == 'bar'))：一对多关系的包含过滤
• query.filter(Address.user.has(name='ed'))：多对一关系的包含过滤
• session.query(Address).with_parent(someuser, 'addresses')：用于任意关系的比较

sqlalchemy 默认行为是懒加载，即：只有在必要的时候才执行 SQL 操作。当然如果你也可以预加载，即还没有用到的时候就提前执行SQL操作，这样的好处是可以在一条查询语句中返回多个查询值，从而减少查询的次数。

有三种预加载：两种是隐式的自动的行为、一种是显式指定。所有的这些预加载都是利用Query.options()方法指定。

通过在 .options()中指定 orm.subqueryload()，可以实现预加载。它利用一个子查询实现。

from sqlalchemy.orm import joinedload
jack = session.query(User).options(joinedload(User.addresses)).filter_by(name='jack').one()

通过.options()中指定 orm.joinedLoad()，可以实现预加载 。它利用一个 LEFT OUTER JOIN 来实现。

from sqlalchemy.orm import subqueryload
jack = session.query(User).options(subqueryload(User.addresses)).filter_by(name='jack').one()

显式利用 join 和 orm.contains_eager()实现预加载。

from sqlalchemy.orm import contains_eager
jacks_addresses = session.query(Address).join(Address.user).filter(User.name=='jack').\
	options(contains_eager(Address.user)).all()

通过 session.delete(mapped_class_obj) 可以删除一条记录。但是，如果该记录有外键关联，则它并不删除其他表中关联的记录。

你可以配置 cascade 来实现关联记录的删除行为：

class User(Base):
  __tablename__ = 'users'
	
  id = Column(Integer, primary_key=True)
  name = Column(String)
  fullname = Column(String)
  password = Column(String)	
  addresses = relationship("Address", back_populates='user',
  cascade="all, delete, delete-orphan")
...
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")

这样当你 session.delete(user1)时，该user1对应的address记录会相应同步地被删除。

定义一个多对多关系需要使用一个未绑定的Table对象作为中介。如：

• 定义一个未绑定 Table，通过 Table()显式构造：

from sqlalchemy import Table, Text
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))

• 定义一端的 mapped class：

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')
	...

• 定义另一端的 mapped class：

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')
	...

这样通过在 relationship中指定 secondary关键字参数（一个未绑定的 Table） 就能实现多对多关系。