ORM操作在實際項目中的應用非常多，涉及到的框架也是根據不同的項目有不同的處理模塊，不過操作流程和步驟都是大同小異基本沒有什么太大變化，唯一需要注意的就是在實際操作過程中你要使用的ORM框架的處理性能和是否支持事務、是否支持分布式等特性來進行確定使用哪個ORM框架進行操作，一般在python程序中ORM操作都是對mysqldb和pymysql這樣的底層模塊進行的封裝處理。例如文章中要講解的sqlalchemy就是底層封裝mysqldb的實現，不過我們的在使用過程中需要使用pymysql進行替代。
1、安裝
$ pip install sqlalchemy
安裝完成之后，可以通過引入sqlalchemy進行版本查看，確認sqlalchemy是否安裝成功
2、連接引擎
使用sqlalchemy進行數據庫操作，首先我們需要建立一個指定數據庫的連接引擎對象。建立引擎對象的方式被封裝在了sqlalchemy.create_engine函數中，通過指定的數據庫連接信息就可以進行創建。
創建數據庫連接引擎時參數設置語法：
# 引入建立引擎的模塊 from sqlalchemy import create_engine # pymsql創建引擎連接 # mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] engine = create_engine("mysql+pymysql://root:123456@192.168.3.109:3306/test", encoding="utf-8", echo=True)
指定的數據庫連接字符串表示了目標數據庫的配置信息；encoding配置參數指定了和和數據庫之間交換的數據的編碼方式，同時echo參數表示隨時在控制台展示和數據庫之間交互的各種信息。
create_engine()函數返回的是sqlalchemy最核心的接口之一，該引擎對象會根據開發人員指定的數據庫進行對應的sql api的調用處理。
3、連接會話
創建了數據庫連接引擎對象之后，我們需要獲取和指定數據庫之間的連接，通過連接進行數據庫中數據的增刪改查操作，和數據庫的連接我們稱之為和指定數據庫之間的會話。
我們使用 sessionmaker 方法創建session。
# 引入創建session連接會話的模塊 from sqlalchemy.orm import sessionmaker # 創建一個連接會話對象；需要指定是和哪個數據庫引擎之間的會話 Session = sessionmaker(bind=engine) session = Session() # 如果在創建會話的時候沒有指定數據庫引擎，可以通過如下的方式完成引擎綁定操作 # Session = sessionmaker() # Session.configure(bind=engine) # session = Session()
此時的session不是線程安全的，並且我們一般session對象都是全局的，那么在多線程情況下，當多個線程共享一個session時，數據處理就會發生錯誤。為了保證線程安全，需使用scoped_session方法。
# 引入創建session連接會話的模塊 from sqlalchemy.orm import sessionmaker, scoped_session # 創建一個連接會話對象；需要指定是和哪個數據庫引擎之間的會話 Session = scoped_session(sessionmaker(bind=engine)) session = Session()
4、ORM之Object操作
我們的程序中的對象要使用sqlalchemy的管理，實現對象的orm操作，就需要按照框架指定的方式進行類型的創建操作，sqlalchemy封裝了基礎類的聲明操作和字段屬性的定義限制方式，開發人員要做的事情就是引入需要的模塊並在創建對象的時候使用它們即可
基礎類封裝在sqlalchemy.ext.declarative.declarative_base模塊中
字段屬性的定義封裝在sqlalchemy模塊中，通過sqlalchemy.Column定義屬性，通過封裝的Integer、String、Float等定義屬性的限制
常用的SQLAlchemy字段類型
| 類型名 | python中類型 | 說明 |
|---|---|---|
| Integer | int | 普通整數，一般是32位 |
| SmallInteger | int | 取值范圍小的整數，一般是16位 |
| BigInteger | int或long | 不限制精度的整數 |
| Float | float | 浮點數 |
| Numeric | decimal.Decimal | 普通整數，一般是32位 |
| String | str | 變長字符串 |
| Text | str | 變長字符串，對較長或不限長度的字符串做了優化 |
| Unicode | unicode | 變長Unicode字符串 |
| UnicodeText | unicode | 變長Unicode字符串，對較長或不限長度的字符串做了優化 |
| Boolean | bool | 布爾值 |
| Date | datetime.date | 時間 |
| Time | datetime.datetime | 日期和時間 |
| LargeBinary | str | 二進制文件 |
常用的SQLAlchemy列選項
| 選項名 | 說明 |
|---|---|
| primary_key | 如果為True，代表表的主鍵 |
| unique | 如果為True，代表這列不允許出現重復的值 |
| index | 如果為True，為這列創建索引，提高查詢效率 |
| nullable | 如果為True，允許有空值，如果為False，不允許有空值 |
| default | 為這列定義默認值 |
常用的SQLAlchemy關系選項
| 選項名 | 說明 |
|---|---|
| backref | 在關系的另一模型中添加反向引用 |
| primary join | 明確指定兩個模型之間使用的聯結條件 |
| uselist | 如果為False，不使用列表，而使用標量值 |
| order_by | 指定關系中記錄的排序方式 |
| secondary | 指定多對多中記錄的排序方式 |
| secondary join | 在SQLAlchemy中無法自行決定時，指定多對多關系中的二級聯結條件 |
4.1 創建基礎類
# 引入需要的模塊 from sqlalchemy.ext.declarative import declarative_base # 創建基礎類 BaseModel = declarative_base()
4.2 創建模型類
定義模型類必須繼承自之前創建的基礎類BaseModel，同時通過指定__tablename__確定和數據庫中某個數據表之間的關聯關系，指定某列類型為primary_key設定的主鍵，其他就是通過Column指定的自定義屬性了。
sqlalchemy會根據指定的__tablename__和對應的Column列字段構建自己的accessors訪問器對象，這個過程可以稱為instrumentation，經過instrumentation映射的類型就可以進行數據庫中數據的操作了。
# 引入需要的模塊 from sqlalchemy import Column, String, Integer, ForeignKey # 創建模型類 class User(BaseModel): # 定義表名 __tabelname__ = "tbl_user" # 創建字段類型 id = Column(Integer, primary_key=True) name = Column(String(50)) age = Column(Integer) class Author(BaseModel): __tablename__ = "tbl_author" id = Column(Integer, primary_key=True) name = Column(String(50)) class Book(BaseModel): __tablename__ = "tbl_book" id = Column(Integer, primary_key=True) book_name = Column(String(50)) # ForeignKey 定義外鍵，需要導入 author_id = Column(Integer, ForeignKey("tbl_author.id"))
4.3 模型類映射操作
完成了模型類的定義之后，Declarative會通過python的metaclass對模型類進行操作，根據定義的模型類創建table對象，構建程序中類型和數據庫table對象之間的映射mapping關系。
通過類型對象的metadata可以實現和數據庫之間的交互，有需要時可以通過metadata發起create table操作，通過Base.metadata.create_all()進行操作，該操作會檢查目標數據庫中是否有需要創建的表，不存在的情況下創建對應的表。
# 刪除所有模型類創建的表格（慎用） BaseModel.metadata.drop_all(engine) # 創建所有模型類的表格，有則不創建 BaseModel.metadata.create_all(engine)
5、數據的增、查、改、刪
5.1 增
在程序代碼中根據定義的數據類型創建對象的方式比較簡單，創建對象並通過會話對象將數據對象持久化到數據庫中。
# 增 user1 = User(name="張三", age=18) user2 = User(name="李四", age=19) user3 = User(name="王五", age=20) # 將單個對象加入到會話對象中 session.add(user1) # 將多個對象加入到會話對象中 session.add_all([user2, user3]) # 提交數據 session.commit()
5.2 查
Session是sqlalchemy和數據庫交互的橋梁，Session提供了一個Query對象實現數據庫中數據的查詢操作
5.2.1 基礎查詢
查詢所有數據
# 返回User對象所有字段的所有信息，返回結果為對象列表 session.query(User)
排序查詢
# 指定排序查詢 session.query(User).order_by(User.id) # 升序查詢 session.query(User).order_by(-User.id) # 降序查詢 session.query(User).order_by(-User.id, User.name) # 多條件排序查詢
查詢指定列
# 返回User對象字段為id和name的所有信息，返回結果為元組列表 session.query(User.id, User.name)
為字段取別名
# 為字段取別名 session.query(User.id, User.name.label("n")) # 對應SQL語句：SELECT tbl_user.id AS tbl_user_id, tbl_user.name AS n FROM tbl_user
分頁查詢
# 使用limit()+offset()切片分頁 # limit()限制返回條數 # offset()偏移 ret = session.query(User).limit(3).offset(2).all() # 返回第2條以后的3條數據，即第3至第5條數據 print(ret) # 使用slice()切片分頁 ret = session.query(User).slice(2, 5).all() # 返回第3至第5條數據 print(ret) # 使用python列表切片分頁 ret = session.query(User).all()[2:5] # 返回第3至第5條數據 print(ret)
分組查詢
# 分組查詢 session.query(User.name).group_by(User.name)
聚合函數
# 查詢常用聚合函數 from sqlalchemy.sql import func session.query(func.max(User.age), # 最大 func.min(User.age), # 最小 func.sum(User.age), # 求和 func.avg(User.age), # 平均 func.count(User.age) # 統計 )
去重
# 去重查詢 from sqlalchemy import distinct session.query(distinct(User.name))
5.2.2 條件篩選
在實際使用過程中經常用到條件查詢，主要通過filter和filter_by進行操作，重點講解使用最為頻繁的filter條件篩選函數
等值條件_equals / not equals
# 等於 session.query(User).filter(User.name == "李四") # 不等於 session.query(User).filter(User.name != "李四")
模糊條件_like/notlike
# 用戶名包含“張” session.query(User).filter(User.name.like("%張%")) # 用戶名不包含“張” session.query(User).filter(User.name.notlike('%張%'))
范圍條件_in / not in
# in session.query(User).filter(User.id.in_([1, 2, 4])) # not in session.query(User).filter(~User.id.in_([1, 2, 4]))
空值條件
# 空值 session.query(User).filter(User.name == None) session.query(User).filter(User.name.is_(None)) # 非空 session.query(User).filter(User.name != None) session.query(User).filter(User.name.isnot(None))
並且條件
from sqlalchemy import and_ session.query(User).filter(User.name == '張三').filter(User.age == 18) session.query(User).filter(User.name == '張三', User.age == 18) session.query(User).filter(and_(User.name == '張三', User.age == 18))
或者條件
from sqlalchemy import or_ session.query(User).filter(or_(User.name == '張三', User.name == '李四'))
5.2.3 多表查詢
# 多表查詢 session.query(Author.name, Book.book_name).filter(Author.id == Book.author_id) # 對應SQL：SELECT tbl_author.name, tbl_book.book_name FROM tbl_author, tbl_book WHERE tbl_author.id = tbl_book.author_id # 內聯查詢join session.query(Author.name, Book.book_name).join(Author, Author.id == Book.author_id) # 對應SQL：SELECT tbl_author.name, tbl_book.book_name FROM tbl_book INNER JOIN tbl_author ON tbl_author.id = tbl_book.author_id # 左聯查詢outerjoin。沒有右聯查詢 session.query(Author.name, Book.book_name).outerjoin(Author, Author.id == Book.author_id) # 對應SQL：SELECT tbl_author.name, tbl_book.book_name FROM tbl_book LEFT OUTER JOIN tbl_author ON tbl_author.id = tbl_book.author_id # 並集查詢 author_name = session.query(Author.name) book_name = session.query(Book.book_name) author_name.union(book_name) # 對應SQL：SELECT anon_1.c_name FROM # (SELECT tbl_author.name as c_name FROM tbl_author UNION SELECT tbl_book.book_name as c_name FROM tbl_book) AS anon_1
5.2.4 獲取查詢結果
獲取查詢對象有這么幾種方法one()，all()，first()，one()，one_or_none()，scalar()，以及get()等。
下面對這幾個方法的用法及效果做簡單解釋。
all()
返回查詢到的所有的結果。
這個方法比較危險的地方是如果數據量大且沒有使用limit子句限制的話，所有的結果都會加載到內存中。
它返回的是一個列表，如果查詢不到任何結果，返回的是空列表。
ret = session.query(User).all() print(ret) # [<__main__.User object at 0x0000000003894F60>, <__main__.User object at 0x00000000038942E8>]
first()
返回查詢到的第一個結果，如果沒有查詢到結果，返回None。
ret = session.query(User).first() print(ret) # <__main__.User object at 0x0000000003894358>
one()
如果只能查詢到一個結果，返回它，否則拋出異常。
沒有結果時拋sqlalchemy.orm.exc.NoResultFound，有超過一個結果時拋sqlalchemy.orm.exc.MultipleResultsFound。
ret = session.query(User).filter(User.id == 2).one() print(ret) # <__main__.User object at 0x00000000038954A8>
one_or_none()
比起one()來，區別只是查詢不到任何結果時不再拋出異常而是返回None。
ret = session.query(User).filter(User.id == 2).one_or_none() print(ret) # <__main__.User object at 0x00000000038954A8>
scalar()
這個方法與.one_or_none()的效果一樣。
如果查詢到很多結果，拋出sqlalchemy.orm.exc.MultipleResultsFound異常。
如果只有一個結果，返回它，沒有結果返回None。
ret = session.query(User).filter(User.id == 2).scalar() print(ret) # <__main__.User object at 0x00000000038954A8>
get()
這是個比較特殊的方法。它用於根據主鍵來返回查詢結果，因此它有個參數就是要查詢的對象的主鍵。
如果沒有該主鍵的結果返回None，否則返回這個結果。
ret = session.query(User).get(2) print(ret) # <__main__.User object at 0x00000000038954A8>
當獲取的結果為一個對象時，可以通過 對象.模型類屬性 的方式取出值
user = session.query(User).get(2) print(user) # <__main__.User object at 0x00000000038954A8> print(user.name) # 李四
5.3 改
# update()更新數據 session.query(User).filter(User.id == 2).update({"name": "李四改"}) session.query(User).filter(User.id >= 4).update({"age": 88}) session.commit()
5.4 刪
# delete()刪除數據 session.query(User).filter(User.id == 2).delete() session.query(User).filter(User.id >= 4).delete() session.commit()
6、執行原生SQL語句
使用原生sql語句需要引入text方法，傳入要執行的sql語句，若需要給sql語句傳參，需要使用 :param 的方式定義參數名，然后使用 params() 並以鍵值對的方式傳入值
from sqlalchemy import text session.query(User).from_statement(text('select * from tbl_user where name=:name and age=:age')).params(name='李四', age=19) session.query(User).from_statement(text('update tbl_user set name=:name where id=:id')).params(name='趙六', id=1) session.commit()
還有一種類似於pymysql使用 execute() 執行原生sql的方法
ret = session.execute('select * from tbl_user where name=:name', {"name": "趙六"}).fetchall() # 所有數據以元組的形式保存在列表中 print(ret) # [(1, '趙六', 18, 2), (4, '趙六', 18, 2)] ret = session.execute('select * from tbl_user where name=:name', {"name": "趙六"}).fetchone() # 返回第一條數據，無則為None print(ret) # (1, '趙六', 18, 2)
附碼：
# 引入建立引擎的模塊 from sqlalchemy import create_engine # 引入創建session連接會話的模塊 from sqlalchemy.orm import sessionmaker, scoped_session # 模型類需要使用的模塊 from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, String, Integer, ForeignKey # 數據操作的模塊 from sqlalchemy import distinct, and_, or_ # 執行原生SQL的模塊 from sqlalchemy import text # 1、pymsql創建引擎連接 # mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] engine = create_engine("mysql+pymysql://root:123456@192.168.3.109:3306/test", encoding="utf-8", echo=True) # 2、創建一個連接會話對象；需要指定是和哪個數據庫引擎之間的會話 Session = sessionmaker(bind=engine) session = Session() # 如果在創建會話的時候沒有指定數據庫引擎，可以通過如下的方式完成引擎綁定操作 # Session = sessionmaker() # Session.configure(bind=engine) # session = Session() # 安全session # Session = scoped_session(sessionmaker(bind=engine)) # session = Session() # 3、模型類 # 創建基礎類 BaseModel = declarative_base() # 創建模型類 class User(BaseModel): # 定義表名 __tabelname__ = "tbl_user" # 創建字段類型 id = Column(Integer, primary_key=True) name = Column(String(50)) age = Column(Integer) class Author(BaseModel): __tablename__ = "tbl_author" id = Column(Integer, primary_key=True) name = Column(String(50)) class Book(BaseModel): __tablename__ = "tbl_book" id = Column(Integer, primary_key=True) book_name = Column(String(50)) # ForeignKey 定義外鍵，需要導入 author_id = Column(Integer, ForeignKey("tbl_author.id")) # 刪除所有模型類創建的表格（慎用） # BaseModel.metadata.drop_all(engine) # 創建所有模型類的表格，有則不創建 BaseModel.metadata.create_all(engine) # 4、數據操作 # 增 user1 = User(name="張三", age=18) user2 = User(name="李四", age=19) user3 = User(name="王五", age=20) # 將單個對象加入到會話對象中 session.add(user1) # 將多個對象加入到會話對象中 session.add_all([user2, user3]) # 提交數據 session.commit() # 返回User對象所有字段的所有信息，返回結果為對象列表 session.query(User) # 指定排序查詢 session.query(User).order_by(User.id) # 升序查詢 session.query(User).order_by(-User.id) # 降序查詢 session.query(User).order_by(-User.id, User.name) # 多條件排序查詢 # 返回User對象字段為id和name的所有信息，返回結果為元組列表 session.query(User.id, User.name) # 為字段取別名 session.query(User.id, User.name.label("n")) # 對應SQL語句：SELECT tbl_user.id AS tbl_user_id, tbl_user.name AS n FROM tbl_user # 分組查詢 session.query(User.name).group_by(User.name) # 查詢常用聚合函數 from sqlalchemy.sql import func session.query(func.max(User.age), # 最大 func.min(User.age), # 最小 func.sum(User.age), # 求和 func.avg(User.age), # 平均 func.count(User.age) # 統計 ) # 去重查詢 session.query(distinct(User.name)) # 使用limit()+offset()切片分頁 # limit()限制返回條數 # offset()偏移 ret = session.query(User).limit(3).offset(2).all() # 返回第2條以后的3條數據，即第3至第5條數據 print(ret) # 使用slice()切片分頁 ret = session.query(User).slice(2, 5).all() # 返回第3至第5條數據 print(ret) # 使用python列表切片分頁 ret = session.query(User).all()[2:5] # 返回第3至第5條數據 print(ret) # 等於 session.query(User).filter(User.name == "李四") # 不等於 session.query(User).filter(User.name != "李四") # 用戶名包含“張” session.query(User).filter(User.name.like("%張%")) # 用戶名不包含“張” session.query(User).filter(User.name.notlike('%張%')) # in session.query(User).filter(User.id.in_([1, 2, 4])) # not in session.query(User).filter(~User.id.in_([1, 2, 4])) # 空值 session.query(User).filter(User.name == None) session.query(User).filter(User.name.is_(None)) # 非空 session.query(User).filter(User.name != None) session.query(User).filter(User.name.isnot(None)) # 並且 session.query(User).filter(User.name == '張三').filter(User.age == 18) session.query(User).filter(User.name == '張三', User.age == 18) session.query(User).filter(and_(User.name == '張三', User.age == 18)) # 或者 session.query(User).filter(or_(User.name == '張三', User.name == '李四')) # 多表查詢 session.query(Author.name, Book.book_name).filter(Author.id == Book.author_id) # 對應SQL：SELECT tbl_author.name, tbl_book.book_name FROM tbl_author, tbl_book WHERE tbl_author.id = tbl_book.author_id # 內聯查詢join session.query(Author.name, Book.book_name).join(Author, Author.id == Book.author_id) # 對應SQL：SELECT tbl_author.name, tbl_book.book_name FROM tbl_book INNER JOIN tbl_author ON tbl_author.id = tbl_book.author_id # 左聯查詢outerjoin。沒有右聯查詢 session.query(Author.name, Book.book_name).outerjoin(Author, Author.id == Book.author_id) # 對應SQL：SELECT tbl_author.name, tbl_book.book_name FROM tbl_book LEFT OUTER JOIN tbl_author ON tbl_author.id = tbl_book.author_id # 並集查詢 author_name = session.query(Author.name) book_name = session.query(Book.book_name) author_name.union(book_name) # 對應SQL：SELECT anon_1.c_name FROM # (SELECT tbl_author.name as c_name FROM tbl_author UNION SELECT tbl_book.book_name as c_name FROM tbl_book) AS anon_1 # all()返回所有的數據 session.query(User).all() # first()返回查詢到的第一個結果，如果沒有查詢到結果，返回None。 session.query(User).first() # one()如果只能查詢到一個結果，返回它，否則拋出異常。 session.query(User).filter(User.id == 2).one() # one_or_none()比起one()來，區別只是查詢不到任何結果時不再拋出異常而是返回None。 session.query(User).filter(User.id == 2).one_or_none() # scalar()這個方法與.one_or_none()的效果一樣。 session.query(User).filter(User.id == 2).scalar() # get()這是個比較特殊的方法。它用於根據主鍵來返回查詢結果，因此它有個參數就是要查詢的對象的主鍵。如果沒有該主鍵的結果返回None，否則返回這個結果。 session.query(User).get(2) # update()更新數據 session.query(User).filter(User.id == 2).update({"name": "李四改"}) session.query(User).filter(User.id >= 4).update({"age": 88}) session.commit() # delete()刪除數據 session.query(User).filter(User.id == 2).delete() session.query(User).filter(User.id >= 4).delete() session.commit() # 執行原生SQL session.query(User).from_statement(text('select * from tbl_user where name=:name and age=:age')).params(name='李四', age=19) session.query(User).from_statement(text('update tbl_user set name=:name where id=:id')).params(name='趙六', id=1) session.commit()
原文參考：https://blog.csdn.net/u012089823/article/details/94650310