python之SQLAIchemy

時間 2019-11-05

標籤 python sqlaichemy 欄目 Python 简体版

原文原文鏈接

一簡介和安裝

1 簡介

SQLALchemy 是一個第三方的ORM 框架，大量使用元編程實現其基本功能python

相關參考文檔:https://docs.sqlalchemy.org/en/13/mysql

2 基本組成

一個鏈接池引擎中存放着鏈接池和方言(dialect)，dialect不一樣的數據庫引擎對應不一樣的方言。sql

DBAPI經過API調用數據庫的各類功能，如控制語句等。數據庫

2 安裝

1 安裝

pip install sqlalchemy

2 版本檢測

pip show  sqlalchemy

二基本實踐

1 建立鏈接

數據庫連接的事情，交給引擎處理編程

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine

#建立存儲引擎
username='root'
password='Admin@Root123'
ip='192.168.1.200'
port=3306
databases='test'

# echo =True 打印SQL執行過程，主要應用於調試模式

engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)

2 建立實體類

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String

#建立存儲引擎
username='root'
password='Admin@Root123'
ip='192.168.1.200'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

3 實例化

#實例化
L=Login(name='admin')
L.age=30

4 建立表

可使用SQLAIchemy 來建立，刪除表安全

刪除繼承自Base的全部表
Base.metadata.drop_all(engine)
建立繼承自Base的全部表
Base.metadata.create_all(engine)session

生產環境不多這樣建立表，都是系統上線的時候由腳本生成的，生產環境不多刪除表，廢棄都不能刪除。app

具體代碼以下框架

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine) # 此處經過引擎提供的鏈接數據庫的能力來建立默認字符集和相關檢測方式，其被存儲在Base元類中。

執行結果以下 ide

數據庫結果以下

5 建立會話

在一個會話中操做數據庫，會話創建在鏈接上，鏈接被引擎管理

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做，此處不會commit
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

6 CRUD 操做(增，刪，改，查)

1 增

1 增長單行數據

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:
    ed_login = Login(name='admin', loginname='admin', password='admin')
    # 另外一種插入方式 
    # ed_login=Login()
    # ed_login.name='root'
    # ed_login.password='root'
    # ed_login.loginname='root'
    session.add(ed_login)  #增長狀態
    session.commit()
except  Exception as  e:   # 若拋出異常，則直接回滾
    print (e)
    session.rollback()
finally:
    pass

結果以下

數據庫結果以下

2 同時插入多行

add_all() 相關源碼

def add_all(self, instances):
        """Add the given collection of instances to this ``Session``."""

        if self._warn_on_events:
            self._flush_warning("Session.add_all()")

        for instance in instances:  # 此處也是經過add的方式進行遍歷加入的
            self.add(instance, _warn=False)

代碼以下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:
    lst=[]
    for  i in range(5):
        lst.append(Login(name='admin'+str(i), loginname='admin'+str(i), password='admin'+str(i)))
    print (lst)
    session.add_all(lst)
    session.commit()
except  Exception as  e:   # 若拋出異常，則直接回滾
    print (e)
    session.rollback()
finally:
    pass

結果以下

數據庫結果以下

3 屢次提交commit是否提交屢次問題

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:
    lst=[]
    for  i in range(5):
        lst.append(Login(name='admin'+str(i), loginname='admin'+str(i), password='admin'+str(i)))
    print (lst)
    session.add_all(lst)
    print  ('-----commit------')
    session.commit()
except  Exception as  e:   # 若拋出異常，則直接回滾
    print (e)
    session.rollback()
finally:
    pass

結果以下

結論：如上，屢次提交時commit只執行一次。

4 修改更新數據並提交

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:
    ed_login = Login(name='admin', loginname='admin', password='admin')
    session.add(ed_login)
    session.commit()

    # 修改相關屬性
    ed_login.loginname='1322534564356464564'
    session.add(ed_login)
    print  ('-----commit------')
    session.commit()
except  Exception as  e:   # 若拋出異常，則直接回滾
    print (e)
    session.rollback()
finally:
    pass

執行代碼結果顯示以下

數據庫結果以下

結論:

常規來講，上述代碼應該是提交了兩次，其並未進行相關的限制，但其顯示結果倒是修改了，而不是兩次提交，此處便涉及到了狀態變化，應該是進行了某種判斷，此處add後狀態未發生變化，所以只添加了一條

