SQLAlchemy最新权威详细教程

1、SQLAlchemy最新权威详细教程前言:最近开始学习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)以上是一个常用的mysq

2、l的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)state

3、ment.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:/) # 告诉它你设置的数据库类型是基于内存的sqliteuser_table = Table( #创建一个表tf_user, met

4、adata,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_nam

5、e, Unicode(255), default=”),Column(created, DateTime, default=datetime.now)users_table = Table(users, metadata, autoload=True) #假设table已经存在.就不需要指定字段,只是加个autoload=Trueclass User(object): pass #虽然SQLAlchemy强大,但是插入更新还是需要手动指定,可以使用ORM,方法就是:设定一个类,定义一个表,把表映射到类里面mapper(User, user_table)下面是一个完整ORM的例子:Source

6、codefrom sqlalchemy.orm import mapper, sessionmaker #sessionmaker() 函数是最常使用的创建最顶层可用于整个应用 Session 的方法,Session 管理着所有与数据库之间的会话from datetime import datetimefrom sqlalchemy import Table, MetaData, Column, ForeignKey, Integer, String, Unicode, DateTime #会SQL的人能理解这些函数吧?engine = create_engine(sqlite:/tutori

7、al.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

8、),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 = sessionm

9、aker() #创建了一个自定义了的 Session类Session.configure(bind=engine) #将创建的数据库连接关联到这个sessionsession = Session()u = User()u.user_name=dongwmu.email_address=u.password=testpass #给映射类添加以下必要的属性,因为上面创建表指定这几个字段不能为空session.add(u) #在session中添加内容session.flush() #保存数据mit() #数据库事务的提交,sisson自动过期而

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

11、mit() #提交事务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_addres

12、s 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 TAB

