SQLAlchemy-技术文档

sqlalchemy的使用
1.sqlalchemy的使用
数据库表结构
[
    ('1', 'Michael'),
    ('2', 'Bob'),
    ('3', 'Adam')
]
ORM技术
ORM技术：Object-Relational Mapping，把关系数据库的表结构映射到对象上
在Python中，最有名的ORM框架是SQLAlchemy
用法setp1：第一步，导入SQLAlchemy，并初始化DBSession：
&& 导入:
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base


&& 创建对象的基类:
Base = declarative_base()

&& 定义User对象:
class User(Base):
    && 表的名字:
    __tablename__ = 'user'

    && 表的结构:
    id = Column(String(20), primary_key=True)
    name = Column(String(20))

&& 初始化数据库连接:
engine = create_engine('mysql+mysqlconnector://root:password@localhost:3306/test')
&& 创建DBSession类型:
DBSession = sessionmaker(bind=engine)
create_engine()用来初始化数据库连接。SQLAlchemy用一个字符串表示连接信息：
'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
setp2:使用session对象  --  Session对象可视为当前数据库连接。
&& 创建session对象:
session = DBSession()
&& 创建新User对象:
new_user = User(id='5', name='Bob')
&& 添加到session:
session.add(new_user)
&& 提交即保存到数据库:
session.commit()
&& 关闭session:
session.close()
step3:查询数据库表
如何从数据库表中查询数据呢？有了ORM，查询出来的可以不再是tuple，而是User对象。SQLAlchemy提供的查询接口如下
&& 创建Session:
session = DBSession()
&& 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
user = session.query(User).filter(User.id=='5').one()
&& 打印类型和对象的name属性:print 'type:', type(user)
print 'name:', user.name
&& 关闭Session:
session.close()


运行结果如下：
type: <class '__main__.User'>
name: Bob
可见，ORM就是把数据库表的行与相应的对象建立关联，互相转换。
setp4：ORM的外键关联 -- relationship(向外连接), ForeignKey(连接主键)
由于关系数据库的多个表还可以用外键实现一对多、多对多等关联，相应地，ORM框架也可以提供两个对象之间的一对多、多对多等功能。
例如，如果一个User拥有多个Book，就可以定义一对多关系如下：
class User(Base):
  __tablename__ = 'user'
  id = Column(String(20), primary_key=True)
  name = Column(String(20))
  && 一对多:
  books = relationship('Book')
class Book(Base):
  __tablename__ = 'book'
  id = Column(String(20), primary_key=True)
  name = Column(String(20))
  && “多”的一方的book表是通过外键关联到user表的:
  user_id = Column(String(20), ForeignKey('user.id'))
当我们查询一个User对象时，该对象的books属性将返回一个包含若干个Book对象的list。
小结
ORM框架的作用就是把数据库表的一行记录与一个对象互相做自动转换。
正确使用ORM的前提是了解关系数据库的原理

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker

db_url = "mysql+mysqlconnector://root:@localhost:3306/test"

&& 通过数据库连接url创建数据库引擎
&& 如果想回显SQLAlchemy操作数据库的日志，设置echo=True
engine = create_engine(db_url, echo=True)

&& 通过数据库引擎绑定元信息
metadata = MetaData(engine)

&& 通过绑定数据库引擎获取数据库会话类
Session = sessionmaker(bind=engine)

&& 获取数据库会话
session = Session()

import sqlalchemy
from datetime import datetime
from sqlalchemy import create_engine,Column,Integer,String,Text,Boolean,Date,DateTime,Float
from sqlalchemy.ext.declarative import declarative_base
&& ======================================================================================================================

&& 数据库的数据类型和常用参数示例

&& 1. Integer：整形，映射到数据库中的int类型。
&& 2. String：字符类型，映射到数据库中的varchar类型，使用时，需要提供一个字符长度。
&& 3. Text：文本类型，映射到数据库中的text类型。
&& 4. Boolean：布尔类型，映射到数据库中的bool类型，在使用的时候，传递`True/False`进去。
&& 5. Date：日期类型，没有时间。映射到数据库中是`date`类型，在使用的时候，传递`datetime.date()`进去。
&& 6. DateTime：日期时间类型。映射到数据库中的是`datetime`类型，在使用的时候，传递`datetime.datetime()`进去。
&& 7. Float：浮点类型。

&& 1. `primary_key`：主键，True和False。
&& 2. `autoincrement`：是否自动增长，True和False。
&& 3. `unique`：是否唯一。
&& 4. `nullable`：是否可空，默认是True。
&& 5. `default`：默认值。
&& 6. `onupdate`：在更新的时候，一般用在时间上面。

&& 常用包引入
from sqlalchemy import and_, func, or_