insert 只能返回你添加了幾行，下面的狀況會修改，其修改會使用update進行處理，更新的時候，通常須要給ID(主鍵)進行處理，若是沒有，則insert後拿取對應的id進行相關的處理，及提交完成後若出現更改，則經過必定方式獲取上次提交的主鍵，並進行相關的操做處理，當一次commit提交完成後，實例的無主鍵狀態會變爲有主鍵狀態。

2 簡單查詢語句

使用query()方法，返回一個Query對象

查詢相關源碼

def query(self, *entities, **kwargs):
        """Return a new :class:`.Query` object corresponding to this
        :class:`.Session`."""

        return self._query_cls(entities, self, **kwargs)

上述可知，其查詢參數是可變參數和關鍵字參數，及支持多級查詢

相關代碼以下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:
    login=session.query(Login)  # 此處是實體類,此處是未進行限制，默認查詢因此
    for  i  in  login:
        print (i)
    print ('----------------')
    data=session.query(Login).get(10)  # 經過主鍵id進行相關的查詢
    print (data)
    print ('```````````````````')
    data1=session.query(Login).filter(Login.id==1)  # 此處經過filter進行操做，注意，此處是python命令，是雙==號
    print (data1)
except  Exception as  e:   # 若拋出異常，則直接回滾
    print (e)
    session.rollback()
finally:
    pass

結果以下

3 修改

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases),echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:

    data=session.query(Login).get(11)  # 經過主鍵id進行相關的查詢
    print  (data)
    data.password='passwd'
    data.loginname='abc'
    session.add(data)
    session.commit()
except  Exception as  e:   # 若拋出異常，則直接回滾
    print (e)
    session.rollback()
finally:
    pass

結果以下

數據庫結果以下

具體步驟以下

1 先獲取到查詢結果
2 進行相關的修改操做
3 最後提交

4 刪除

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases),echo=True)
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

#  實例化並進行插入操做
try:

    login=Login(id=10)  # 實例化
    session.delete(login)  # 刪除id對應的值
    session.commit()  # 提交 
except  Exception as  e:   # 若拋出異常，則直接回滾
    print ('異常以下：--------',e)
    session.rollback()
finally:
    pass

結果以下

5 狀態處理

上述產生一個異常，是未進行持久化的異常，每個實體，都有狀態屬性_sa_instance_state，其類型是sqlchemy.orm.state.InstanceState,可使用sqlalchemy.inspect(entity)函數查看狀態

常見的狀態值有 transient，pending，persistent，deleted，detached。

狀態	說明
transient	實體類還沒有加入到session中，同時並無保存到數據庫中
pending	transient的實體被add()到session中，狀態切換到pending，可是尚未flush到數據庫中
persistent	session中的實體對象對應着數據庫中的真實記錄，pending狀態在提交成功後能夠變成persistent狀態，或者查詢成功返回的實體也是persistent狀態
deleted	實體被刪除且已經flush但未commit完成，事務提交成功了，實體變成detached，事務失敗，返回persistent
detached	刪除成功的實體進入此狀態

解釋

新建一個實體，狀態是transient臨時的
一旦add()後從transient狀態變爲pending狀態
成功commit()後從pending變爲persistent狀體
成功查詢返回的實體對象，也是persistent狀態
及就是 persistent狀態能夠由兩種形式組成(commit和 get成功)
persistent狀態的實體，修改依然是persistent狀態
persistent狀態的實體，刪除後，flus後但沒有commit，就變成了deleted狀態，成功提交，變成detached狀態，失敗，還原persistent狀態
刪除，修改操做，須要對應一個真實的記錄，因此要求實體對象必須是persistent狀態

狀態處理實例

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String,inspect
# 導入會話模塊
from sqlalchemy.orm import sessionmaker

#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 導入基類
Base=declarative_base()

# 建立實體類
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 設置id列爲整形，並設置其爲主鍵 ，整形時Inteager
    name=Column(String(48))  # 定義name名稱，並定義其字段爲charset類型和長度爲48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return "<Login (id='%s',name='%s',loginname='%s',password='%s')>"  %(self.id,self.name,self.loginname,self.password)

#實例化
L=Login(name='admin')
L.age=30

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)

#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()

# 建立一個狀態查詢函數
def show(entiry):
    ins=inspect(entiry)
    return (ins.transient,ins.pending,ins.persistent,ins.deleted,ins.detached)
