Follower me on GitHub

前言:最近开始学习SQLAlchemy,本教程是其官方文档以及在读英文版<Essential SQLAlchemy>的翻译加一些自己的理解和总结

1 什么是 SQLAlchemy?

它是给mysql, oracle,sqlite等关系型数据库的python接口,不需要大幅修改原有的python代码,它已经包含了SQL表达式语言和ORM,看一些例子:

sql=”INSERT INTO user(user_name, password) VALUES (%s, %s)” cursor = conn.cursor() cursor.execute(sql, (‘dongwm’, ‘testpass’))

以上是一个常用的mysql的SQL语句,但是冗长也容易出错,并且可能导致安全问题(因为是字符串的语句,会存在SQL注入),并且代码不跨平台,在不同数据库软件的语句不同(以下是一个 Oracle例子),不具备客移植性:

sql=”INSERT INTO user(user_name, password) VALUES (:1, :2)” cursor = conn.cursor() cursor.execute(sql, ‘dongwm’, ‘testpass’)

而在SQLAlchemy里只需要这样写:

statement = user_table.insert(user_name=’rick’, password=’parrot’) statement.execute()  #护略是什么数据库环境

SQLAlchemy还能让你写出很pythonic的语句:

statement = user_table.select(and_( user_table.c.created >= date(2007,1,1), user_table.c.created < date(2008,1,1)) result = statement.execute()  #检索所有在2007年创建的用户

metadata=MetaData(‘sqlite://’) # 告诉它你设置的数据库类型是基于内存的sqlite user_table = Table(  #创建一个表 ‘tf_user’, metadata, Column(‘id’, Integer, primary_key=True),  #一些字段,假设你懂SQL,那么以下的字段很好理解 Column(‘user_name’, Unicode(16), unique=True, nullable=False), Column(‘email_address’, Unicode(255), unique=True, nullable=False), Column(‘password’, Unicode(40), nullable=False), Column(‘first_name’, Unicode(255), default=”), Column(‘last_name’, Unicode(255), default=”), Column(‘created’, DateTime, default=datetime.now))

users_table = Table(‘users’, metadata, autoload=True) #假设table已经存在.就不需要指定字段,只是加个autoload=True

class User(object): pass  #虽然SQLAlchemy强大,但是插入更新还是需要手动指定,可以使用ORM,方法就是:设定一个类,定义一个表,把表映射到类里面 mapper(User, user_table)

下面是一个完整ORM的例子:

from sqlalchemy.orm import mapper, sessionmaker #sessionmaker() 函数是最常使用的创建最顶层可用于整个应用 Session 的方法,Session 管理着所有与数据库之间的会话
from datetime import datetime
from sqlalchemy import Table, MetaData, Column, ForeignKey, Integer, String, Unicode, DateTime #会SQL的人能理解这些函数吧?
engine = create_engine("sqlite:///tutorial.db", echo=True) #创建到数据库的连接,echo=True 表示用logging输出调试结果
metadata = MetaData() #跟踪表属性
user_table = Table( #创建一个表所需的信息:字段,表名等
'tf_user', metadata,
Column('id', Integer, primary_key=True),
Column('user_name', Unicode(16), unique=True, nullable=False),
Column('email_address', Unicode(255), unique=True, nullable=False),
Column('password', Unicode(40), nullable=False),
Column('first_name', Unicode(255), default=''),
Column('last_name', Unicode(255), default=''),
Column('created', DateTime, default=datetime.now))
metadata.create_all(engine)  #在数据库中生成表
class User(object): pass #创建一个映射类
mapper(User, user_table) #把表映射到类
Session = sessionmaker() #创建了一个自定义了的 Session类
Session.configure(bind=engine)  #将创建的数据库连接关联到这个session
session = Session()
u = User()
u.user_name='dongwm'
u.email_address='dongwm@dongwm.com'
u.password='testpass'  #给映射类添加以下必要的属性,因为上面创建表指定这几个字段不能为空
session.add(u)  #在session中添加内容

session.flush() #保存数据 session.commit() #数据库事务的提交,sisson自动过期而不需要关闭

query = session.query(User) #query() 简单的理解就是select() 的支持 ORM 的替代方法,可以接受任意组合的 class/column 表达式 print list(query) #列出所有user print query.get(1) #根据主键显示 print query.filter_by(user_name='dongwm').first() #类似于SQL的where,打印其中的第一个 u = query.filter_by(user_name='dongwm').first() u.password = 'newpass' #修改其密码字段 session.commit() #提交事务 print query.get(1).password #打印会出现新密码

for instance in session.query(User).order_by(User.id): #根据id字段排序,打印其中的用户名和邮箱地址 print instance.user_name, instance.email_address

既然是ORM框架,我们来一个更复杂的包含关系的例子,先看sql语句:

CREATE TABLE tf_user ( id INTEGER NOT NULL, user_name VARCHAR(16) NOT NULL, email_address VARCHAR(255) NOT NULL, password VARCHAR(40) NOT NULL, first_name VARCHAR(255), last_name VARCHAR(255), created TIMESTAMP, PRIMARY KEY (id), UNIQUE (user_name), UNIQUE (email_address)); CREATE TABLE tf_group ( id INTEGER NOT NULL, group_name VARCHAR(16) NOT NULL, PRIMARY KEY (id), UNIQUE (group_name)); CREATE TABLE tf_permission ( id INTEGER NOT NULL, permission_name VARCHAR(16) NOT NULL, PRIMARY KEY (id), UNIQUE (permission_name)); CREATE TABLE user_group ( user_id INTEGER, group_id INTEGER, PRIMARY KEY(user_id, group_id), FOREIGN KEY(user_id) REFERENCES tf_user (id), #user_group的user_id关联了tf_user的id字段 FOREIGN KEY(group_id) REFERENCES tf_group (id));  #group_id关联了 tf_group 的id字段

CREATE TABLE group_permission ( group_id INTEGER, permission_id INTEGER, PRIMARY KEY(group_id, permission_id), FOREIGN KEY(group_id) REFERENCES tf_group (id),  #group_permission的id关联 tf_group的id字段 FOREIGN KEY(permission_id) REFERENCES tf_permission (id)); #permission_id关联了tf_permission 的id字段

这是一个复杂的多对多的关系,比如检查用户是否有admin权限,sql需要这样:

SELECT COUNT(*) FROM tf_user, tf_group, tf_permission WHERE tf_user.user_name=’dongwm’ AND tf_user.id=user_group.user_id AND user_group.group_id = group_permission.group_id AND group_permission.permission_id = tf_permission.id AND permission_name=’admin’;  看起来太复杂并且繁长了

在面向对象的世界里,是这样的:

class User(object): groups=[] class Group(object): users=[] permissions=[] class Permission(object): groups=[]

print 'Summary for %s' % user.user_name
for g in user.groups:
    print ' Member of group %s' % g.group_name
    for p in g.permissions:
        print '... which has permission %s' % p.permission_name

 

def user_has_permission(user, permission_name):  #检查用户是否有permission_name的权限的函数
    for g in user.groups:
        for p in g.permissions: #可以看出来使用了for循环
            if p.permission_name == 'admin':
                return True
    return False

而在SQLAlchemy中,这样做:

mapper(User, user_table, properties=dict( groups=relation(Group, secondary=user_group, backref=’users’))) #properties是一个字典值。增加了一个groups 值,它又是一个 relation 对象,这个对象实现

#了Group类与user_group的  映射。这样我通过user_table的groups 属性就可以反映出RssFeed的值来,

#中间表对象(user_group)传给secondary参数,backref为自己的表(users) mapper(Group, group_table, properties=dict( permissions=relation(Permission, secondary=group_permission, backref=’groups’))) mapper(Permission, permission_table)

 

q = session.query(Permission) dongwm_is_admin = q.count_by(permission_name=’admin’,user_name=’dongwm’)

假如计算组里用户数(不包含忘记删除但是重复的)

for p in permissions: users = set() for g in p.groups: for u in g.users: users.add(u) print ‘Permission %s has %d users’ % (p.permission_name, len(users))

在SQLAlchemy可以这样:

q=select([Permission.c.permission_name, func.count(user_group.c.user_id)], and_(Permission.c.id==group_permission.c.permission_id, Group.c.id==group_permission.c.group_id, Group.c.id==user_group.c.group_id), group_by=[Permission.c.permission_name], distinct=True) rs=q.execute() for permission_name, num_users in q.execute(): print ‘Permission %s has %d users’ % (permission_name, num_users) #虽然也长,但是减少了数据库查询次数,也就是让简单事情简单化,复杂事情可能简单解决

看一个综合的例子:

class User(object):  #这些类设计数据库的模型

def __init__(self, group_name=None, users=None, permissions=None): if users is None: users = [] if permissions is None: permissions = [] self.group_name = group_name self._users = users self._permissions = permissions

def add_user(self, user): self._users.append(user)

def del_user(self, user): self._users.remove(user)

def add_permission(self, permission): self._permissions.append(permission)

def del_permission(self, permission): self._permissions.remove(permission)

class Permission(object):

def __init__(self, permission_name=None, groups=None): self.permission_name = permission_name self._groups = groups

def join_group(self, group): self._groups.append(group)

def leave_group(self, group): self._groups.remove(group)

用sqlalchemy的效果是这样的:

user_table = Table( ‘tf_user’, metadata, Column(‘id’, Integer, primary_key=True), Column(‘user_name’, Unicode(16), unique=True, nullable=False), Column(‘password’, Unicode(40), nullable=False))

group_table = Table( ‘tf_group’, metadata, Column(‘id’, Integer, primary_key=True), Column(‘group_name’, Unicode(16), unique=True, nullable=False))

permission_table = Table( ‘tf_permission’, metadata, Column(‘id’, Integer, primary_key=True), Column(‘permission_name’, Unicode(16), unique=True, nullable=False))

user_group = Table( ‘user_group’,  metadata, Column(‘user_id’, None, ForeignKey(‘tf_user.id’), primary_key=True), Column(‘group_id’, None, ForeignKey(‘tf_group.id’), primary_key=True))

group_permission = Table( ‘group_permission’,  metadata, Column(‘group_id’, None, ForeignKey(‘tf_group.id’), primary_key=True), Column(‘permission_id’, None, ForeignKey(‘tf_permission.id’), primary_key=True))

mapper(User, user_table, properties=dict( _groups=relation(Group, secondary=user_group, backref=’_users’))) mapper(Group, group_table, properties=dict( _permissions=relation(Permission, secondary=group_permission, backref=_’groups’))) mapper(Permission, permission_table)

这里没有修改对象,而join_group,leave_group这样的函数依然可用,sqlalchemy会跟踪变化,并且自动刷新数据库

上面介绍了一个完整的例子,连接数据库嗨可以这样:

engine = create_engine(‘sqlite://’) connection = engine.connect()  #使用connect result = connection.execute(“select user_name from tf_user”) for row in result: print ‘user name: %s’ % row[‘user_name’] result.close()

engine = create_engine(‘sqlite://’, strategy=’threadlocal’)  #,strategy=’threadlocal’表示重用其它本地线程减少对数据库的访问

from sqlalchemy.databases.mysql import MSEnum, MSBigInteger  #这个 sqlalchemy.databases是某数据库软件的’方言’集合,只支持特定平台 user_table = Table(‘tf_user’, meta, Column(‘id’, MSBigInteger), Column(‘honorific’, MSEnum(‘Mr’, ‘Mrs’, ‘Ms’, ‘Miss’, ‘Dr’, ‘Prof’)))

以下是几个MetaData的应用: unbound_meta = MetaData()  #这个metadata没有绑定 db1 = create_engine(‘sqlite://’) unbound_meta.bind = db1  #关联引擎

db2 = MetaData(‘sqlite:///test1.db’)  #直接设置引擎 bound_meta1 = MetaData(db2)

# Create a bound MetaData with an implicitly created engine bound_meta2 = MetaData(‘sqlite:///test2.db’)  #隐式绑定引擎 meta = MetaData(‘sqlite://’) #直接绑定引擎可以让源数据直接访问数据库

user_table = Table( ‘tf_user’, meta, Column(‘id’, Integer, primary_key=True), Column(‘user_name’, Unicode(16), unique=True, nullable=False), Column(‘password’, Unicode(40), nullable=False))

group_table = Table( ‘tf_group’, meta, Column(‘id’, Integer, primary_key=True), Column(‘group_name’, Unicode(16), unique=True, nullable=False))

meta.create_all() #创建所有的数据库(以上2个),函数无参数

result_set = group_table.select().execute() #选取 group_table的所有表数据

以下看一个关联多引擎的例子:

meta = MetaData()  #这里不能直接关联了 engine1 = create_engine(‘sqlite:///test1.db’)  #2个引擎 engine2 = create_engine(‘sqlite:///test2.db’)

# Use the engine parameter to load tables from the first engineuser_table = Table( ‘tf_user’, meta, autoload=True, autoload_with=engine1)  #从第一个引擎加载这些表 group_table = Table( ‘tf_group’, meta, autoload=True, autoload_with=engine1) permission_table = Table( ‘tf_permission’, meta, autoload=True, autoload_with=engine1) user_group_table = Table( ‘user_group’, meta, autoload=True, autoload_with=engine1) group_permission_table = Table( ‘group_permission’, meta, autoload=True, autoload_with=engine1)

meta.create_all(engine2) #在第二个引擎里面创建表

class ImageType(sqlalchemy.types.Binary):  #自定义我们的table的类 def convert_bind_param(self, value, engine): sfp = StringIO() value.save(sfp, ‘JPEG’) return sfp.getvalue() def convert_result_value(self, value, engine): sfp = StringIO(value) image = PIL.Image.open(sfp) return image  #这里我们定义了一个图形处理的类型

当定义了metadata后,会自定生成一个table.c object: q = user_table.select(  #查询创建在2007年6月1号之前的用户,并且第一个字母是’r’ user_table.c.user_name.like(‘r%’)  #这里的c就是那个特殊的类,当使用sql表达式会用到 & user_table.c.created < datetime(2007,6,1)) 或者替代这样: q = user_table.select(and_( user_table.c.user_name.like(‘r%’), user_table.c.created < datetime(2007,6,1))) 也可以使用rom映射: q = session.query(User) q = q.filter(User.c.user_name.like(‘r%’) & User.c.created > datetime(2007,6,1))

还是一个ORM的例子:

user_table = Table( ‘tf_user’, metadata, Column(‘id’, Integer, primary_key=True), Column(‘user_name’, Unicode(16), unique=True, nullable=False), Column(‘email_address’, Unicode(255), unique=True, nullable=False), Column(‘password’, Unicode(40), nullable=False), Column(‘first_name’, Unicode(255), default=”), Column(‘last_name’, Unicode(255), default=”), Column(‘created’, DateTime, default=datetime.now))  #这是一个定义的表类型

group_table = Table( ‘tf_group’, metadata, Column(‘id’, Integer, primary_key=True), Column(‘group_name’, Unicode(16), unique=True, nullable=False))

user_group = Table( ‘user_group’,  metadata, Column(‘user_id’, None, ForeignKey(‘tf_user.id’), primary_key=True), Column(‘group_id’, None, ForeignKey(‘tf_group.id’), … primary_key=True))

import sha class User(object):  #映射类

def _get_password(self): return self._password def _set_password(self, value): self._password = sha.new(value).hexdigest() #只存储用户的哈希密码 password=property(_get_password, _set_password)

def password_matches(self, password): return sha.new(password).hexdigest() == self._password

mapper(User, user_table, properties=dict( #映射将创建id, user_name, email_address, password, first_name, last_name, created等字段 _password=user_table.c.password)) #使用哈希后的密码替换真实密码,数据库只保存哈希后的,这里在orm上修改

mapper(User, user_table, properties=dict( _password=user_table.c.password, groups=relation(Group, secondary=user_group, backref=’users’))) #这里表示可以访问所有的组,用户只需访问一个成员团体属性,user_group映射类添加group和Group关联,

# User类添加users访问group属性,看效果: group1.users.append(user1)  #给group1添加用户user1,自动更新 user2.groups.append(group2) #把user2添加到group2组,自动更新