import sqlalchemy
print dir(sqlalchemy)
In [2]: print dir(sqlalchemy)
['ARRAY', 'BIGINT', 'BINARY', 'BLANK_SCHEMA', 'BLOB', 'BOOLEAN', 'BigInteger', 'Binary', 'Boolean', 'CHAR', 'CLOB', 'CheckConstraint', 'Column', 'ColumnDefault', 'Constraint', 'DATE', 'DATETIME', 'DDL', 'DECIMAL', 'Date', 'DateTime', 'DefaultClause', 'Enum', 'FLOAT', 'FetchedValue', 'Float', 'ForeignKey', 'ForeignKeyConstraint', 'INT', 'INTEGER', 'Index', 'Integer', 'Interval', 'JSON', 'LargeBinary', 'MetaData', 'NCHAR', 'NUMERIC', 'NVARCHAR', 'Numeric', 'PassiveDefault', 'PickleType', 'PrimaryKeyConstraint', 'REAL', 'SMALLINT', 'Sequence', 'SmallInteger', 'String', 'TEXT', 'TIME', 'TIMESTAMP', 'Table', 'Text', 'ThreadLocalMetaData', 'Time', 'TypeDecorator', 'Unicode', 'UnicodeText', 'UniqueConstraint', 'VARBINARY', 'VARCHAR', '__all__', '__builtins__', '__doc__', '__file__', '__go', '__name__', '__package__', '__path__', '__version__', 'alias', 'all_', 'and_', 'any_', 'asc', 'between', 'bindparam', 'case', 'cast', 'collate', 'column', 'cprocessors', 'create_engine', 'cresultproxy', 'cutils', 'delete', 'desc', 'dialects', 'distinct', 'engine', 'engine_from_config', 'event', 'events', 'exc', 'except_', 'except_all', 'exists', 'extract', 'false', 'func', 'funcfilter', 'insert', 'inspect', 'inspection', 'interfaces', 'intersect', 'intersect_all', 'join', 'lateral', 'literal', 'literal_column', 'log', 'modifier', 'not_', 'null', 'nullsfirst', 'nullslast', 'or_', 'outerjoin', 'outparam', 'over', 'pool', 'processors', 'schema', 'select', 'sql', 'subquery', 'table', 'tablesample', 'text', 'true', 'tuple_', 'type_coerce', 'types', 'union', 'union_all', 'update', 'util', 'within_group']

&& 查询去重
session.query(Phone.brand).distinct().all()
&& 查询数据是否存在
session.query(User).filter(exists().where(Address.user_id == User.id))
session.query(User).filter(User.addresses.any())
&& 查询数据条数
sessino.query(User).all().count()

&& 查询模糊匹配
session.query(User).filter(User.code.like('%'+code.strip().upper()+'%')).all()

from sqlalchemy import func
&& 忽略大小写
session.query(User).filter(func.upper(User.code)=='AAA').all()
session.query(User).filter(func.lower(User.code)=='aaa').all()
&& 日期格式转字符
session.query(User).filter(func.to_char(User.code)=='2018-01-01').all()
&& 列表查询 + 倒序/顺序查询
from sqlalchemy import asc, desc
User.query.filter(User.id.in_((1, 2, 3))).order_by(desc(User.datetime)).all()
&&

session.query(User).all()
session.query(User.name, User.fullname).all()
session.query(User, User.name).all()

print(session.query(User).filter(and_(User.name.like("user%"), User.fullname.like("first%"))).all())
print(session.query(User).filter(or_(User.name.like("user%"), User.password != None)).all())

session.query(User).filter("id>:id").params(id=1).all()

session.query(User, Address).filter(User.id == Address.user_id).all()
session.query(User).join(User.addresses).all()
session.query(User).outerjoin(User.addresses).all()

session.query(User.name, func.count('*').label("user_count")).group_by(User.name).all()
session.query(User.name, func.sum(User.id).label("user_id_sum")).group_by(User.name).all()

stmt = session.query(Address.user_id, func.count('*').label("address_count")).group_by(Address.user_id).subquery()
print(session.query(User, stmt.c.address_count).outerjoin((stmt, User.id == stmt.c.user_id)).order_by(User.id).all())

限制返回字段查询
person = session.query(Person.name, Person.created_at,Person.updated_at).filter_by(name="zhongwei").order_by(Person.created_at).first()

In [7]: print(sqlalchemy.__version__)
1.2.10

In [3]: from sqlalchemy import create_engine
&& 连接到 Sqlite
In [4]: engine = create_engine('sqlite:///:memory:', echo=True)
&& echo 是否生成日志。True 生成所有Sql日志，False 不生成
&& 连接到 其他数据库
engine = create_engine('postgresql://scott:tiger@localhost:5432/mydatabase')