#  實例化並進行插入操做
try:
    login=Login(id=14)  #  實例化，此處的iD必須不存在，不然會報主鍵衝突
    print  ('實例化對應狀態:',show(login))
    session.add(login)
    print  ('添加成功對應狀態:',show(login))
    session.commit()  # 提交
    print  ('提交成功對應狀態:',show(login))
    session.delete(login)  # 刪除id對應的值
    print  ('刪除對應狀態:',show(login))
    session.commit()  # 提交
    print  ('刪除提交對應狀態:',show(login))

except  Exception as  e:   # 若拋出異常，則直接回滾
    print ('異常以下：--------',e)
    session.rollback()
finally:
    pass

結果以下

7 複雜查詢語句

1基本查詢

1 簡單條件查詢語句

emps=session.query(Employess).filter(Employess.emp_no > 10025)  # 工號大於10025 的顯示

2 in 語句

emps=session.query(Employess).filter(Employess.emp_no.in_(emplist))

3 not in 語句

emps=session.query(Employess).filter(~Employess.emp_no.in_(emplist))

4 消費者方法

消費者方法調用後，Query對象(可迭代)就轉換成了一個容器

# 消費者方法
    emps=session.query(Employess)  # 查詢全部
    print (emps.all())  # 返回列表
    print (emps.count())  # 聚合函數查詢
    print (len(list(emps))) # 返回列表元素數量，及返回值數量
        emps=session.query(Employee).filter(Employee.emp_no==10025)

5 LIKE

emps=session.query(Employess).filter(Employess.last_name.like('P%'))

6 上述結果以下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String,Enum,DATE
# 導入會話模塊
from sqlalchemy.orm import sessionmaker
# 導入枚舉類型
import enum
#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 導入基類
Base=declarative_base()
# 建立要查詢的表對應的實體類
#表字段以下
'''
CREATE TABLE `employees` (
  `emp_no` int(11) NOT NULL COMMENT '主鍵',
  `birth_date` date NOT NULL COMMENT '生日',
  `first_name` varchar(14) NOT NULL COMMENT '用戶-姓',
  `last_name` varchar(16) NOT NULL COMMENT '用戶-名',
  `gender` enum('M','F') NOT NULL COMMENT '性別',
  `hire_date` date NOT NULL COMMENT '入職時間',
  PRIMARY KEY (`emp_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
'''

# 建立枚舉類
class  MyEnum(enum.Enum):
    M='M'
    F='F'

# 建立對應實體類
class  Employess(Base):
    __tablename__='employees'
    emp_no=Column(Integer,nullable=False,primary_key=True)  # 整形，非空，主鍵
    birth_date=Column(DATE,nullable=False)  # 日期類型，非空
    first_name=Column(String(14),nullable=False)  # 字符串類型，長度爲14，非空
    last_name=Column(String(14),nullable=False)  # 字符串類型，長度爲16，非空
    gender=Column(Enum(MyEnum),nullable=False)  # 枚舉，非空
    hire_date=Column(DATE,nullable=False)  # 字符類型，非空
    def __repr__(self):  # 定義查詢返回
        return "< {}  emp_no:{}  name:{}>  ".format(self.__class__.__name__,self.emp_no,"{} {}".format(self.emp_no,self.last_name))
    __str__=__repr__

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)
#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()
#  實例化並進行插入操做

def  show(emps):
    for x  in emps:
        print (x)
    print ('------------',end='\n\n')

try:
    # 簡單查詢語句

    emps=session.query(Employess).filter(Employess.emp_no > 10025)  # 工號大於10025 的顯示
    print ("簡單查詢語句")
    show(emps)
    # in 語句查詢
    emplist=[10001,10002,10025]
    emps=session.query(Employess).filter(Employess.emp_no.in_(emplist))
    print ("in 語句查詢")
    show(emps)
    # not in 語句查詢
    emps=session.query(Employess).filter(~Employess.emp_no.in_(emplist))
    print ("not in 語句查詢")
    show(emps)
    # 消費者方法
    emps=session.query(Employess)  # 查詢全部
    print ("消費者方法")
    print (emps.all())  # 返回列表
    print (emps.count())  # 聚合函數查詢
    print (len(list(emps))) # 返回列表元素數量，及返回值數量
    # like
    emps=session.query(Employess).filter(Employess.last_name.like('P%'))
    print ('Like方法')
    print (show(emps))
except  Exception as  e:   # 若拋出異常，則直接回滾
    print ('異常以下：--------',e)
finally:
    pass

結果以下

2 聚合，分組

1 聚合函數

