SQLAlchemy ORM提供了一个连接数据库表和用户自定义Python类的方法。
在此输入正文
定义一个映射(Mapping)
类的映射使用已经在基类中定义的声明式系统,这个基类(base class)维护了一个和这个基类相关的类和表的目录, 这就是所谓的declarative base class。我们的应用在一般导入的模块中将只有一个这个基类的实例。我们使用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(32), nullable=False)
gender = Column(String(1), nullable=False, server_default='M')
def __repr__(self):
return "<User(name='%s', gender='%s')" % (self.name, self.gender)
# 执行创建表的语句
Base.metadata.create_all(engine)一个继承自Base的类至少有一个__tablename__属性，并且至少有一个含有主键(primary key)的Column。当一个类创建时， Declarative 将所有的Column类用特殊的Python属性访问器替代。
创建Session
ORM操作数据库的句柄就是Session.当我们启动我们的应用时，在create_engine()的同时，我们定义了一个Session类将作为一个创建Session实例的工厂。
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)如果我们还没有Engine的实例，我们可以先创建Session，当创建Engine后再绑定。
Session = sessionmaker()
Session.configure(bind=engine)当需要和数据库有一个会话时，可以初始化一个Session
session = Session()添加新的类
为了持久化User对象， 我们使用add()将它加入到Session中。
user = User(name="zhangsan", gender="M")
session.add(user)此时，这个user实例是待定的(pending);不执行任何SQL也不代表数据库中的一行数据。当使用一个flush过程时，Session将执行SQL来持久化user。
our_user = session.query(User).filter_by(name='zhangsan').first()
# 此时our_user 和 user是同一个对象事实上，Session返回同一行(对象)就是我们刚刚在Session内部的类的字典持久化的对象，所以我们事实上获得的就是我们加入Session中的实例。
我们需要告诉Session我们想要将所有的改变存入到数据库，提交事务。我们通过commit()来执行。
session.commit()在执行commit之前，执行query后，即使user已经有了标识符id, 但是数据库中并没有提交数据，只有在commit之后才会提交数据。
我们可以在数据提交commit之前调用事务的rollback来回滚之前的修改。
session.rollback()查询Query
Query查询返回的是元组tuples,是KeyedTuple提供的类，并且更像一个原生的Python对象。
for row in session.query(User, User.name).all():
print(row.User, row.name)filter()方法通常接收的是Python操作符，而filter_by()方法使用关键字参数
session.query(User).filter(User.name == "ed")
session.query(User).filter_by(name = "ed")通用的查询语句
equals:
query.filter(User.name == 'ed')not equals:
query.filter(User.name != 'ed')LIKE:
query.filter(User.name.like('%ed%'))IN:
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%'))
))NOT IN:
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))IS NULL:
query.filter(User.name == None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))IS NOT NULL:
query.filter(User.name != None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.isnot(None))AND:
# 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')OR:
from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))
MATCH:
query.filter(User.name.match('wendy'))返回列表List和标量Scalar
all()返回一个列表
query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)
query.all()first()使用限制，返回结果集的第一条作为Scalar
query.first()one() 如果返回结果集不止一个对象，将raise一个错误。
from sqlalchemy.orm.exc import MultipleResultsFound
try:
users = query.one()
except MultiResultsFound e:
print(e)没有数据返回时
from sqlalchemy.orm.exc import NoResultFound
try:
users = query.one()
except NoResultFound, e:
print(e)scalar()调用one()方法，并且成功时返回第一列数据
query.scalar()使用文本SQL语句
文本字符串可以在查询中灵活使用，通过text()构建。
from sqlalchemy import text
for user in session.query(User).filter(text("id>20")).order_by(text("id")).all()
print(user.name)可以绑定参数，使用params()方法
session.query(User).filter(text("id<:value and name=:name")).params(value=220, name="ed").order_by(User.id).one()计数方法count()
session.query(User).filter(Usre.name.like("%ed")).count()也可以使用func.count()来计数
from sqlalchemy import func
session.query(func.count(User.name), User.name).group_by(User.name).all()
