SQLAlchemy 学习
编程    Python ORM    2015-08-15 03:47:22    1649    0    0
  Python ORM

Python 的数据库 ORM (Object Relational Mapper)有很多, django
下自带一个,但是模块联系比较紧密,无法拆分. PonyORM 用法简单灵活,是个人项目的好选择. SQLAlchemy 也是被广泛使用的权威框架. 目前的版本的 0.9 , 官网 中文资料较少,英文资料蛮清晰的,也有 PDF.

 安装

直接敲命令安装.

pip install SQLAlchemy

但是在 windows 下会出出现 C 扩展无法编译的问题,但是不影响使用,除非你需要在 windows 下部署应用.

在 iPython 下查看版本:

In [1]: import sqlalchemy
In [2]: sqlalchemy.__version__
Out[2]: '0.9.8'

连接数据库

ORM 是数据库的抽象层, 具体使用哪种哪个数据数据需要配置.首先导入数据库引擎: from sqlalchemy import create_engine

如果简单的尝试一些功能, sqlite 就很合适. Python 自带了 sqlite 的驱动.
engine = create_engine('sqlite:///test.sqlite', echo=True)

更简单一点就放在内存中.
engine = create_engine('sqlite:///:memory:', echo=True)

常用的 MySql 连接就是:
engine = create_engine("mysql://user:password@localhost/dbname",echo=True)

user 是数据库用户名, password 是密码 , localhost 是数据库所在位置, dbname 是数据库名字. 这些都需要事先安装配置好.

定义数据模型

所有模型都必须从一个 Base 类继承

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

一个简单的示例类

from sqlalchemy import Column, Integer, String , DateTime, ForeignKey
```
class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String(50), index=True, unique=True)
     fullname = Column(String(50), index=True, unique=True)
     joindate = Column(DateTime, default=datetime.now())
     password = Column(String(50))

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

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

生成表

Base.metadata.create_all(engine)

由于链接数据库指定了echo=True 所以会把生成表所用的 SQL 语句打出来.

CREATE TABLE users (
    id INTEGER NOT NULL AUTO_INCREMENT, 
    name VARCHAR(50), 
    fullname VARCHAR(50), 
    joindate DATETIME, 
    password VARCHAR(50), 
    PRIMARY KEY (id)
)

已存在的表 create_all 是没有效果的.如果要修改表结构,需要额外的插件: sqlalchemy-migrate.

表结构

 数据操作

操作数据库前需要先创建或获取 session

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

插入

插入一个用户

user = User('ed', 'Ed Jones', 'edspassword')
session.add(user)
session.commit()

插入多个用户

session.add_all([
    User('wendy', 'Wendy Williams', 'foobar'),
    User('mary', 'Mary Contrary', 'xxg527'),
    User('fred', 'Fred Flinstone', 'blah')
])
session.commit()

插入只在调用 session.commit() 后才真正写入数据库.

表结构

查询

查询指定用户名修改提交

user = session.query(User).filter_by(name='mary').first()
user.password = '123'
session.commit()

批量查询

for instance in session.query(User)\
    .order_by(User.id):
    print instance.id,instance.name

print session.query(User).count()

可以看到 sqlalchemy 与 python 语法结合的很好.

查询的过滤条件

  • 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’))
    

    MATCH 对有些数据库不支持,例如 SQLite .

查询的结果

  • all()
    返回一个查询结果的列表

  • first()
    返回第一条结果

  • one()
    返回一条结果, 如果查询结果不唯一,则抛出 MultipleResultsFound 异常. 如果没有结果 则抛出 NoResultFound 异常. 异常定义都在 sqlalchemy.orm.exc 中.

  • scalar
    获取最相近的一条结果

  • count
    返回结果的个数

 兼容原生 SQL

from sqlalchemy import text

用原始 SQL 部分替换查询命令

for user in session.query(User).\
    filter(text("id<224")).\
    order_by(text("id")).all():
    print user.name
session.query(User).filter(text("id<:value and name=:name")).\
    params(value=224, name=’fred’).order_by(User.id).one()
```
session.query(User).from_statement(
    text("SELECT
*
FROM users where name=:name")).\
    params(name=’ed’).all()

建立关系

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref

新的表, 关联到之前的表.

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String(50), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    user = relationship("User", backref=backref('addresses', order_by=id))

    def __repr__(self):
        return "<Address(email_address='%s')>" % self.email_address

插入数据

jack = User(name='jack', fullname='Jack Bean', password='gjffdd')
jack.addresses = [
    Address(email_address='jack@google.com'),
    Address(email_address='j25@yahoo.com')
]
session.add(jack)
session.commit()

并表查询

jack = session.query(User).join(Address).\
    filter(Address.email_address=='jack@google.com').\
    one()

删除

session.delete(jack)

但是 Address 依然保留了,需要自己在删除的时候一起删除. 如果需要实现自动删除关联.需要将关系声明在 User 类中,并添加 cascade 说明. Address 类中的关系声明要去掉. 修改后的模型类如下.

class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String(50), index=True, unique=True)
     fullname = Column(String(50), index=True, unique=True)
     joindate = Column(DateTime, default=datetime.now())
     password = Column(String(50))

     addresses = relationship("Address", backref='user',cascade="all, delete, delete-orphan")

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

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


class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String(50), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    #user = relationship("User", backref=backref('addresses', order_by=id))

    def __repr__(self):
        return "<Address(email_address='%s')>" % self.email_address

因为修改了数据表,所以需要删除所有的表重建数据库.
Base.metadata.drop_all(engine),或者可以使用 sqlalchemy-migrate 插件,进行升级.

 多对多关系

需要额外的建一个 Table

from sqlalchemy import Table
post_keywords = Table(’post_keywords’, Base.metadata,
    Column(’post_id’, Integer, ForeignKey(’posts.id’)),
    Column(’keyword_id’, Integer, ForeignKey(’keywords.id’))
    )

post 和 keyword 之间是多对多的关系

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, backref=’posts’)

    def __init__(self, headline, body):
    self.headline = headline
    self.body = body

    def __repr__(self):
    return "BlogPost(%r, %r)" % (self.headline, self.body)



class Keyword(Base):
    __tablename__ = ’keywords’

    id = Column(Integer, primary_key=True)
    keyword = Column(String(50), nullable=False, unique=True)

    def __init__(self, keyword):
    self.keyword = keyword
```
post = BlogPost("Wendy’s Blog Post", "This is a test")
session.add(post)
post.keywords.append(Keyword(’wendy’))
post.keywords.append(Keyword(’firstpost’))
session.commit()
```
session.query(BlogPost).\
    filter(BlogPost.keywords.any(keyword=’firstpost’)).\
    all()
session.commit()
文档导航