>>> from sqlalchemy.ext.declarative import declarative_base
>>> Base = declarative_base()

>>> from sqlalchemy.ext.declarative import declarative_base

>>> Base = declarative_base()

>>> from sqlalchemy import Column, Integer, String
>>> class User(Base):
...     __tablename__ = 'users'
...
...     id = Column(Integer, primary_key=True)
...     name = Column(String)
...     fullname = Column(String)
...     password = Column(String)
...
...     def __repr__(self):
...        return "<User(name='%s', fullname='%s', password='%s')>" % (
...                             self.name, self.fullname, self.password)

>>> 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('password', String(), table=<users>), schema=None)

>>> Base.metadata.create_all(engine)
SELECT ...
PRAGMA table_info("users")
()
CREATE TABLE users (
    id INTEGER NOT NULL, name VARCHAR,
    fullname VARCHAR,
    password VARCHAR,
    PRIMARY KEY (id)
)
()
COMMIT

from sqlalchemy import Sequence
Column(Integer, Sequence('user_id_seq'), primary_key=True)
Table因此，通过我们的声明性映射生成的完整，万无一失的因素是：

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
    name = Column(String(50))
    fullname = Column(String(50))
    password = Column(String(12))

    def __repr__(self):
        return "<User(name='%s', fullname='%s', password='%s')>" % (
                                self.name, self.fullname, self.password)

>>> ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')
>>> ed_user.name
'ed'
>>> ed_user.password
'edspassword'
>>> str(ed_user.id)
'None'

>>> from sqlalchemy.orm import sessionmaker
>>> Session = sessionmaker(bind=engine)
如果您的应用程序尚未 Engine定义模块级对象，请将其设置为：

>>> Session = sessionmaker()
稍后，当您使用创建引擎时create_engine()，将其连接到Session使用 configure()：

>>> Session.configure(bind=engine)  && once engine is available
会话生命周期模式

>>> ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')
>>> session.add(ed_user)

query.filter(User.name == 'ed')

query.filter(User.name != 'ed')

query.filter(User.name.like('%ed%'))
ColumnOperators.like()呈现LIKE运算符，对某些后端不区分大小写，对其他后端区分大小写。对于保证不区分大小写的比较，请使用 ColumnOperators.ilike()。

query.filter(User.name.ilike('%ed%'))
注意

大多数后端不直接支持ILIKE。对于那些，ColumnOperators.ilike()运算符呈现一个表达式，将LIKE与应用于每个操作数的LOWER SQL函数相结合。

query.filter(User.name.in_(['ed', 'wendy', 'jack']))

&&works with query objects too:
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)
&& alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))

query.filter(User.name != None)
&& alternatively, if pep8/linters are a concern
query.filter(User.name.isnot(None))

&& use and_()
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
&& or send multiple expressions to .filter()
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
&& or chain multiple filter()/filter_by() calls
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'))

>>> query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)
SQL>>> query.all()
[<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>,
      <User(name='fred', fullname='Fred Flinstone', password='blah')>]
first() 应用限制为1并将第一个结果作为标量返回：

SQL>>> query.first()
<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>
one()完全提取所有行，如果结果中不存在一个对象标识或复合行，则会引发错误。找到多行：

>>> user = query.one()
Traceback (most recent call last):
...
MultipleResultsFound: Multiple rows were found for one()
没有找到行：

>>> user = query.filter(User.id == 99).one()
Traceback (most recent call last):
...
NoResultFound: No row was found for one()
该one()方法非常适用于希望处理“找不到物品”而不是“找到多件物品”的系统; 例如RESTful Web服务，可能希望在找不到结果时引发“未找到404”，但在找到多个结果时引发应用程序错误。

one_or_none()就像one()，除非没有找到结果，它不会引起错误; 它只是回来了None。像 one()，但是，它如果有多个结果发现引发错误。

scalar()调用该one()方法，并在成功时返回该行的第一列：

>>> query = session.query(User.id).filter(User.name == 'ed').\
...    order_by(User.id)
SQL>>> query.scalar()
1

>>> from sqlalchemy import text
SQL>>> for user in session.query(User).\
...             filter(text("id<224")).\
...             order_by(text("id")).all():
...     print(user.name)
ed
wendy
mary
fred
可以使用冒号使用基于字符串的SQL指定绑定参数。要指定值，请使用以下params() 方法：

SQL>>> 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 Flinstone', password='blah')>
要使用完全基于字符串的语句，text()可以将表示完整语句的构造传递给 from_statement()。如果没有其他说明符，字符串SQL中的列将根据名称与模型列匹配，例如下面我们只使用星号表示加载所有列：

