一 介紹

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://[email protected]/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() #獲取所有行

View Code

3、DB API

SQLAlchemy本身無法操作資料庫，其必須以來pymsql等第三方外掛，Dialect用於和資料API進行交流，根據配置檔案的不同調用不同的資料庫API，從而實現對資料庫的操作，如：

#1、MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
#2、pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
#3、MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
#4、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創建出它們，並建立好它們直接的關係

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 sessionmaker

egine=create_engine('mysql+pymysql://[email protected]: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()

View Code

注：設定外來鍵的另一種方式 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 sessionmaker

egine=create_engine('mysql+pymysql://[email protected]: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()

View Code

增

#增
row_obj=Dep(dname='銷售') #按關鍵字傳參,無需指定id,因其是自增長的
session.add(row_obj)
session.add_all([
    Dep(dname='技術'),
    Dep(dname='運營'),
    Dep(dname='人事'),
])

session.commit()

View Code

刪

#刪
session.query(Dep).filter(Dep.id > 3).delete()
session.commit()

View Code

改

#改
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()

View Code

查

#查所有,取所有欄位
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])

View Code

四 其他查詢相關

一 準備表和資料

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://[email protected]: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()

View Code

二 條件、萬用字元、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就是關鍵字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()


#二、萬用字元
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 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()


#六、連表
#笛卡爾積
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()]) #['林海峰', '李傑', '武配齊', '元昊', '元昊', '林海峰']

View Code

三 子查詢

有三種形式的子查詢，注意：子查詢的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()

形式一:子查詢當做一張表來用,呼叫subquery()

#示例:#查出銷售部門的員工姓名

#原生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()

形式二:子查詢當做in的範圍用,呼叫in_

#示例:查詢所有的員工姓名與部門名

#原生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()

形式三:子查詢當做select後的欄位,呼叫as_scalar()

五 正查、反查

一 表修改

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://[email protected]: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()

View Code

二 標準連表查詢

# 示例:查詢員工名與其部門名
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])

一 介紹

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://[email protected]/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() #獲取所有行

View Code

3、DB API

SQLAlchemy本身無法操作資料庫，其必須以來pymsql等第三方外掛，Dialect用於和資料API進行交流，根據配置檔案的不同調用不同的資料庫API，從而實現對資料庫的操作，如：

#1、MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
#2、pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
#3、MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
#4、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創建出它們，並建立好它們直接的關係

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 sessionmaker

egine=create_engine('mysql+pymysql://[email protected]: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()

View Code

注：設定外來鍵的另一種方式 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 sessionmaker

egine=create_engine('mysql+pymysql://[email protected]: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()

View Code

增

#增
row_obj=Dep(dname='銷售') #按關鍵字傳參,無需指定id,因其是自增長的
session.add(row_obj)
session.add_all([
    Dep(dname='技術'),
    Dep(dname='運營'),
    Dep(dname='人事'),
])

session.commit()

View Code

刪

#刪
session.query(Dep).filter(Dep.id > 3).delete()
session.commit()

View Code

改

#改
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()

View Code

查

#查所有,取所有欄位
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])

View Code

四 其他查詢相關

一 準備表和資料

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://[email protected]: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()

View Code

二 條件、萬用字元、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就是關鍵字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()


#二、萬用字元
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 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()


#六、連表
#笛卡爾積
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()]) #['林海峰', '李傑', '武配齊', '元昊', '元昊', '林海峰']

View Code

三 子查詢

有三種形式的子查詢，注意：子查詢的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()

形式一:子查詢當做一張表來用,呼叫subquery()

#示例:#查出銷售部門的員工姓名

#原生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()

形式二:子查詢當做in的範圍用,呼叫in_

#示例:查詢所有的員工姓名與部門名

#原生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()

形式三:子查詢當做select後的欄位,呼叫as_scalar()

五 正查、反查

一 表修改

相關推薦