# 導入聚合函數
from  sqlalchemy  import  func

try:
    # count命令
    print  (session.query(func.count(Employess.emp_no)).all())
    print  (session.query(func.count(Employess.emp_no)).one())  # 若使用one,則只能由一個結果，如有多個結果，則會報錯
    print  (session.query(func.count(Employess.emp_no)).scalar())  #取回one()元組返回的第一個元素
    # AVG 平均值
    print  (session.query(func.avg(Employess.emp_no)).all())
    print  (session.query(func.avg(Employess.emp_no)).scalar())

    # min 最小值
    print  (session.query(func.min(Employess.emp_no)).all())
    print  (session.query(func.min(Employess.emp_no)).scalar())
    # max 最大值
    print  (session.query(func.max(Employess.emp_no)).all())
    print  (session.query(func.max(Employess.emp_no)).scalar())
except  Exception as  e:   # 若拋出異常，則直接回滾
    print ('異常以下：--------',e)
finally:
    pass

結果以下

2 分組

print (session.query(Employess.gender,func.count(Employess.emp_no)).group_by(Employess.gender).all())

結果以下

3 邏輯與或非

# 導入與或非模塊
from  sqlalchemy  import  or_,and_,not_

    # AND 第一個和第二個都要成立
    # 方式一and
    emps=session.query(Employess).filter(Employess.emp_no>10015).filter(Employess.emp_no<10018)
    show(emps)
    # 方式二 and
    emps=session.query(Employess).filter(and_(Employess.emp_no>10015,Employess.emp_no<10018))
    show(emps)
    # 方式三 &
    emps=session.query(Employess).filter((Employess.emp_no>10015)&(Employess.emp_no<10018))
    show(emps)

    # OR 或
    # 方式一
    emps=session.query(Employess).filter((Employess.emp_no>10025)|(Employess.emp_no<10005))
    show(emps)
    # 方式二
    emps=session.query(Employess).filter(or_(Employess.emp_no>10025),(Employess.emp_no>10005))
    show(emps)

    #NOT
    # 方式一
    emps=session.query(Employess).filter(not_(Employess.emp_no>10002))
    show(emps)

    #方式二
    emps=session.query(Employess).filter(~(Employess.emp_no>10002))
    show(emps)

4 排序和分頁

# 排序，默認是升序排列
    # 升序排列
    emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.emp_no)
    print ("升序排列")
    show(emps)
    emps = session.query(Employess).filter(Employess.emp_no > 10025).order_by(Employess.emp_no.asc())
    show(emps)
    # 降序排列
    emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.emp_no.desc())
    print ("降序排列")
    show(emps)

    # 多列排序,first_name 爲降序排列,emp_no爲升序排列，默認只有兩個first_name相同時，後面的纔有意義
    emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.first_name.desc()).order_by(Employess.emp_no)
    print  ("多列排序")
    show(emps)
    # 分頁
    emps=session.query(Employess).limit(4)
    print  ("分頁")
    show(emps)
    emps=session.query(Employess).limit(4).offset(20)
    show(emps)

結果以下

5 查詢刪除

emps=session.query(Employess.emp_no).filter(Employess.emp_no>10025).delete()
    print ('查詢結果爲:',emps)
    # session.commit()

結果以下

查詢到的東西就是想要的，直接刪除，此處對其進行了優化，直接使用where語句進行處理，而沒有通過select處理

6 關聯查詢

1 需求

查詢10010 員工所在的部門編號
使用隱式內鏈接

# 通常鏈接方式處理以下
    results=session.query(Employess,Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    show(results)
    # 隱式內鏈接第一種寫法
    results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==10010).all()
    show(results)
    # 隱式內鏈接第二種寫法
    results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    print (results)

第一種寫法最終翻譯結果以下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date, dept_emp.emp_no AS dept_emp_emp_no, dept_emp.dept_no AS dept_emp_dept_no, dept_emp.from_date AS dept_emp_from_date, dept_emp.to_date AS dept_emp_to_date 
FROM employees, dept_emp 
WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s

第二種寫法最終翻譯結果以下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date 
FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no 
WHERE employees.emp_no = %(emp_no_1)s

第三種寫法最終翻譯結果以下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date 
FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no 
WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s

第四種寫法最終翻譯結果以下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date 
FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s
2019-09-21 15:20:08,776 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 10010}

顯示以下