SQL>>> session.query(User).from_statement(
...                     text("SELECT * FROM users where name=:name")).\
...                     params(name='ed').all()
[<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>]
匹配名称上的列适用于简单的情况，但在处理包含重复列名的复杂语句或使用不易与特定名称匹配的匿名ORM构造时可能会变得难以处理。此外，在处理结果行时，我们可能会发现在映射列中存在键入行为。对于这些情况，text()构造允许我们在位置上将其文本SQL链接到Core或ORM映射的列表达式; 我们可以通过将列表达式作为位置参数传递给TextClause.columns()方法来实现这一点 ：

>>> stmt = text("SELECT name, id, fullname, password "
...             "FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id, User.fullname, User.password)
SQL>>> session.query(User).from_statement(stmt).params(name='ed').all()
[<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>]
版本1.1中的新增功能：该TextClause.columns()方法现在接受列表达式，这些列表达式将与纯文本SQL结果集进行位置匹配，从而无需在SQL语句中匹配甚至是唯一的列名。

从text()构造中进行选择时，Query 仍然可以指定要返回的列和实体; 而不是 query(User)我们也可以单独要求列，如在任何其他情况下：

>>> stmt = text("SELECT name, id FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id)
SQL>>> session.query(User.id, User.name).\
...          from_statement(stmt).params(name='ed').all()
[(1, u'ed')]
也可以看看

使用Textual SQL -text()从仅核心查询的角度解释构造。

SQL>>> session.query(User).filter(User.name.like('%ed')).count()
2
指望 count()

Query.count()曾经是一个非常复杂的方法，当它试图猜测现有查询周围是否需要子查询时，在某些奇特的情况下它不会做正确的事情。现在它每次都使用一个简单的子查询，它只有两行长并且总是返回正确的答案。使用func.count()如果一个特定的语句绝对不能容忍的子查询存在。

该count()方法用于确定SQL语句将返回多少行。查看上面生成的SQL，SQLAlchemy总是将我们查询的任何内容放入子查询中，然后从中计算行数。在某些情况下，这可以简化为更简单，但SQLAlchemy的现代版本不会尝试猜测何时合适，因为可以使用更明确的方法发出确切的SQL。SELECT count(*) FROM table

对于需要具体指出“要计数的东西”的情况，我们可以直接使用构造中func.count()可用 的表达式指定“计数”函数func。下面我们用它来返回每个不同用户名的计数：

>>> from sqlalchemy import func
SQL>>> 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')]
为了实现我们的简单，我们可以将其应用为：SELECT count(*) FROM table

SQL>>> session.query(func.count('*')).select_from(User).scalar()
4
select_from()如果我们User直接用主键表示计数，则可以删除用法：

SQL>>> session.query(func.count(User.id)).scalar()
4

>>> 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构造，它是一个应用于的指令，Column它指示此列中的值应该被约束为指定的远程列中存在的值。这是关系数据库的核心功能，并且是“粘合剂”，它将未连接的表集合转换为具有丰富的重叠关系。的ForeignKey是，在以上的值表示addresses.user_id列应被约束在这些值users.id列中，即它的主键。

第二个指令，称为relationship()ORM ，使用该属性告诉ORM Address类本身应该链接到User类Address.user。 relationship()使用两个表之间的外键关系来确定此链接的性质，确定Address.user将是多对一。另一个relationship()指令放在 User属性下的映射类中User.addresses。在两个 relationship()指令中，relationship.back_populates分配参数 以引用补充属性名称; 通过这样做，每个人都relationship() 可以做出与反向表达的相同关系的智能决策; 一方面，Address.user指的是一个User实例，另一方面User.addresses指的是一个列表 Address 实例。

注意

该relationship.back_populates参数是一个非常常见的SQLAlchemy功能的较新版本 relationship.backref。该relationship.backref 参数还没有到哪里去了，永远保持可用！这relationship.back_populates是一回事，除了更冗长，更容易操纵。有关整个主题的概述，请参阅使用Backref链接关系部分。

多对一关系的反面总是一对多。基本关系模式的完整可用relationship()配置目录。

两个互补关系Address.user并且User.addresses 被称为一个双向关系，并且是SQLAlchemy的ORM的一个关键特征。Linking Relationships with Backref部分详细 讨论了“backref”功能。

relationship()假设声明系统正在使用中，可以使用字符串指定远程类所关注的参数。一旦所有映射完成，这些字符串将被计算为Python表达式，以便生成实际参数，在上面的例子中是User类。在此评估期间允许使用的名称包括根据声明的基础创建的所有类的名称。

有关relationship()参数样式的更多详细信息，请参阅docstring 。

你知道吗 ？

大多数(尽管不是全部)关系数据库中的FOREIGN KEY约束只能链接到主键列或具有UNIQUE约束的列。
引用多列主键的FOREIGN KEY约束，本身有多列，称为“复合外键”。它还可以引用这些列的子集。
FOREIGN KEY列可以自动更新，以响应引用的列或行的更改。这称为CASCADE 引用操作，是关系数据库的内置函数。
FOREIGN KEY可以参考自己的表。这被称为“自引用”外键。
在外键- 维基百科上阅读更多关于外键的信息。
我们需要addresses在数据库中创建表，因此我们将从元数据中发出另一个CREATE，它将跳过已经创建的表：

SQL>>> Base.metadata.create_all(engine)

>>> jack = User(name='jack', fullname='Jack Bean', password='gjffdd')
>>> jack.addresses
[]
我们可以自由地在Address对象上添加User对象。在这种情况下，我们只是直接指定一个完整列表：

>>> jack.addresses = [
...                 Address(email_address='jack@google.com'),
...                 Address(email_address='j25@yahoo.com')]
使用双向关系时，在一个方向上添加的元素会自动在另一个方向上可见。此行为基于属性on-change事件发生，并在Python中进行评估，而不使用任何SQL：

>>> jack.addresses[1]
<Address(email_address='j25@yahoo.com')>

>>> jack.addresses[1].user
<User(name='jack', fullname='Jack Bean', password='gjffdd')>
让我们添加并提交到数据库。以及相应 集合中的两个成员都使用称为级联的过程一次添加到会话中：Jack BeanjackAddressaddresses

>>> session.add(jack)
SQL>>> session.commit()
查询杰克，我们得到杰克回来。尚未为Jack的地址发布SQL：

SQL>>> jack = session.query(User).\
... filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', password='gjffdd')>
我们来看看这个addresses系列吧。观看SQL：

SQL>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
当我们访问addresses集合时，突然发出了SQL。这是延迟加载关系的一个例子。该addresses集合现在已加载，其行为就像普通列表一样。我们将介绍一些优化加载这个集合的方法。

SQL>>> 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', password='gjffdd')>
<Address(email_address='jack@google.com')>
另一方面，实际的SQL JOIN语法最容易使用以下Query.join()方法实现：

SQL>>> session.query(User).join(Address).\
...         filter(Address.email_address=='jack@google.com').\
...         all()
[<User(name='jack', fullname='Jack Bean', password='gjffdd')>]
Query.join()知道如何加入User ，Address因为它们之间只有一个外键。如果没有外键或多个外键，Query.join() 则在使用以下表单之一时效果更好：

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')                          && same, using a string
正如您所期望的那样，使用以下outerjoin()函数将相同的想法用于“外部”连接 ：

query.outerjoin(User.addresses)   && LEFT OUTER JOIN
参考文档join()包含此方法接受的调用样式的详细信息和示例; join() 对于任何SQL-fluent应用程序而言，它是使用中心的重要方法。

Query如果有多个实体，可以选择什么？

当省略ON子句或ON子句是纯SQL表达式时，该Query.join()方法通常从实体列表中最左边的项加入。要控制JOIN列表中的第一个实体，请使用以下Query.select_from()方法：

query = session.query(User, Address).select_from(Address).join(User)
@@@@@ 使用别名
在跨多个表进行查询时，如果需要多次引用同一个表，则SQL通常要求使用其他名称对该表进行别名，以便可以将该表与该表的其他实例区分开来。Query最明确使用aliased构造的支持。下面我们Address 两次加入实体，找到同时拥有两个不同电子邮件地址的用户：

>>> from sqlalchemy.orm import aliased
>>> adalias1 = aliased(Address)
>>> adalias2 = aliased(Address)
SQL>>> 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=='jack@google.com').\
...     filter(adalias2.email_address=='j25@yahoo.com'):
...     print(username, email1, email2)
jack jack@google.com j25@yahoo.com

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
使用Query，我们从内到外构建一个这样的语句。的statement存取器返回一个表示由特定生成的声明SQL表达式 Query-这是一个实例select() 构建体，其中描述了SQL表达式语言教程：

>>> from sqlalchemy.sql import func
>>> stmt = session.query(Address.user_id, func.count('*').\
...         label('address_count')).\
...         group_by(Address.user_id).subquery()
所述func关键字生成SQL函数，以及subquery()关于方法 Query产生表示嵌入的别名内SELECT语句中的SQL表达构建体(它实际上是简写query.statement.alias())。

一旦我们得到了语句，它就像一个Table构造，比如我们在users本教程开始时创建的 构造 。语句中的列可通过以下属性访问c：

SQL>>> 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', password='f8s7ccs')> None
<User(name='wendy', fullname='Wendy Williams', password='foobar')> None
<User(name='mary', fullname='Mary Contrary', password='xxg527')> None
<User(name='fred', fullname='Fred Flinstone', password='blah')> None
<User(name='jack', fullname='Jack Bean', password='gjffdd')> 2
@@@@@ 从子查询中选择实体
上面，我们刚刚选择了一个包含子查询列的结果。如果我们希望子查询映射到实体怎么办？为此，我们使用aliased() 将映射类的“别名”与子查询相关联：

SQL>>> 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):
...     print(user)
...     print(address)
<User(name='jack', fullname='Jack Bean', password='gjffdd')>
<Address(email_address='jack@google.com')>

