SQLAlchemy最新權(quán)威詳細(xì)教程

1、SQLAlchemy最新權(quán)威詳細(xì)教程前言:最近開(kāi)始學(xué)習(xí)SQLAlchemy,本教程是其官方文檔以及在讀英文版的翻譯加一些自己的理解和總結(jié)1 什么是 SQLAlchemy?它是給mysql, oracle,sqlite等關(guān)系型數(shù)據(jù)庫(kù)的python接口,不需要大幅修改原有的python代碼,它已經(jīng)包含了SQL表達(dá)式語(yǔ)言和ORM,看一些例子:sql=”INSERT INTO user(user_name, password) VALUES (%s, %s)”cursor = conn.cursor()cursor.execute(sql, (dongwm, testpass)以上是一個(gè)常用的mysq

2、l的SQL語(yǔ)句,但是冗長(zhǎng)也容易出錯(cuò),并且可能導(dǎo)致安全問(wèn)題(因?yàn)槭亲址恼Z(yǔ)句,會(huì)存在SQL注入),并且代碼不跨平臺(tái),在不同數(shù)據(jù)庫(kù)軟件的語(yǔ)句不同(以下是一個(gè)Oracle例子),不具備客移植性:sql=”INSERT INTO user(user_name, password) VALUES (:1, :2)”cursor = conn.cursor()cursor.execute(sql, dongwm, testpass)而在SQLAlchemy里只需要這樣寫(xiě):statement = user_table.insert(user_name=rick, password=parrot)state

3、ment.execute() #護(hù)略是什么數(shù)據(jù)庫(kù)環(huán)境SQLAlchemy還能讓你寫(xiě)出很pythonic的語(yǔ)句: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年創(chuàng)建的用戶metadata=MetaData(sqlite:/) # 告訴它你設(shè)置的數(shù)據(jù)庫(kù)類型是基于內(nèi)存的sqliteuser_table = Table( #創(chuàng)建一個(gè)表tf_user, met

4、adata,Column(id, Integer, primary_key=True), #一些字段,假設(shè)你懂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) #假設(shè)table已經(jīng)存在.就不需要指定字段,只是加個(gè)autoload=Trueclass User(object): pass #雖然SQLAlchemy強(qiáng)大,但是插入更新還是需要手動(dòng)指定,可以使用ORM,方法就是:設(shè)定一個(gè)類,定義一個(gè)表,把表映射到類里面mapper(User, user_table)下面是一個(gè)完整ORM的例子:Source

6、codefrom sqlalchemy.orm import mapper, sessionmaker #sessionmaker() 函數(shù)是最常使用的創(chuàng)建最頂層可用于整個(gè)應(yīng)用 Session 的方法,Session 管理著所有與數(shù)據(jù)庫(kù)之間的會(huì)話from datetime import datetimefrom sqlalchemy import Table, MetaData, Column, ForeignKey, Integer, String, Unicode, DateTime #會(huì)SQL的人能理解這些函數(shù)吧?engine = create_engine(sqlite:/tutori

7、al.db, echo=True) #創(chuàng)建到數(shù)據(jù)庫(kù)的連接,echo=True 表示用logging輸出調(diào)試結(jié)果metadata = MetaData() #跟蹤表屬性u(píng)ser_table = Table( #創(chuàng)建一個(gè)表所需的信息:字段,表名等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) #在數(shù)據(jù)庫(kù)中生成表class User(object): pass #創(chuàng)建一個(gè)映射類mapper(User, user_table) #把表映射到類Session = sessionm

9、aker() #創(chuàng)建了一個(gè)自定義了的 Session類Session.configure(bind=engine) #將創(chuàng)建的數(shù)據(jù)庫(kù)連接關(guān)聯(lián)到這個(gè)sessionsession = Session()u = User()u.user_name=dongwmu.email_address=u.password=testpass #給映射類添加以下必要的屬性,因?yàn)樯厦鎰?chuàng)建表指定這幾個(gè)字段不能為空session.add(u) #在session中添加內(nèi)容session.flush() #保存數(shù)據(jù)mit() #數(shù)據(jù)庫(kù)事務(wù)的提交,sisson自動(dòng)過(guò)期而

10、不需要關(guān)閉query = session.query(User) #query() 簡(jiǎn)單的理解就是select() 的支持 ORM 的替代方法,可以接受任意組合的 class/column 表達(dá)式print list(query) #列出所有userprint query.get(1) #根據(jù)主鍵顯示print query.filter_by(user_name=dongwm).first() #類似于SQL的where,打印其中的第一個(gè)u = query.filter_by(user_name=dongwm).first()u.password = newpass #修改其密碼字段sessi

