SQLAlchemy 學習筆記（一）：Engine 與 SQL 表示式語言

摘要： 個人筆記，如有錯誤煩請指正。 SQLAlchemy 是一個用 Python 實現的 ORM （Object Relational Mapping）框架，它由多個元件構成，這些元件可以單獨使用，也能獨立使用。它的元件層次結構如下： 其中最常用的元件，應該是 ORM 和 SQL...

個人筆記，如有錯誤煩請指正。

SQLAlchemy 是一個用 Python 實現的 ORM （Object Relational Mapping）框架，它由多個元件構成，這些元件可以單獨使用，也能獨立使用。它的元件層次結構如下：

其中最常用的元件，應該是 ORM 和 SQL 表示式語言 ，這兩者既可以獨立使用，也能結合使用。

ORM的好處在於它

1. 自動處理了資料庫和 Python 物件之間的對映關係，遮蔽了兩套系統之間的差異。程式設計師不需要再編寫複雜的 SQL 語句，直接操作 Python 物件就行。
2. 遮蔽了各資料庫之間的差異，更換底層資料庫不需要修改 SQL 語句，改下配置就行。
3. 使資料庫結構文件化，models 定義很清晰地描述了資料庫的結構。
4. 避免了不規範、冗餘、風格不統一的 SQL 語句，可以避免很多人為 Bug，也方便維護。

但是 ORM 需要消耗額外的效能來處理物件關係對映，此外用 ORM 做多表關聯查詢或複雜 SQL 查詢時，效率低下。因此它適用於場景不太複雜，效能要求不太苛刻的場景。

都說 ORM 學習成本高，我自己也更傾向於直接使用 SQL 語句（畢竟更熟悉），因此這一篇筆記不涉及 ORM 部分，只記錄 SQLAlchemy 的 Engine 與 SQL 表示式語言。

一、直接使用 Engine 和 Connections

第一步是建立資料庫引擎例項：

from sqlalchemy import create_engine

engine = create_engine('sqlite:///:memory:', 
echo=True,# echo=True 表示打印出自動生成的 SQL 語句（通過 logging）
pool_size=5,# 連線池容量，預設為 5，生產環境下太小，需要修改。
# 下面是 connection 回收的時間限制，預設 -1 不回收
pool_recyle=7200)# 超過 2 小時就重新連線（MySQL 預設的連線最大閒置時間為 8 小時）

create_engine 接受的第一個引數是資料庫 URI，格式為 dialect[+driver]://user:password@host/dbname[?key=value..] ，dialect 是具體的資料庫名稱，driver 是驅動名稱。key-value 是可選的引數。舉例：

# PostgreSQL
postgresql+psycopg2://scott:tiger@localhost/dbtest

# MySQL + PyMySQL（或者用更快的mysqlclient）
mysql+pymysql://scott:tiger@localhost/dbtest

# sqlite 記憶體資料庫
# 注意 sqlite 要用三個斜槓，表示不存在 hostname，sqlite://<nohostname>/<path>
sqlite:///:memory:

# sqlite 檔案資料庫
# 四個斜槓是因為檔案的絕對路徑以 / 開頭：/home/ryan/Codes/Python/dbtest.db
sqlite:////home/ryan/Codes/Python/dbtest.db

# SQL Server + pyodbc
# 首選基於 dsn 的連線，dsn 的配置請搜尋hhh
mssql+pyodbc://scott:tiger@some_dsn

引擎建立後，我們就可以直接獲取 connection，然後執行 SQL 語句了。這種用法相當於把 SQLAlchemy 當成帶 log 的資料庫連線池使用：

with engine.connect() as conn:
res = conn.execute("select username from users")# 無參直接使用

# 使用問號作佔位符，前提是下層的 DBAPI 支援。更好的方式是使用 text()，這個後面說
conn.execute("INSERT INTO table (id, value) VALUES (?, ?)", 1, "v1")# 引數不需要包裝成元組

# 查詢返回的是 ResultProxy 物件，有和 dbapi 相同的 fetchone()、fetchall()、first() 等方法，還有一些拓展方法
for row in result:
print("username:", row['username'])

但是要注意的是，connection 的 execute 是自動提交的（autocommit），這就像在 shell 裡開啟一個數據庫客戶端一樣，分號結尾的 SQL 會被自動提交。

只有在 BEGIN TRANSACTION 內部， AUTOCOMMIT 會被臨時設定為 FALSE ，可以通過如下方法開始一個內部事務：

def transaction_a(connection):
trans = connection.begin()# 開啟一個 transaction
try:
# do sthings
trans.commit()# 這裡需要手動提交
except:
trans.rollback()# 出現異常則 rollback
raise
# do other things