這種寫法，都返回一行數據，緣由是query(Employee)這隻能返回一個實體對象中去，爲了解決這個問題，須要修改實體類Employee，增長屬性用來存放部門信息
sqlalchemy.orm.relationship(實體類名字符串)

2 使用relationship來顯示其餘表的信息

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 導入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導入實體類相關屬性模塊
from sqlalchemy import Column, Integer, String,Enum,DATE,ForeignKey
# 導入會話模塊
from sqlalchemy.orm import sessionmaker,relationship
# 導入枚舉類型
import enum
#建立存儲引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 導入基類
Base=declarative_base()
# 建立要查詢的表對應的實體類
class Dept_emp(Base):
    __tablename__='dept_emp'
    emp_no=Column(Integer,ForeignKey('employees.emp_no',ondelete='CASAED'),primary_key=True)  # 整形，非空，主鍵,ForeignKey用於定義外鍵，
    dept_no=Column(String(4),ForeignKey('departments.dept_no',ondelete='CASCAED'),primary_key=True)
    from_date=Column(DATE,nullable=False)
    to_date=Column(DATE,nullable=False)
    def __repr__(self):
        return "< {}  empno={}  dept_no={}>".format(self.__class__.__name__,self.emp_no,self.dept_no)
# 建立枚舉類
class  MyEnum(enum.Enum):
    M='M'
    F='F'
# 建立對應實體類
class  Employess(Base):
    __tablename__='employees'
    emp_no=Column(Integer,nullable=False,primary_key=True)  # 整形，非空，主鍵
    birth_date=Column(DATE,nullable=False)  # 日期類型，非空
    first_name=Column(String(14),nullable=False)  # 字符串類型，長度爲14，非空
    last_name=Column(String(14),nullable=False)  # 字符串類型，長度爲16，非空
    gender=Column(Enum(MyEnum),nullable=False)  # 枚舉，非空
    hire_date=Column(DATE,nullable=False)  # 字符類型，非空
    # 增長屬性
    depts=relationship('Dept_emp')  # 後面接一個類名，這幾個的外鍵關係已經被記錄了，在dept_mp中已經表達過了，生成語句的投影和employee有關， 此處生成一個屬性來存儲部門之間的關係，若是相反，則沒法記錄上面的不存在的名字，是依靠Colum中的外鍵中進行定義處理這個東西的
    def __repr__(self):  # 定義查詢返回
        return "< {}  emp_no:{}  name:{}  depts:{}>  ".format(self.__class__.__name__,self.emp_no,"{} {}".format(self.first_name,self.last_name),self.depts)
    __str__=__repr__

class  Department(Base):
    __tablename__='departments'
    dept_no=Column(String(32),primary_key=True)
    dept_name=Column(String(40),nullable=False,unique=True)
    def __repr__(self):
        return "< {}  no={}  name={}>".format(self.__class__.__name__,self.dept_no,self.dept_name)

    __str__=__repr__

#將對應實體類加入存儲引擎。執行相關指令
Base.metadata.create_all(engine)
#建立會話
Session = sessionmaker(bind=engine)  #此處返回的是一個類，須要對其進行實例化相關操做
#對返回的類進行實例化
# session對象線程不安全，因此不一樣線程使用不能的session對象，。Session和engine都是線程安全的，有一個就好了
session=Session()
#  實例化並進行插入操做

def  show(emps):
    for x  in emps:
        print (x)
    print ('------------',end='\n\n')

try:
    # 通常鏈接方式處理以下
    results=session.query(Employess,Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    show(results)
    # 隱式內鏈接第一種寫法，此處的query 寫誰，最終其顯示結果就是誰的表的列處理 results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==10010).all()
    show(results)
    # 隱式內鏈接第二種寫法
    results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    show(results)
    # 隱式鏈接第三種想法
    results=session.query(Employess).join(Dept_emp,(Employess.emp_no==Dept_emp.emp_no)&(Employess.emp_no==10010)).all()
    show(results)
except  Exception as  e:   # 若拋出異常，則直接回滾
    print ('異常以下：--------',e)
finally:
    pass

結果以下

3 總結 :

第二種寫法中join(Dept_empo)中沒有等值條件，會自動生成等值條件，若是後面有filter，哪怕是filter(Employess.emp_no==Dept_emp.emp_no),這個條件會在where中出現，第二種自動增長的join等值條件的方式很差，不建議這樣寫

第三種在join中增長等值條件，組織了自動的等值條件生成，這種方式是推薦的

第四種方式和第三種是相同的，也可使用