>>> from sqlalchemy.sql import exists
>>> stmt = exists().where(Address.user_id==User.id)
SQL>>> for name, in session.query(User.name).filter(stmt):
...     print(name)
jack
该Query功能几家运营商，这使得使用情况自动存在。上面，声明可以User.addresses使用any()以下关系表达 ：

SQL>>> for name, in session.query(User.name).\
...         filter(User.addresses.any()):
...     print(name)
jack
any() 采用标准，限制匹配的行：

SQL>>> for name, in session.query(User.name).\
...     filter(User.addresses.any(Address.email_address.like('%google%'))):
...     print(name)
jack
has()与any()多对一关系是同一个运算符 (请注意~这里的运算符，这意味着“NOT”)：

SQL>>> session.query(Address).\
...         filter(~Address.user.has(User.name=='jack')).all()
[]

query.filter(User.addresses.any(Address.email_address == 'bar'))

&& also takes keyword arguments:
query.filter(User.addresses.any(email_address='bar'))

query.filter(Address.user.has(name='ed'))
Query.with_parent() (用于任何关系)：

session.query(Address).with_parent(someuser, 'addresses')

>>> from sqlalchemy.orm import joinedload

SQL>>> jack = session.query(User).\
...                        options(joinedload(User.addresses)).\
...                        filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', password='gjffdd')>