11、mit() #提交事務(wù)print query.get(1).password #打印會(huì)出現(xiàn)新密碼for instance in session.query(User).order_by(User.id): #根據(jù)id字段排序,打印其中的用戶名和郵箱地址 print instance.user_name, instance.email_address既然是ORM框架,我們來(lái)一個(gè)更復(fù)雜的包含關(guān)系的例子,先看sql語(yǔ)句: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關(guān)聯(lián)了tf_user的id字段FOREIGN KEY(gr

14、oup_id) REFERENCES tf_group (id); #group_id關(guān)聯(lián)了 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關(guān)聯(lián) tf_group的id字段FOREIGN KEY(permission_id) REFERENCES tf_permiss

15、ion (id); #permission_id關(guān)聯(lián)了tf_permission 的id字段這是一個(gè)復(fù)雜的多對(duì)多的關(guān)系,比如檢查用戶是否有admin權(quán)限,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; 看起來(lái)太復(fù)雜并且繁長(zhǎng)了在面向?qū)ο蟮氖澜缋?是這樣的: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的權(quán)限的函數(shù) for g in user.groups: for p in g.permissions: #可以看出來(lái)使用了for循環(huán) 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是一個(gè)字典值。增加了一個(gè)groups 值，它又是一個(gè) relation 對(duì)象，這個(gè)對(duì)象實(shí)現(xiàn)#了Group類與user_group的 映射。這樣我通過(guò)user_table的groups 屬性就可以反映出RssFeed的值來(lái),#中間表對(duì)象(user_group)傳給secondary參數(shù)，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)假如計(jì)算組里用戶數(shù)(不包含忘記刪除但是重復(fù)的)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) #雖然也長(zhǎng),但是減少了數(shù)據(jù)庫(kù)查詢次數(shù),也就是讓簡(jiǎn)單事情簡(jiǎn)單化,復(fù)雜事情可能簡(jiǎn)單解決看一個(gè)綜合的例子:class User(object): #這些類設(shè)計(jì)數(shù)據(jù)庫(kù)的模型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)這里沒(méi)有修改對(duì)象,而join_group,leave_group這樣的函數(shù)依然可用,sqlalchemy會(huì)跟蹤變化,并且自動(dòng)刷新數(shù)據(jù)庫(kù)上面介紹了一個(gè)

28、完整的例子,連接數(shù)據(jù)庫(kù)嗨可以這樣: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、線程減少對(duì)數(shù)據(jù)庫(kù)的訪問(wèn)from sqlalchemy.databases.mysql import MSEnum, MSBigInteger #這個(gè) sqlalchemy.databases是某數(shù)據(jù)庫(kù)軟件的方言集合,只支持特定平臺(tái)user_table = Table(tf_user, meta,Column(id, MSBigInteger),Column(honorific, MSEnum(Mr, Mrs, Ms, Miss, Dr, Prof)以下是幾個(gè)MetaData的應(yīng)用:unbound_meta = MetaData() #這個(gè)metadata沒(méi)有綁定db1 = create_eng

30、ine(sqlite:/)unbound_meta.bind = db1 #關(guān)聯(lián)引擎db2 = MetaData(sqlite:/test1.db) #直接設(shè)置引擎bound_meta1 = MetaData(db2)# Create a bound MetaData with an implicitly created enginebound_meta2 = MetaData(sqlite:/test2.db) #隱式綁定引擎meta = MetaData(sqlite:/) #直接綁定引擎可以讓源數(shù)據(jù)直接訪問(wèn)數(shù)據(jù)庫(kù)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() #創(chuàng)建所有的數(shù)據(jù)庫(kù)(以

32、上2個(gè)),函數(shù)無(wú)參數(shù)result_set = group_table.select().execute() #選取 group_table的所有表數(shù)據(jù)以下看一個(gè)關(guān)聯(lián)多引擎的例子:meta = MetaData() #這里不能直接關(guān)聯(lián)了engine1 = create_engine(sqlite:/test1.db) #2個(gè)引擎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) #從第一個(gè)引擎加載這些表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) #在第二個(gè)引擎里面創(chuàng)建表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 #這里我們定義了一個(gè)圖形處理的類型當(dāng)定義了metadata后,會(huì)自定生成一個(gè)table.c object:q = user_table.select( #查詢創(chuàng)建在2007年6月1號(hào)之前的用戶,并且第一個(gè)字母是ruser_table.c.user_name.like(r%) #這里的c就是那個(gè)特殊的類,當(dāng)使用sql表達(dá)式會(huì)用到& 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)還是一個(gè)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) #這是一個(gè)定義的表類型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() #只存儲(chǔ)用戶的哈希密碼password=property(_get_password, _set_password)def password_matches(self, password):return sha.new(password).hexdigest() = self._passwordmapper(User, user_table, properties=dict( #映射將創(chuàng)建id, user_name, email_address, password, first_nam

