• SqlAlchemy使用教程(一) 原理與環(huán)境搭建
• SqlAlchemy使用教程(三) CoreAPI訪問與操作數(shù)據(jù)庫詳解

二、入門示例與基本編程步驟

在第一章中提到，Sqlalchemy提供了兩套方法來訪問數(shù)據(jù)庫，由于Sqlalchemy 官方文檔結(jié)構(gòu)有些亂，對于ORM的使用步驟的描述散布于各個章節(jié)，再加上SqlAlchemy2.x 與j1.x版本差異較大，很多介紹SqlAlchemy的文章上來就講ORM，但示例時又使用CoreAPI，常令初學者困惑。本人建議先使用Core API來訪問數(shù)據(jù)庫，使用上更接近于 Sqlite3, Mysql-connector 等的方式，入門容易，而且也可以實現(xiàn)1套代碼支持各類數(shù)據(jù)庫。等熟悉CoreAPI的 MetaData， Sql Express Language以后，再學習使用ORM就會容易很多。
因此，本教程開頭3章均以Core API方式為主, 文中示例均在Python3.10 + SQLAlchemy 2.0.23 版本上通過測試。

1、DB API訪問數(shù)據(jù)庫入門示例

示例功能：

• 建立數(shù)據(jù)庫連接
• 通過Core API訪問數(shù)據(jù)庫（創(chuàng)建表，插入數(shù)據(jù)，查詢數(shù)據(jù)）

1.1建立數(shù)據(jù)庫連接

Step-1: 創(chuàng)建數(shù)據(jù)庫引擎對象
DB Engine 是個全局變量，允許在其上建立多個connection訪問數(shù)據(jù)庫。

創(chuàng)建 DB Engine 實例的方法：
create_engine( db_url )
db_url參數(shù)在后面章節(jié)中詳解介紹。本例使用sqlite3 內(nèi)存數(shù)據(jù)庫。

from sqlalchemy import create_engine
engine = create_engine("sqlite:///:memory:", echo=True)

Step-2 創(chuàng)建connect對象
connection 對象用于數(shù)據(jù)庫操作。其支持context with語法

from sqlalchemy import text
with engine.connect() as conn:result = conn.execute(text("select 'hello world'"))print(result.all())

output

[('hello world',)]

Step-3 執(zhí)行SQL Express 語句
text() 是SQL express 的最簡單使用形式, 方便傳值

創(chuàng)建1張表

conn.execute(text("CREATE TABLE some_table (x int, y int)"))

插入數(shù)據(jù),

conn.execute(text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),[ { "x": 1, "y": 1}, {"x": 2, "y": 4 } ],)

SQL express傳參語法：

• 參數(shù)占位使用　:x, :y ， 參數(shù)名前加:分號
• 實際值用 [ dict, … ] 方式給出。

提交事務(wù), 即將操作保存至數(shù)據(jù)庫

conn.commit()

Step4 執(zhí)行查詢并獲取結(jié)果

with engine.connect() as conn:result = conn.execute(text("SELECT x, y FROM some_table"))for row in result:print(f"x: {row.x}  y: {row.y}")

本例中，select x,y from some_table 將返回所有行

返回結(jié)果類型為 sqlalchemy.engine.cursor.CursorResult，是1個由 object 組成的可迭代對象。提供了多種方法訪問結(jié)果數(shù)據(jù)：

• fetchall(), fetchone(), fetchmany() ，使用tuple方式讀取全部、單條、多條數(shù)據(jù)
• all() 獲取所有數(shù)據(jù)，返回列表
• mappings(), 返回列表，元素為dict類型，
• keys() 獲取對象屬性名（字段名）

還可以向查詢語句傳參：

result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :y"), {"y": 2})

2、SqlAlchemy 異常處理

編寫代碼時1個好習慣：先寫出異常與錯誤處理語句框架，再寫正常流程部分，這樣的習慣可以讓代碼更健壯，避免程序運行中斷或出錯。

雖然看似麻煩，但最終代碼測試中遇到的問題更少，而且錯誤日志也更精準，問題定位效率更高，所以這樣做將更省時間。

2.1 異常處理代碼結(jié)構(gòu)建議

from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError, ProgrammingErrorengine = create_engine('mysql://username:password@localhost/mydatabase')try:# 執(zhí)行數(shù)據(jù)庫操作connection = engine.connect()# ...# 這里是可能引發(fā)異常的代碼# ...connection.close()
except IntegrityError as e:# 處理唯一性約束違反等完整性錯誤print(f'Integrity error occurred: {str(e)}')
except ProgrammingError as e:# 處理SQL語法或參數(shù)錯誤print(f'Programming error occurred: {str(e)}')
except SQLAlchemyError as e:# 處理其他SQLAlchemy異常print(f'An error occurred: {str(e)}')

2.2 SQLAlchemy常用的內(nèi)置異常類

• sqlalchemy.exc.SQLAlchemyError：所有SQLAlchemy異常的基類
• sqlalchemy.exc.InvalidRequestError：無效的請求異常，包括無效的查詢或表達式
• sqlalchemy.exc.StatementError：執(zhí)行SQL語句時出錯的異常
• sqlalchemy.exc.IntegrityError：完整性約束錯誤，例如唯一性約束或外鍵約束違反等
• sqlalchemy.exc.OperationalError：操作數(shù)據(jù)庫時出錯的異常
• sqlalchemy.exc.ProgrammingError：編程錯誤，例如錯誤的SQL語法或參數(shù)錯誤等