>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
请注意，即使OUTER JOIN导致两行，我们仍然只有一个User返回实例。这是因为Query基于对象标识将“uniquing”策略应用于返回的实体。具体来说，可以应用联合的预先加载而不会影响查询结果。

虽然joinedload()已经存在了很长时间，但是subqueryload() 是一种新形式的渴望加载。 subqueryload()往往更适合加载相关集合，而joinedload()往往更适合多对一关系，因为只有一行加载了潜在客户和相关对象。

joinedload() 不是替代品 join()

>>> from sqlalchemy.orm import contains_eager
SQL>>> 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', password='gjffdd')>
有关预先加载的更多信息，包括默认情况下如何配置各种加载形式，请参阅关系加载技术一节。

>>> session.delete(jack)
SQL>>> session.query(User).filter_by(name='jack').count()
0
到现在为止还挺好。杰克的Address物品怎么样？

SQL>>> session.query(Address).filter(
...     Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
...  ).count()
2
哦，他们还在那里！分析刷新SQL，我们可以看到user_id每个地址的 列都设置为NULL，但是没有删除行。SQLAlchemy并不认为删除级联，你必须告诉它这样做。

>>> session.close()
ROLLBACK
并使用新的declarative_base()：

>>> Base = declarative_base()
接下来我们将声明User该类，添加addresses包括级联配置的关系(我们也将构造函数保留在外)：

>>> 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")
...
...     def __repr__(self):
...        return "<User(name='%s', fullname='%s', password='%s')>" % (
...                                self.name, self.fullname, self.password)
然后我们重新创建Address，注意到在这种情况下我们已经Address.user通过User类创建了关系：

>>> 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

&& load Jack by primary key
SQL>>> jack = session.query(User).get(5)

&& remove one Address (lazy load fires off)
SQL>>> del jack.addresses[1]

&& only one address remains
SQL>>> session.query(Address).filter(
...     Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
1

>>> session.delete(jack)

SQL>>> session.query(User).filter_by(name='jack').count()
0

SQL>>> session.query(Address).filter(
...    Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
0
更多关于瀑布

>>> from sqlalchemy import Table, Text
>>> && association table
>>> 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)
... )
在上面，我们可以看到声明a Table直接与声明映射类有点不同。 Table是构造函数，因此每个单独的Column参数用逗号分隔。该 Column对象也明确地赋予其名称，而不是从指定的属性名称中获取。

接下来，我们定义BlogPost并Keyword使用互补 relationship()结构，每个引用post_keywords 表作为关联表：

>>> 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 = relationship(User, back_populates="posts")
>>> User.posts = relationship(BlogPost, back_populates="author", lazy="dynamic")
创建新表：

SQL>>> Base.metadata.create_all(engine)
用法与我们一直在做的不同。让我们给Wendy一些博文：