40、e, last_name, created等字段_password=user_table.c.password) #使用哈希后的密碼替換真實(shí)密碼,數(shù)據(jù)庫(kù)只保存哈希后的,這里在orm上修改mapper(User, user_table, properties=dict(_password=user_table.c.password,groups=relation(Group, secondary=user_group, backref=users) #這里表示可以訪問(wèn)所有的組，用戶只需訪問(wèn)一個(gè)成員團(tuán)體屬性,user_group映射類添加group和Group關(guān)聯(lián),# User類添加users訪問(wèn)

41、group屬性,看效果:group1.users.append(user1) #給group1添加用戶user1,自動(dòng)更新user2.groups.append(group2) #把user2添加到group2組,自動(dòng)更新對(duì)于SQLAlchemy的一些總結(jié):1 metadata.create_all()創(chuàng)建多個(gè)table可以這樣使用,但是他還有個(gè)功能,它添加了”IF NOT EXISTS”,就是在數(shù)據(jù)庫(kù)存在的時(shí)候,他還是安全的2 交互模式下的一個(gè)全過(guò)程: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() #插入數(shù)據(jù) 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) #這個(gè)實(shí)例可以多次插入,和sql區(qū)別很大 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() #只獲取符合要求的第一項(xiàng) 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) #過(guò)濾留下user_name=dongwm1的項(xiàng) print stmt.execute().fetchall() #獲取所有符合項(xiàng)(1, udongwm1, usecret, utestdongwm1, datetime.datetime(2012, 7, 17, 11, 57, 48, ) s

48、tmt = user_table.update(user_table.c.user_name=dongwm1) #更新數(shù)據(jù) stmt.execute(password=secret123) #修改密碼 stmt = user_table.delete(user_table.c.user_name != dongwm1) #刪除user_name不是dongwm1的條目 stmt.execute() user_table.select().execute().fetchall() #查詢發(fā)現(xiàn)就剩一條了(1, udongwm1, usecret123, utestdongwm1, datetime

49、.datetime(2012, 7, 17, 11, 57, 48, )3 sission上面已經(jīng)說(shuō)過(guò)了,補(bǔ)充一些:session.delete(u) #把映射類從會(huì)話中刪除4 關(guān)于引擎引擎就是根據(jù)不同的數(shù)據(jù)庫(kù)方言連接數(shù)據(jù)庫(kù)的方法以下是一些例子(方法 driver:/username:passwordhost:port/database):engine = create_engine(sqlite:/) #連接基于內(nèi)存的sqliteengine = create_engine(sqlite:/data.sqlite) #連接基于硬盤(pán)文件的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協(xié)議的Oracleengine =create_engine(oracle:/dongwm:foolocalhost:1521/oracle_sid) #連接沒(méi)有TNS名字的Oracle也可以帶一些參數(shù):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)還可以通過(guò)函數(shù)完全控制連接: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) #可以給它添加一個(gè)日志文件處理類handler.level = logging.DEBUGlogging.getLogger(sqlalchemy.engine).addHandler(handler)上面說(shuō)的操作表,也可以直接操作數(shù)據(jù)庫(kù):conn = engine.connect()result = conn.execute(select user_name, email_address from tf_user) #結(jié)果是一個(gè)sqlalche

53、my.engine.ResultProxy的實(shí)例for row in result:print User name: %s Email address: %s % (rowuser_name, rowemail_address)conn.close()from sqlalchemy import pool #本來(lái)它已經(jīng)自動(dòng)通過(guò)數(shù)據(jù)庫(kù)連接管理數(shù)據(jù)池,但是也可以手動(dòng)管理import psycopg2psycopg = pool.manage(psycopg2) #結(jié)果是一個(gè)sqlalchemy.pool.DBProxy實(shí)例connection = psycopg.connect(database=mydb,username=rick, password=foo)5 關(guān)于元數(shù)據(jù)metadata它收集了描述table對(duì)象等的元數(shù)據(jù)類,當(dāng)使用ORM等時(shí)必須使用metadata如果他被綁定了,那么使用table.create()就會(huì)生成表,沒(méi)有綁定需要:table.create(