with engine.connect() as conn:
transaction_a(conn)

1. 使用 text() 構建 SQL

相比直接使用 string，text() 的優勢在於它：

1. 提供了統一的引數繫結語法，與具體的 DBAPI 無關。

# 1. 引數繫結語法
from sqlalchemy import text

result = connection.execute(
# 使用 :key 做佔位符
text('select * from table where id < :id and typeName=:type'),
{'id': 2,'type':'USER_TABLE'})# 用 dict 傳引數，更易讀

# 2. 引數型別指定
from sqlalchemy import DateTime

date_param=datetime.today()+timedelta(days=-1*10)
sql="delete from caw_job_alarm_logwhere alarm_time<:alarm_time_param"

# bindparams 是 bindparam 的列表，bindparam 則提供引數的一些額外資訊（型別、值、限制等）
t=text(sql, bindparams=[bindparam('alarm_time_param', type_=DateTime, required=True)])
connection.execute(t, {"alarm_time_param": date_param})

1. 可以很方便地轉換 Result 中列的型別

stmt = text("SELECT * FROM table",
# 使用 typemap 指定將 id 列對映為 Integer 型別，name 對映為 String 型別
typemap={'id': Integer, 'name': String},
)

二、SQL 表示式語言

複雜的 SQL 查詢可以直接用 raw sql 寫，而增刪改一般都是單表操作，用 SQL 表示式語言最方便。

SQLAlchemy 表示式語言是一個使用 Python 結構表示關係資料庫結構和表示式的系統。

1. 定義並建立表

SQL 表示式語言使用 Table 來定義表，而表的列則用 Column 定義。Column 總是關聯到一個 Table 物件上。

一組 Table 物件以及它們的子物件的集合就被稱作「資料庫元資料（database metadata）」。metadata 就像你的網頁分類收藏夾，相關的 Table 放在一個 metadata 中。

下面是建立元資料（一組相關聯的表）的例子，：

from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

metadata = MetaData()# 先建立元資料（收藏夾）

users = Table('user', metadata,# 建立 user 表，並放到 metadata 中
Column('id', Integer, primary_key=True),
Column('name', String),
Column('fullname', String)
)

addresses = Table('address', metadata,
Column('id', Integer, primary_key=True),
Column('user_id', None, ForeignKey('user.id')),# 外來鍵，引用 user 表的 id 列
Column('email_address', String, nullable=False)
)

metadata.create_all(engine)# 使用 engine 建立 metadata 內的所有 Tables（會檢測表是否已經存在，所以可以重複呼叫）

2. 增刪改語句

1. 增 :

# 方法一，使用 values 傳參
ins = users.insert().values(name="Jack", fullname="Jack Jones")# 可以通過 str(ins) 檢視自動生成的 sql
connection.execute(ins)

# 方法二，引數傳遞給 execute()
conn.execute(users.insert(), id=2, name='wendy', fullname='Wendy Williams')

# 方法三，批量 INSERT，相當於 executemany
conn.execute(addresses.insert(), [# 插入到 addresses 表
{'user_id': 1, 'email_address': '[email protected]'},# 傳入 dict 列表
{'user_id': 1, 'email_address': '[email protected]'},
{'user_id': 2, 'email_address': '[email protected]'},
{'user_id': 2, 'email_address': '[email protected]'}
])

# 此外，通過使用 bindparam，INSERT 還可以執行更復雜的操作
stmt = users.insert() \
.values(name=bindparam('_name') + " .. name")# string 拼接
conn.execute(stmt, [
{'id':4, '_name':'name1'},
{'id':5, '_name':'name2'},
{'id':6, '_name':'name3'},
])

1. 刪 ：

_table.delete() \
.where(_table.c.f1==value1) \
.where(_table.c.f2==value2)# where 指定條件

1. 改 ：

# 舉例
stmt = users.update() \
.where(users.c.name == 'jack') \
.values(name='tom')
conn.execute(stmt)

# 批量更新
stmt = users.update().\
where(users.c.name == bindparam('oldname')).\
values(name=bindparam('newname'))
conn.execute(stmt, [
{'oldname':'jack', 'newname':'ed'},
{'oldname':'wendy', 'newname':'mary'},
{'oldname':'jim', 'newname':'jake'},
])

可以看到，所有的條件都是通過 where 指定的，它和後面 ORM 的 filter 接受的引數是一樣的。（詳細的會在第二篇文章裡講）

查詢暫時略過（因為感覺手寫 sql 更方便。。）

連結

• 使用 Engines 和 Connections
• SQL 表示式語言入門
• sqlAlchemy個人學習筆記完整彙總