SQL>>> wendy = session.query(User).\
...                 filter_by(name='wendy').\
...                 one()
>>> post = BlogPost("Wendy's Blog Post", "This is a test", wendy)
>>> session.add(post)
我们将关键字唯一地存储在数据库中，但我们知道我们还没有，所以我们可以创建它们：

>>> post.keywords.append(Keyword('wendy'))
>>> post.keywords.append(Keyword('firstpost'))
我们现在可以使用关键字“firstpost”查找所有博文。我们将使用 any运算符来查找“其中任何关键字都包含关键字字符串'firstpost'的博客帖子”：

SQL>>> 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', password='foobar')>)]
如果我们想要查找用户拥有的帖子wendy，我们可以告诉查询缩小到该User对象作为父对象：

SQL>>> 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', password='foobar')>)]
或者我们可以使用Wendy自己的posts关系，这是一种“动态”关系，直接从那里查询：

SQL>>> wendy.posts.\
...         filter(BlogPost.keywords.any(keyword='firstpost')).\
...         all()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', password='foobar')>)]

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey

Base = declarative_base()

class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)
上面是一个包含四列的基本单表映射。其他属性(例如与其他映射类的关系)也在类定义中内联声明：

class User(Base):
    __tablename__ = 'user'

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

    addresses = relationship("Address", backref="user", order_by="Address.id")

class Address(Base):
    __tablename__ = 'address'

    id = Column(Integer, primary_key=True)
    user_id = Column(ForeignKey('user.id'))
    email_address = Column(String)
声明映射系统在Object Relational Tutorial中引入 。有关此系统如何工作的其他详细信息，请参阅声明。

from sqlalchemy import Table, MetaData, Column, Integer, String, ForeignKey
from sqlalchemy.orm import mapper

metadata = MetaData()

user = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(50)),
            Column('fullname', String(50)),
            Column('password', String(12))
        )

class User(object):
    def __init__(self, name, fullname, password):
        self.name = name
        self.fullname = fullname
        self.password = password

mapper(User, user)
通过properties字典提供有关映射属性的信息，例如与其他类的关系。下面的示例说明了第二个Table 对象，映射到一个名为的类Address，然后链接到Uservia relationship()：

address = Table('address', metadata,
            Column('id', Integer, primary_key=True),
            Column('user_id', Integer, ForeignKey('user.id')),
            Column('email_address', String(50))
            )

mapper(User, user, properties={
    'addresses' : relationship(Address, backref='user', order_by=address.c.id)
})

mapper(Address, address)
使用经典映射时，必须直接提供类，而不必使用Declarative提供的“字符串查找”系统。SQL表达式通常根据Table对象来指定，即address.c.id上面是Address关系，而不是Address.id，因为Address可能尚未链接到表元数据，也不能在此处指定字符串。

文档中的一些示例仍然使用经典方法，但请注意，经典方法和声明方法是完全可互换的。两个系统最终都创建了相同的配置，由一个Table用户定义的类组成，并与一个链接在一起mapper()。当我们谈论“行为mapper()”时，这包括使用声明系统时 - 它仍然在幕后使用。

>>> from sqlalchemy import inspect
>>> insp = inspect(User)
提供详细信息，包括Mapper.columns：

>>> insp.columns
<sqlalchemy.util._collections.OrderedProperties object at 0x102f407f8>
这是一个名称空间，可以以列表格式或通过单个名称查看：

>>> list(insp.columns)
[Column('id', Integer(), table=<user>, primary_key=True, nullable=False), Column('name', String(length=50), table=<user>), Column('fullname', String(length=50), table=<user>), Column('password', String(length=12), table=<user>)]
>>> insp.columns.name
Column('name', String(length=50), table=<user>)
其他名称空间包括Mapper.all_orm_descriptors，包括所有映射属性以及混合，关联代理：

>>> insp.all_orm_descriptors
<sqlalchemy.util._collections.ImmutableProperties object at 0x1040e2c68>
>>> insp.all_orm_descriptors.keys()
['fullname', 'password', 'name', 'id']
以及Mapper.column_attrs：

>>> list(insp.column_attrs)
[<ColumnProperty at 0x10403fde0; id>, <ColumnProperty at 0x10403fce8; name>, <ColumnProperty at 0x1040e9050; fullname>, <ColumnProperty at 0x1040e9148; password>]
>>> insp.column_attrs.name
<ColumnProperty at 0x10403fce8; name>
>>> insp.column_attrs.name.expression
Column('name', String(length=50), table=<user>)

from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
要在一对多中建立双向关系，其中“反向”是多对一，请指定一个附加relationship()并使用以下relationship.back_populates参数连接两者：

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", back_populates="parent")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    parent = relationship("Parent", back_populates="children")
Child将获得parent具有多对一语义的属性。

