ORM框架SQLAlchemy

一 介紹 二 創(chuàng)建表 三 增刪改查 四 其他查詢相關(guān) 五 正查、反查 一 介紹 SQLAlchemy是Python編程語言下的一款ORM框架，該框架建立在數(shù)據(jù)庫API之上，使用關(guān)系對象映射進(jìn)行數(shù)據(jù)庫操作，簡言之便是：將對象轉(zhuǎn)換成SQL，然后使用數(shù)據(jù)API執(zhí)行SQL并獲取執(zhí)行結(jié)果。1、安裝pip3 install sqlalchemy 2、架構(gòu)與流程  #1、使用者通過ORM對象提交命令 #2、將命令交給SQLAlchemy Core（Schema/Types SQL Expression Language）轉(zhuǎn)換成SQL #3、使用 Engine/ConnectionPooling/Dialect 進(jìn)行數(shù)據(jù)庫操作 #3.1、匹配使用者事先配置好的egine #3.2、egine從連接池中取出一個鏈接 #3.3、基于該鏈接通過Dialect調(diào)用DB API，將SQL轉(zhuǎn)交給它去執(zhí)行 ！！！上述流程分析，可以大致分為兩個階段！！！：#第一個階段（流程1-2）：將SQLAlchemy的對象換成可執(zhí)行的sql語句#第二個階段（流程3）：將sql語句交給數(shù)據(jù)庫執(zhí)行 如果我們不依賴于SQLAlchemy的轉(zhuǎn)換而自己寫好sql語句，那是不是意味著可以直接從第二個階段開始執(zhí)行了，事實上正是如此，我們完全可以只用SQLAlchemy執(zhí)行純sql語句，如下from sqlalchemy import create_engine#1 準(zhǔn)備 # 需要事先安裝好pymysql # 需要事先創(chuàng)建好數(shù)據(jù)庫:create database db1 charset utf8;#2 創(chuàng)建引擎 egine=create_engine('mysql+pymysql://root@127.0.0.1/db1?charset=utf8')#3 執(zhí)行sql # egine.execute('create table if not EXISTS t1(id int PRIMARY KEY auto_increment,name char(32));')# cur=egine.execute('insert into t1 values(%s,%s);',[(1,"egon1"),(2,"egon2"),(3,"egon3")]) #按位置傳值# cur=egine.execute('insert into t1 values(%(id)s,%(name)s);',name='egon4',id=4) #按關(guān)鍵字傳值#4 新插入行的自增id # print(cur.lastrowid)#5 查詢 cur=egine.execute('select * from t1')cur.fetchone() #獲取一行 cur.fetchmany(2) #獲取多行 cur.fetchall() #獲取所有行3、DB APISQLAlchemy本身無法操作數(shù)據(jù)庫，其必須以來pymsql等第三方插件，Dialect用于和數(shù)據(jù)API進(jìn)行交流，根據(jù)配置文件的不同調(diào)用不同的數(shù)據(jù)庫API，從而實現(xiàn)對數(shù)據(jù)庫的操作，如：#1、MySQL-Pythonmysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>#2、pymysqlmysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]#3、MySQL-Connectormysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>#4、cx_Oracleoracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]更多詳見：http://docs.sqlalchemy.org/en/latest/dialects/index.html 二 創(chuàng)建表 ORM中：#類===>表 #對象==>表中的一行記錄 四張表:業(yè)務(wù)線,服務(wù),用戶,角色，利用ORM創(chuàng)建出它們，并建立好它們直接的關(guān)系from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,DateTime,Enum,ForeignKey,UniqueConstraint,ForeignKeyConstraint,Index from sqlalchemy.orm import sessionmakeregine=create_engine('mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8',max_overflow=5)Base=declarative_base()#創(chuàng)建單表:業(yè)務(wù)線 class Business(Base):__tablename__='business'id=Column(Integer,primary_key=True,autoincrement=True)bname=Column(String(32),nullable=False,index=True)#多對一:多個服務(wù)可以屬于一個業(yè)務(wù)線,多個業(yè)務(wù)線不能包含同一個服務(wù) class Service(Base):__tablename__='service'id=Column(Integer,primary_key=True,autoincrement=True)sname=Column(String(32),nullable=False,index=True)ip=Column(String(15),nullable=False)port=Column(Integer,nullable=False)business_id=Column(Integer,ForeignKey('business.id'))__table_args__=(UniqueConstraint(ip,port,name='uix_ip_port'),Index('ix_id_sname',id,sname))#一對一:一種角色只能管理一條業(yè)務(wù)線,一條業(yè)務(wù)線只能被一種角色管理 class Role(Base):__tablename__='role'id=Column(Integer,primary_key=True,autoincrement=True)rname=Column(String(32),nullable=False,index=True)priv=Column(String(64),nullable=False)business_id=Column(Integer,ForeignKey('business.id'),unique=True)#多對多:多個用戶可以是同一個role,多個role可以包含同一個用戶 class Users(Base):__tablename__='users'id=Column(Integer,primary_key=True,autoincrement=True)uname=Column(String(32),nullable=False,index=True)class Users2Role(Base):__tablename__='users2role'id=Column(Integer,primary_key=True,autoincrement=True)uid=Column(Integer,ForeignKey('users.id'))rid=Column(Integer,ForeignKey('role.id'))__table_args__=(UniqueConstraint(uid,rid,name='uix_uid_rid'),)def init_db():Base.metadata.create_all(egine)def drop_db():Base.metadata.drop_all(egine)if __name__ == '__main__':init_db()注：設(shè)置外鍵的另一種方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])三 增刪改查 表結(jié)構(gòu)from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmakeregine=create_engine('mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8',max_overflow=5)Base=declarative_base()#多對一:假設(shè)多個員工可以屬于一個部門,而多個部門不能有同一個員工(只有創(chuàng)建公司才把員工當(dāng)駱駝用,一個員工身兼數(shù)職) class Dep(Base):__tablename__='dep'id=Column(Integer,primary_key=True,autoincrement=True)dname=Column(String(64),nullable=False,index=True)class Emp(Base):__tablename__='emp'id=Column(Integer,primary_key=True,autoincrement=True)ename=Column(String(32),nullable=False,index=True)dep_id=Column(Integer,ForeignKey('dep.id'))def init_db():Base.metadata.create_all(egine)def drop_db():Base.metadata.drop_all(egine)drop_db() init_db() Session=sessionmaker(bind=egine) session=Session()增#增 row_obj=Dep(dname='銷售') #按關(guān)鍵字傳參,無需指定id,因其是自增長的 session.add(row_obj) session.add_all([Dep(dname='技術(shù)'),Dep(dname='運營'),Dep(dname='人事'), ])session.commit()刪#刪 session.query(Dep).filter(Dep.id > 3).delete() session.commit() 改#改 session.query(Dep).filter(Dep.id > 0).update({'dname':'哇哈哈'}) session.query(Dep).filter(Dep.id > 0).update({'dname':Dep.dname+'_SB'},synchronize_session=False) session.query(Dep).filter(Dep.id > 0).update({'id':Dep.id*100},synchronize_session='evaluate')session.commit()查#查所有,取所有字段 res=session.query(Dep).all() #for row in res:print(row.id,row.dname)#查所有,取指定字段 res=session.query(Dep.dname).order_by(Dep.id).all() #for row in res:print(row.dname)res=session.query(Dep.dname).first() print(res) # ('哇哈哈_SB',)#過濾查 res=session.query(Dep).filter(Dep.id > 1,Dep.id <1000) #逗號分隔,默認(rèn)為and print([(row.id,row.dname) for row in res])四 其他查詢相關(guān) 一 準(zhǔn)備表和數(shù)據(jù)from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmakeregine=create_engine('mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8',max_overflow=5)Base=declarative_base()#多對一:假設(shè)多個員工可以屬于一個部門,而多個部門不能有同一個員工(只有創(chuàng)建公司才把員工當(dāng)駱駝用,一個員工身兼數(shù)職) class Dep(Base):__tablename__='dep'id=Column(Integer,primary_key=True,autoincrement=True)dname=Column(String(64),nullable=False,index=True)class Emp(Base):__tablename__='emp'id=Column(Integer,primary_key=True,autoincrement=True)ename=Column(String(32),nullable=False,index=True)dep_id=Column(Integer,ForeignKey('dep.id'))def init_db():Base.metadata.create_all(egine)def drop_db():Base.metadata.drop_all(egine)drop_db() init_db() Session=sessionmaker(bind=egine) session=Session()# 準(zhǔn)備數(shù)據(jù) session.add_all([Dep(dname='技術(shù)'),Dep(dname='銷售'),Dep(dname='運營'),Dep(dname='人事'), ])session.add_all([Emp(ename='林海峰',dep_id=1),Emp(ename='李杰',dep_id=1),Emp(ename='武配齊',dep_id=1),Emp(ename='元昊',dep_id=2),Emp(ename='李鋼彈',dep_id=3),Emp(ename='張二丫',dep_id=4),Emp(ename='李坦克',dep_id=2),Emp(ename='王大炮',dep_id=4),Emp(ename='牛榴彈',dep_id=3) ])session.commit()二 條件、通配符、limit、排序、分組、連表、組合#一、條件 sql=session.query(Emp).filter_by(ename='林海峰') #filter_by只能傳參數(shù):什么等于什么 res=sql.all() #sql語句的執(zhí)行結(jié)果res=session.query(Emp).filter(Emp.id>0,Emp.ename == '林海峰').all() #filter內(nèi)傳的是表達(dá)式,逗號分隔,默認(rèn)為and, res=session.query(Emp).filter(Emp.id.between(1,3),Emp.ename == '林海峰').all() res=session.query(Emp).filter(Emp.id.in_([1,3,99,101]),Emp.ename == '林海峰').all() res=session.query(Emp).filter(~Emp.id.in_([1,3,99,101]),Emp.ename == '林海峰') #~代表取反,轉(zhuǎn)換成sql就是關(guān)鍵字notfrom sqlalchemy import and_,or_ res=session.query(Emp).filter(and_(Emp.id > 0,Emp.ename=='林海峰')).all() res=session.query(Emp).filter(or_(Emp.id < 2,Emp.ename=='功夫熊貓')).all() res=session.query(Emp).filter(or_(Emp.dep_id == 3,and_(Emp.id > 1,Emp.ename=='功夫熊貓'),Emp.ename != '') ).all()#二、通配符 res=session.query(Emp).filter(Emp.ename.like('%海_%')).all() res=session.query(Emp).filter(~Emp.ename.like('%海_%')).all()#三、limit res=session.query(Emp)[0:5:2]#四、排序 res=session.query(Emp).order_by(Emp.dep_id.desc()).all() res=session.query(Emp).order_by(Emp.dep_id.desc(),Emp.id.asc()).all()#五、分組 from sqlalchemy.sql import funcres=session.query(Emp.dep_id).group_by(Emp.dep_id).all() res=session.query(func.max(Emp.dep_id),func.min(Emp.dep_id),func.sum(Emp.dep_id),func.avg(Emp.dep_id),func.count(Emp.dep_id), ).group_by(Emp.dep_id).all()res=session.query(Emp.dep_id,func.count(1), ).group_by(Emp.dep_id).having(func.count(1) > 2).all()#六、連表 #笛卡爾積 res=session.query(Emp,Dep).all() #select * from emp,dep;#where條件 res=session.query(Emp,Dep).filter(Emp.dep_id==Dep.id).all() # for row in res: # emp_tb=row[0] # dep_tb=row[1] # print(emp_tb.id,emp_tb.ename,dep_tb.id,dep_tb.dname)#內(nèi)連接 res=session.query(Emp).join(Dep) #join默認(rèn)為內(nèi)連接,SQLAlchemy會自動幫我們通過foreign key字段去找關(guān)聯(lián)關(guān)系 #但是上述查詢的結(jié)果均為Emp表的字段,這樣鏈表還有毛線意義,于是我們修改為 res=session.query(Emp.id,Emp.ename,Emp.dep_id,Dep.dname).join(Dep).all()#左連接:isouter=True res=session.query(Emp.id,Emp.ename,Emp.dep_id,Dep.dname).join(Dep,isouter=True).all()#右連接:同左連接,只是把兩個表的位置換一下#七、組合 q1=session.query(Emp.id,Emp.ename).filter(Emp.id > 0,Emp.id < 5) q2=session.query(Emp.id,Emp.ename).filter(or_(Emp.ename.like('%海%'),Emp.ename.like('%昊%'),) ) res1=q1.union(q2) #組合+去重 res2=q1.union_all(q2) #組合,不去重print([i.ename for i in q1.all()]) #['林海峰', '李杰', '武配齊', '元昊'] print([i.ename for i in q2.all()]) #['林海峰', '元昊'] print([i.ename for i in res1.all()]) #['林海峰', '李杰', '武配齊', '元昊'] print([i.ename for i in res2.all()]) #['林海峰', '李杰', '武配齊', '元昊', '元昊', '林海峰']三 子查詢有三種形式的子查詢，注意：子查詢的sql必須用括號包起來,尤其在形式三中需要注意這一點#示例:查出id大于2的員工,當(dāng)做子查詢的表使用#原生SQL: # select * from (select * from emp where id > 2);#ORM: res=session.query(session.query(Emp).filter(Emp.id > 8).subquery() ).all()#示例:#查出銷售部門的員工姓名#原生SQL: # select ename from emp where dep_id in (select id from dep where dname='銷售');#ORM: res=session.query(Emp.ename).filter(Emp.dep_id.in_(session.query(Dep.id).filter_by(dname='銷售'), #傳的是參數(shù)# session.query(Dep.id).filter(Dep.dname=='銷售') #傳的是表達(dá)式 )).all()#示例:查詢所有的員工姓名與部門名#原生SQL: # select ename as 員工姓名,(select dname from dep where id = emp.dep_id) as 部門名 from emp;#ORM: sub_sql=session.query(Dep.dname).filter(Dep.id==Emp.dep_id) #SELECT dep.dname FROM dep, emp WHERE dep.id = emp.dep_id sub_sql.as_scalar() #as_scalar的功能就是把上面的sub_sql加上了括號res=session.query(Emp.ename,sub_sql.as_scalar()).all()五 正查、反查 一 表修改復(fù)制代碼 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmaker,relationshipegine=create_engine('mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8',max_overflow=5)Base=declarative_base()class Dep(Base):__tablename__='dep'id=Column(Integer,primary_key=True,autoincrement=True)dname=Column(String(64),nullable=False,index=True)class Emp(Base):__tablename__='emp'id=Column(Integer,primary_key=True,autoincrement=True)ename=Column(String(32),nullable=False,index=True)dep_id=Column(Integer,ForeignKey('dep.id'))#在ForeignKey所在的類內(nèi)添加relationship的字段,注意:#1:Dep是類名#2:depart字段不會再數(shù)據(jù)庫表中生成字段#3:depart用于Emp表查詢Dep表(正向查詢),而xxoo用于Dep表查詢Emp表(反向查詢),depart=relationship('Dep',backref='xxoo') def init_db():Base.metadata.create_all(egine)def drop_db():Base.metadata.drop_all(egine)drop_db() init_db() Session=sessionmaker(bind=egine) session=Session()# 準(zhǔn)備數(shù)據(jù) session.add_all([Dep(dname='技術(shù)'),Dep(dname='銷售'),Dep(dname='運營'),Dep(dname='人事'), ])session.add_all([Emp(ename='林海峰',dep_id=1),Emp(ename='李杰',dep_id=1),Emp(ename='武配齊',dep_id=1),Emp(ename='元昊',dep_id=2),Emp(ename='李鋼彈',dep_id=3),Emp(ename='張二丫',dep_id=4),Emp(ename='李坦克',dep_id=2),Emp(ename='王大炮',dep_id=4),Emp(ename='牛榴彈',dep_id=3) ])session.commit()二 標(biāo)準(zhǔn)連表查詢# 示例:查詢員工名與其部門名 res=session.query(Emp.ename,Dep.dname).join(Dep) #迭代器 for row in res:print(row[0],row[1]) #等同于print(row.ename,row.dname) 三 基于relationship的正查、反查#SQLAlchemy的relationship在內(nèi)部幫我們做好表的鏈接#查詢員工名與其部門名(正向查) res=session.query(Emp) for row in res:print(row.ename,row.id,row.depart.dname)#查詢部門名以及該部門下的員工(反向查) res=session.query(Dep) for row in res:# print(row.dname,row.xxoo)print(row.dname,[r.ename for r in row.xxoo])

轉(zhuǎn)載于:https://www.cnblogs.com/alex3174/p/11494837.html

總結(jié)

以上是生活随笔為你收集整理的ORM框架SQLAlchemy学习笔记的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。