[关闭]
@zwenqiang 2015-06-06T01:16:50.000000Z 字数 3728 阅读 12307

SQLAlchemy(二) --- ORM

数据库


SQLAlchemy ORM提供了一个连接数据库表和用户自定义Python类的方法。
在此输入正文

定义一个映射(Mapping)

类的映射使用已经在基类中定义的声明式系统,这个基类(base class)维护了一个和这个基类相关的类和表的目录, 这就是所谓的declarative base class。我们的应用在一般导入的模块中将只有一个这个基类的实例。我们使用declarative_base()创建这个基类的实例。

  1. from sqlalchemy.ext.declarative import declarative_base
  2. Base = declarative_base()

有了这个基类我们可以定义任何数量的映射类。

  1. from sqlalchemy import Column, Integer, String
  2. class User(Base):
  3. __tablename__ = 'users'
  4. id = Column(Integer, primary_key=True)
  5. name = Column(String(32), nullable=False)
  6. gender = Column(String(1), nullable=False, server_default='M')
  7. def __repr__(self):
  8. return "<User(name='%s', gender='%s')" % (self.name, self.gender)
  9. # 执行创建表的语句
  10. Base.metadata.create_all(engine)

一个继承自Base的类至少有一个__tablename__属性,并且至少有一个含有主键(primary key)的Column。当一个类创建时, Declarative 将所有的Column类用特殊的Python属性访问器替代。

创建Session

ORM操作数据库的句柄就是Session.当我们启动我们的应用时,在create_engine()的同时,我们定义了一个Session类将作为一个创建Session实例的工厂。

  1. from sqlalchemy.orm import sessionmaker
  2. Session = sessionmaker(bind=engine)

如果我们还没有Engine的实例,我们可以先创建Session,当创建Engine后再绑定。

  1. Session = sessionmaker()
  2. Session.configure(bind=engine)

当需要和数据库有一个会话时,可以初始化一个Session

  1. session = Session()

添加新的类

为了持久化User对象, 我们使用add()将它加入到Session中。

  1. user = User(name="zhangsan", gender="M")
  2. session.add(user)

此时,这个user实例是待定的(pending);不执行任何SQL也不代表数据库中的一行数据。当使用一个flush过程时,Session将执行SQL来持久化user

  1. our_user = session.query(User).filter_by(name='zhangsan').first()
  2. # 此时our_user 和 user是同一个对象

事实上,Session返回同一行(对象)就是我们刚刚在Session内部的类的字典持久化的对象,所以我们事实上获得的就是我们加入Session中的实例。

我们需要告诉Session我们想要将所有的改变存入到数据库,提交事务。我们通过commit()来执行。

  1. session.commit()

在执行commit之前,执行query后,即使user已经有了标识符id, 但是数据库中并没有提交数据,只有在commit之后才会提交数据。

我们可以在数据提交commit之前调用事务的rollback来回滚之前的修改。

  1. session.rollback()

查询Query

Query查询返回的是元组tuples,是KeyedTuple提供的类,并且更像一个原生的Python对象。

  1. for row in session.query(User, User.name).all():
  2. print row.User, row.name

filter()方法通常接收的是Python操作符,而filter_by()方法使用关键字参数

  1. session.query(User).filter(User.name == "ed")
  2. session.query(User).filter_by(name = "ed")

通用的查询语句
equals:

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

not equals:

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

LIKE:

  1. query.filter(User.name.like('%ed%'))

IN:

  1. query.filter(User.name.in_(['ed', 'wendy', 'jack']))
  2. # works with query objects too:
  3. query.filter(User.name.in_(
  4. session.query(User.name).filter(User.name.like('%ed%'))
  5. ))

NOT IN:

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

IS NULL:

  1. query.filter(User.name == None)
  2. # alternatively, if pep8/linters are a concern
  3. query.filter(User.name.is_(None))

IS NOT NULL:

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

AND:

  1. # use and_()
  2. from sqlalchemy import and_
  3. query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
  4. # or send multiple expressions to .filter()
  5. query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
  6. # or chain multiple filter()/filter_by() calls
  7. query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')

OR:

  1. from sqlalchemy import or_
  2. query.filter(or_(User.name == 'ed', User.name == 'wendy'))
  3. MATCH:
  4. query.filter(User.name.match('wendy'))

返回列表List和标量Scalar

all()返回一个列表

  1. query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)
  2. query.all()

first()使用限制,返回结果集的第一条作为Scalar

  1. query.first()

one() 如果返回结果集不止一个对象,将raise一个错误。

  1. from sqlalchemy.orm.exc import MultipleResultsFound
  2. try:
  3. users = query.one()
  4. except MultiResultsFound e:
  5. print e

没有数据返回时

  1. from sqlalchemy.orm.exc import NoResultFound
  2. try:
  3. users = query.one()
  4. except NoResultFound, e:
  5. print e

scalar()调用one()方法,并且成功时返回第一列数据

  1. query.scalar()

使用文本SQL语句

文本字符串可以在查询中灵活使用,通过text()构建。

  1. from sqlalchemy import text
  2. for user in session.query(User).filter(text("id>20")).order_by(text("id")).all()
  3. print user.name

可以绑定参数,使用params()方法

  1. session.query(User).filter(text("id<:value and name=:name")).params(value=220, name="ed").order_by(User.id).one()

计数方法count()

  1. session.query(User).filter(Usre.name.like("%ed")).count()

也可以使用func.count()来计数

  1. from sqlalchemy import func
  2. session.query(func.count(User.name), User.name).group_by(User.name).all()
添加新批注
在作者公开此批注前,只有你和作者可见。
回复批注