或者，该backref选项可用于单个relationship()而不是使用 back_populates：

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", backref="parent")

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    child_id = Column(Integer, ForeignKey('child.id'))
    child = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
通过添加秒relationship() 并relationship.back_populates在两个方向上应用参数来实现双向行为：

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    child_id = Column(Integer, ForeignKey('child.id'))
    child = relationship("Child", back_populates="parents")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parents = relationship("Parent", back_populates="child")
或者，backref参数可以应用于单个relationship()，例如Parent.child：

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    child_id = Column(Integer, ForeignKey('child.id'))
    child = relationship("Child", backref="parents")

&& row will be deleted from the "secondary" table
&& automatically
myparent.children.remove(somechild)
经常出现的一个问题是当子对象直接递送到“辅助”表中的行时如何删除Session.delete()：

session.delete(somechild)
这里有几种可能性：

如果存在relationship()from Parentto Child，但 没有将特定链接Child到每个的反向关系Parent，则SQLAlchemy将不会意识到在删除此特定 Child对象时，它需要维护将其链接到的“辅助”表Parent。不会删除“辅助”表。
如果存在将特定链接Child到每个特定的关系Parent，假设它被调用Child.parents，默认情况下SQLAlchemy将加载到Child.parents集合中以查找所有Parent对象，并从建立此链接的“辅助”表中删除每一行。请注意，此关系不需要是bidrectional; SQLAlchemy严格查看relationship()与Child被删除对象相关的每个内容。
这里性能更高的选项是使用ON DELETE CASCADE指令和数据库使用的外键。假设数据库支持此功能，则可以使数据库本身自动删除“辅助”表中的行，因为删除了“child”中的引用行。Child.parents 在这种情况下，可以指示SQLAlchemy 使用passive_deletes 指令on 来放弃在集合中的主动加载relationship(); 有关详细信息，请参阅使用被动删除。
请再次注意，这些行为仅与使用的secondary选项相关relationship()。如果处理显式映射且不存在于secondary相关选项中的关联表，则relationship()可以使用级联规则来自动删除实体以响应被删除的相关实体 - 有关此功能的信息，请参阅级联。

&& create parent, append a child via association
p = Parent()
a = Association(extra_data="some data")
a.child = Child()
p.children.append(a)

&& iterate through child objects via association, including association
&& attributes
for assoc in p.children:
    print(assoc.extra_data)
    print(assoc.child)
为了增强关联对象模式以便直接访问Association对象是可选的，SQLAlchemy提供了关联代理扩展。此扩展允许配置属性，这些属性将通过单个访问访问两个“跳”，一个“跳”到关联对象，第二个跳转到目标属性。

&& -*- coding:utf-8 -*-
from nose.tools import eq_, raises, assert_true
from sqlalchemy.orm import aliased
from ..model.report import *

def re_create_table(ORM):
    try:
        ORM.__table__.drop(current_engine)
    except Exception ,e:
        print u"表不存在"
        &&print Exception ,e
    ORM.__table__.create(current_engine)

def re_create_tables(ORMS):
    for orm in ORMS:
        try:
            orm.__table__.drop(current_engine)
        except Exception ,e:
            print u"表不存在"
            &&print Exception ,e
    ORMS.reverse()
    for orm in ORMS:
        orm.__table__.create(current_engine)

if __name__ == '__main__':
    re_create_tables([Interest_Summary_test,FtpAcct_test,FTPAdj_test])
    re_create_table(SearchDataset)

res = session.query(User).first()
res.__dict__

Base = sqlalchemy.ext.declarative.declarative_base()

class Post(Base):
    __tablename__ = 'post'
    id = Column(Integer, primary_key=True)
    title = Column(String)

&& 在基类中添加to_dict方法
def to_dict(self):
    return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}

Base.to_dict = to_dict

p = session.query(Post).first()
p.to_dict()

&& to_dict代码样例
def to_dict(self):
    attrlist = [a for a in self.__dict__.keys() if not a.startswith('_')]
    data = {}
    for name in attrlist:
        d = getattr(self, name, None)
        if isinstance(d, datetime.datetime):
            d = d.strftime('%Y-%m-%d %H:%M:%S')
        elif isinstance(d, datetime.date):
            d = d.strftime('%Y-%m-%d')
        elif isinstance(d, datetime.time):
            d =  d.strftime('%H:%M:%S')
        data[name] = d
    return data

setattr(Base, 'to_dict', to_dict)

from exts import db

class User(db.Model):
    __tablename__ = 'user'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(50), nullable=False)
    age = db.Column(db.Integer, nullable=False)

    # 把SQLAlchemy查询对象转换成字典
    def to_dict(self):
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}


# 将查出来的所有对象都转换成json的函数
def to_json(all_vendors):
    v = [ven.to_dict() for ven in all_vendors]
    return v