13、LE 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(gr

14、oup_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_permiss

15、ion (id); #permission_id关联了tf_permission 的id字段这是一个复杂的多对多的关系,比如检查用户是否有admin权限,sql需要这样:SELECT COUNT(*) FROM tf_user, tf_group, tf_permission WHEREtf_user.user_name=dongwm AND tf_user.id=user_group.user_idAND user_group.group_id = group_permission.group_idAND group_permission.permission_id = tf_permiss

16、ion.idAND permission_name=admin; 看起来太复杂并且繁长了在面向对象的世界里,是这样的:class User(object):groups=class Group(object):users=permissions=class Permission(object):groups=Source codeprint Summary for %s % user.user_namefor g in user.groups: print Member of group %s % g.group_name for p in g.permissions: print . whi

17、ch has permission %s % p.permission_nameSource codedef 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=di

18、ct(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(

19、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 Per

20、mission %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.permissi

21、on_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

22、 is None: users = if permissions is None: permissions = self.group_name = group_nameself._users = usersself._permissions = permissionsdef 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(permis

23、sion)def del_permission(self, permission):self._permissions.remove(permission)class Permission(object):def _init_(self, permission_name=None, groups=None):self.permission_name = permission_nameself._groups = groupsdef join_group(self, group):self._groups.append(group)def leave_group(self, group):sel

24、f._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),C

25、olumn(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),prima

26、ry_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, propert

27、ies=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会跟踪变化,并且自动刷新数据库上面介绍了一个

28、完整的例子,连接数据库嗨可以这样:engine = create_engine(sqlite:/)connection = engine.connect() #使用connectresult = connection.execute(“select user_name from tf_user”)for row in result:print user name: %s % rowuser_nameresult.close()engine = create_engine(sqlite:/, strategy=threadlocal) #,strategy=threadlocal表示重用其它本地

29、线程减少对数据库的访问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_eng

30、ine(sqlite:/)unbound_meta.bind = db1 #关联引擎db2 = MetaData(sqlite:/test1.db) #直接设置引擎bound_meta1 = MetaData(db2)# Create a bound MetaData with an implicitly created enginebound_meta2 = MetaData(sqlite:/test2.db) #隐式绑定引擎meta = MetaData(sqlite:/) #直接绑定引擎可以让源数据直接访问数据库user_table = Table(tf_user, meta,Colum

31、n(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() #创建所有的数据库(以

32、上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

33、, 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_permis

34、sion_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(

35、self, value, engine):sfp = StringIO(value)image = PIL.Image.open(sfp)return image #这里我们定义了一个图形处理的类型当定义了metadata后,会自定生成一个table.c object:q = user_table.select( #查询创建在2007年6月1号之前的用户,并且第一个字母是ruser_table.c.user_name.like(r%) #这里的c就是那个特殊的类,当使用sql表达式会用到& user_table.c.created datetime(2007,6,1)或者替代这样:q = us

36、er_table.select(and_(user_table.c.user_name.like(r%),user_table.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=Fals

37、e),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(1

38、6), 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 shaclass User(object): #映射类def _get_password(self):return self._passworddef _set_password(se

39、lf, value):self._password = sha.new(value).hexdigest() #只存储用户的哈希密码password=property(_get_password, _set_password)def password_matches(self, password):return sha.new(password).hexdigest() = self._passwordmapper(User, user_table, properties=dict( #映射将创建id, user_name, email_address, password, first_nam

40、e, 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访问

41、group属性,看效果:group1.users.append(user1) #给group1添加用户user1,自动更新user2.groups.append(group2) #把user2添加到group2组,自动更新对于SQLAlchemy的一些总结:1 metadata.create_all()创建多个table可以这样使用,但是他还有个功能,它添加了”IF NOT EXISTS”,就是在数据库存在的时候,他还是安全的2 交互模式下的一个全过程:Source codedongwmlocalhost $ pythonPython 2.7.3 (default, Jul 11 2012,

42、10:10:17) GCC 4.5.3 on linux2Type help, copyright, credits or license for more information. from sqlalchemy import create_engine from sqlalchemy import Table, MetaData, Column, ForeignKey, Integer, String, Unicode, DateTime from datetime import datetime metadata = MetaData(sqlite:/tutorial.db) user_

43、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),. Column(display_name, Unicode(255), default=),. Column(created, DateTime, default=datetime.now)_main_:7: SAWarning

44、: Unicode column received non-unicode default value. stmt = user_table.insert() #插入数据 stmt.execute(user_name=dongwm1, password=secret,display_name=testdongwm1)/usr/lib/python2.7/site-packages/SQLAlchemy-0.7.8-py2.7-linux-i686.egg/sqlalchemy/engine/default.py:463: SAWarning: Unicode type received non

45、-unicode bind param value. param.append(processorskey(compiled_paramskey) stmt.execute(user_name=dongwm2, password=secret,display_name=testdongwm2) #这个实例可以多次插入,和sql区别很大 stmt = user_table.select() #select查询 result = stmt.execute() for row in result:. print row . (1, udongwm1, usecret, utestdongwm1, d

46、atetime.datetime(2012, 7, 17, 11, 57, 48, )(2, udongwm2, usecret, utestdongwm2, datetime.datetime(2012, 7, 17, 11, 58, 5, ) result = stmt.execute() row =result.fetchone() #只获取符合要求的第一项 print rowuser_namedongwm1 print row.passwordsecret print row.items()(uid, 1), (uuser_name, udongwm1), (upassword, us

47、ecret), (udisplay_name, utestdongwm1), (ucreated, datetime.datetime(2012, 7, 17, 11, 57, 48, ) stmt = user_table.select(user_table.c.user_name=dongwm1) #过滤留下user_name=dongwm1的项 print stmt.execute().fetchall() #获取所有符合项(1, udongwm1, usecret, utestdongwm1, datetime.datetime(2012, 7, 17, 11, 57, 48, ) s

48、tmt = user_table.update(user_table.c.user_name=dongwm1) #更新数据 stmt.execute(password=secret123) #修改密码 stmt = user_table.delete(user_table.c.user_name != dongwm1) #删除user_name不是dongwm1的条目 stmt.execute() user_table.select().execute().fetchall() #查询发现就剩一条了(1, udongwm1, usecret123, utestdongwm1, datetime

49、.datetime(2012, 7, 17, 11, 57, 48, )3 sission上面已经说过了,补充一些:session.delete(u) #把映射类从会话中删除4 关于引擎引擎就是根据不同的数据库方言连接数据库的方法以下是一些例子(方法 driver:/username:passwordhost:port/database):engine = create_engine(sqlite:/) #连接基于内存的sqliteengine = create_engine(sqlite:/data.sqlite) #连接基于硬盘文件的sqliteengine = create_engine

50、(postgres:/dongwm:foolocalhost:5432/pg_db) #连接postgresqlengine = create_engine(mysql:/localhost/mysql_db) #连接mysqlengine = create_engine(oracle:/dongwm:foooracle_tns) #连接基于TNS协议的Oracleengine =create_engine(oracle:/dongwm:foolocalhost:1521/oracle_sid) #连接没有TNS名字的Oracle也可以带一些参数:url=postgres:/dongwm:fo

51、olocalhost/pg_db?arg1=foo&arg2=barengine = create_engine(url)或者:engine = create_engine(postgres:/dongwm:foolocalhost/pg_db,connect_args=dict(arg1=foo, arg2=bar)还可以通过函数完全控制连接:import psycopgdef connect_pg():return psycopg.connect(user=rick, host=localhost)engine = create_engine(postgres:/, creator=con

52、nect_pg)import logginghandler = logging.FileHandler(sqlalchemy.engine.log) #可以给它添加一个日志文件处理类handler.level = logging.DEBUGlogging.getLogger(sqlalchemy.engine).addHandler(handler)上面说的操作表,也可以直接操作数据库:conn = engine.connect()result = conn.execute(select user_name, email_address from tf_user) #结果是一个sqlalche

53、my.engine.ResultProxy的实例for row in result:print User name: %s Email address: %s % (rowuser_name, rowemail_address)conn.close()from sqlalchemy import pool #本来它已经自动通过数据库连接管理数据池,但是也可以手动管理import psycopg2psycopg = pool.manage(psycopg2) #结果是一个sqlalchemy.pool.DBProxy实例connection = psycopg.connect(database=mydb,username=rick, password=foo)5 关于元数据metadata它收集了描述table对象等的元数据类,当使用ORM等时必须使用metadata如果他被绑定了,那么使用table.create()就会生成表,没有绑定需要:table.create(