一 介紹
SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。javascript
一、安裝
pip3 install sqlalchemy
二、架構與流程
#一、使用者經過ORM對象提交命令 #二、將命令交給SQLAlchemy Core（Schema/Types SQL Expression Language）轉換成SQL #三、使用 Engine/ConnectionPooling/Dialect 進行數據庫操做 #3.一、匹配使用者事先配置好的egine #3.二、egine從鏈接池中取出一個連接 #3.三、基於該連接經過Dialect調用DB API，將SQL轉交給它去執行
！！！上述流程分析，能夠大體分爲兩個階段！！！：html
#第一個階段（流程1-2）：將SQLAlchemy的對象換成可執行的sql語句 #第二個階段（流程3）：將sql語句交給數據庫執行
若是咱們不依賴於SQLAlchemy的轉換而本身寫好sql語句，那是否是意味着能夠直接從第二個階段開始執行了，事實上正是如此，咱們徹底能夠只用SQLAlchemy執行純sql語句，以下java
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() #獲取全部行
三、DB APImysql
SQLAlchemy自己沒法操做數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：sql
#一、MySQL-Python mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> #二、pymysql mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] #三、MySQL-Connector mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> #四、cx_Oracle oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多詳見：http://docs.sqlalchemy.org/en/latest/dialects/index.html 數據庫
二 建立表
ORM中：編程
#類===>表 #對象==>表中的一行記錄
四張表:業務線,服務,用戶,角色，利用ORM建立出它們，並創建好它們直接的關係session
View Code
注：設置外鍵的另外一種方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])架構
三 增刪改查
表結構oracle
from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmaker egine=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 sessionmaker egine=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、排序、分組、連表、組合
#1、條件 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就是關鍵字not from 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() #2、通配符 res=session.query(Emp).filter(Emp.ename.like('%海_%')).all() res=session.query(Emp).filter(~Emp.ename.like('%海_%')).all() #3、limit res=session.query(Emp)[0:5:2] #4、排序 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() #5、分組 from sqlalchemy.sql import func res=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() #6、連表 #笛卡爾積 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() #右鏈接:同左鏈接,只是把兩個表的位置換一下 #7、組合 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,relationship egine=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])