ORM框架SQLAlchemy

一 介紹 二 創建表 三 增刪改查 四 其他查詢相關 五 正查、反查 一 介紹 SQLAlchemy是Python編程語言下的一款ORM框架，該框架建立在數據庫API之上，使用關系對象映射進行數據庫操作，簡言之便是：將對象轉換成SQL，然后使用數據API執行SQL并獲取執行結果。1、安裝pip3 install sqlalchemy 2、架構與流程  #1、使用者通過ORM對象提交命令 #2、將命令交給SQLAlchemy Core（Schema/Types SQL Expression Language）轉換成SQL #3、使用 Engine/ConnectionPooling/Dialect 進行數據庫操作 #3.1、匹配使用者事先配置好的egine #3.2、egine從連接池中取出一個鏈接 #3.3、基于該鏈接通過Dialect調用DB API，將SQL轉交給它去執行 ！！！上述流程分析，可以大致分為兩個階段！！！：#第一個階段（流程1-2）：將SQLAlchemy的對象換成可執行的sql語句#第二個階段（流程3）：將sql語句交給數據庫執行 如果我們不依賴于SQLAlchemy的轉換而自己寫好sql語句，那是不是意味著可以直接從第二個階段開始執行了，事實上正是如此，我們完全可以只用SQLAlchemy執行純sql語句，如下from sqlalchemy import create_engine#1 準備 # 需要事先安裝好pymysql # 需要事先創建好數據庫:create database db1 charset utf8;#2 創建引擎 egine=create_engine('mysql+pymysql://root@127.0.0.1/db1?charset=utf8')#3 執行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) #按關鍵字傳值#4 新插入行的自增id # print(cur.lastrowid)#5 查詢 cur=egine.execute('select * from t1')cur.fetchone() #獲取一行 cur.fetchmany(2) #獲取多行 cur.fetchall() #獲取所有行3、DB APISQLAlchemy本身無法操作數據庫，其必須以來pymsql等第三方插件，Dialect用于和數據API進行交流，根據配置文件的不同調用不同的數據庫API，從而實現對數據庫的操作，如：#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 二 創建表 ORM中：#類===>表 #對象==>表中的一行記錄 四張表:業務線,服務,用戶,角色，利用ORM創建出它們，并建立好它們直接的關系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()#創建單表:業務線 class Business(Base):__tablename__='business'id=Column(Integer,primary_key=True,autoincrement=True)bname=Column(String(32),nullable=False,index=True)#多對一:多個服務可以屬于一個業務線,多個業務線不能包含同一個服務 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))#一對一:一種角色只能管理一條業務線,一條業務線只能被一種角色管理 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()注：設置外鍵的另一種方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])三 增刪改查 表結構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()#多對一:假設多個員工可以屬于一個部門,而多個部門不能有同一個員工(只有創建公司才把員工當駱駝用,一個員工身兼數職) 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='銷售') #按關鍵字傳參,無需指定id,因其是自增長的 session.add(row_obj) session.add_all([Dep(dname='技術'),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) #逗號分隔,默認為and print([(row.id,row.dname) for row in res])四 其他查詢相關 一 準備表和數據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()#多對一:假設多個員工可以屬于一個部門,而多個部門不能有同一個員工(只有創建公司才把員工當駱駝用,一個員工身兼數職) 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()# 準備數據 session.add_all([Dep(dname='技術'),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只能傳參數:什么等于什么 res=sql.all() #sql語句的執行結果res=session.query(Emp).filter(Emp.id>0,Emp.ename == '林海峰').all() #filter內傳的是表達式,逗號分隔,默認為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 == '林海峰') #~代表取反,轉換成sql就是關鍵字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)#內連接 res=session.query(Emp).join(Dep) #join默認為內連接,SQLAlchemy會自動幫我們通過foreign key字段去找關聯關系 #但是上述查詢的結果均為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的員工,當做子查詢的表使用#原生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='銷售'), #傳的是參數# session.query(Dep.id).filter(Dep.dname=='銷售') #傳的是表達式 )).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()五 正查、反查 一 表修改復制代碼 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所在的類內添加relationship的字段,注意:#1:Dep是類名#2:depart字段不會再數據庫表中生成字段#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()# 準備數據 session.add_all([Dep(dname='技術'),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()二 標準連表查詢# 示例:查詢員工名與其部門名 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在內部幫我們做好表的鏈接#查詢員工名與其部門名(正向查) 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])

轉載于:https://www.cnblogs.com/alex3174/p/11494837.html

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。