SQLAlchemy 學習筆記（一）：Engine 與 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_recycle=7200)   # 超過 2 小時就重新連接（MySQL 默認的連接最大閑置時間為 8 小時）

create_engine 接受的第一個參數是數據庫 URI，格式為 dialect+driver://username:password@host:port/database，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

如果你的密碼中含有 '@' 等特殊字符，就不能直接放入 URI 中，必須使用 urllib.parse.quote_plus 編碼，然后再插入 URI.

引擎創建后，我們就可以直接獲取 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_log  where 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},
          )
result = connection.execute(stmt)
# 對多個查詢結果，可以用 for obj in result 遍歷
# 也可用 fetchone() 只獲取一個

二、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（會檢測表是否已經存在，所以可以重復調用）

表定義中的約束

應該給所有的約束命名，即為 name 參數指定一個不沖突的列名。詳見 The Importance of Naming Constraints

表還有一個屬性：約束條件。下面一一進行說明。

1. 外鍵約束：用於在刪除或更新某個值或行時，對主鍵/外鍵關系中一組數據列強制進行的操作限制。
   1. 用法一：Column('user_id', None, ForeignKey('user.id'))，直接在 Column 中指定。這也是最常用的方法
   2. 用法二：通過 ForeignKeyConstraint(columns, refcolumns) 構建約束，作為參數傳給 Table.

item = Table('item', metadata,  # 商品 table
    Column('id', Integer, primary_key=True),
    Column('name', String(60), nullable=False),
    Column('invoice_id', Integer, nullable=False),  # 發票 id，是外鍵
    Column('ref_num', Integer, nullable=False),
    ForeignKeyConstraint(['invoice_id', 'ref_num'],  # 當前表中的外鍵名稱
                                        ['invoice.id', 'invoice.ref_num'])  # 被引用的外鍵名稱的序列（被引用的表）
)

1. on delete 與 on update：外鍵約束的兩個約束條件，通過 ForeignKey() 或 ForeignKeyConstraint() 的關鍵字參數 ondelete/onupdate 傳入。
   可選值有：
   1. 默認行為 NO ACTION：什么都不做，直接報錯。
   2. CASCADE：刪除/更新 父表數據時，從表數據會同時被 刪除/更新。（無報錯）
   3. RESTRICT：不允許直接 刪除/更新 父表數據，直接報錯。（和默認行為基本一致）
   4. SET NULL or SET DEFAULT：刪除/更新 父表數據時，將對應的從表數據重置為 NULL 或者默認值。
2. 唯一性約束：UniqueConstraint('col2', 'col3', name='uix_1')，作為參數傳給 Table.
3. CHECK 約束：CheckConstraint('col2 > col3 + 5', name='check1')， 作為參數傳給 Table.
4. 主鍵約束：不解釋
   1. 方法一：通過 Column('id', Integer, primary_key=True) 指定主鍵。（參數 primary_key 可在多個 Column 上使用）
   2. 方法二：使用 PrimaryKeyConstraint

from sqlalchemy import PrimaryKeyConstraint

my_table = Table('mytable', metadata,
            Column('id', Integer),
            Column('version_id', Integer),
            Column('data', String(50)),
            PrimaryKeyConstraint('id', 'version_id', name='mytable_pk')
        )

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': '[e-mail]'},  # 傳入 dict 列表
    {'user_id': 1, 'email_address': '[e-mail]'},
    {'user_id': 2, 'email_address': '[e-mail]'},
    {'user_id': 2, 'email_address': '[e-mail]'}
])

# 此外，通過使用 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'},
     ])

2. 刪：

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

3. 改：

# 舉例
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 接受的參數是一樣的。（詳細的會在第二篇文章里講）

4. 查

from sqlalchemy.sql import select

# 1. 字段選擇
s1 = select([users])  # 相當於 select * from users
s2 = select([users.c.name, users.c.fullname])  # 這個就是只 select 一部分

# 2. 添加過濾條件
s3 = select([users]) \
    .where(users.c.id == addresses.c.user_id) 

res = conn.execute(s1)
# 可用 for row in res 遍歷結果集，也可用 fetchone() 只獲取一行

查詢返回的是 ResultProxy 對象，這是 SQLAlchemy 對 Python DB-API 的 cursor 的一個封裝類，要從中獲取結果行，主要有下列幾個方法：

row1 = result.fetchone()  # 對應 cursor.fetchone()
row2 = result.fetchall()  # 對應 cursor.fetchall()
row3 = result.fetchmany(size=3)  # 對應 cursor.fetchmany(size=3)

row4 = result.first()  # 獲取一行，然后立即調用 result 的 close() 方法

col = row[mytable.c.mycol] # 獲取 mycol 這一列

result.rowcount  # 結果集的行數

同時，result 也實現了 next protocol，因此可以直接用 for 循環遍歷

where 進階

通過使用 or_、and_、in_ model.join 等方法，where 可以構建更復雜的 SQL 語句。

from sqlalchemy.sql import and_, or_, not_

s = select([(users.c.fullname +
               ", " + addresses.c.email_address).
                label('title')]).\
        where(users.c.id == addresses.c.user_id).\
        where(users.c.name.between('m', 'z')).\
        where(
               or_(
                  addresses.c.email_address.like('[e-mail]'),
                  addresses.c.email_address.like('[e-mail]')
               )
        )

鏈接

• 使用 Engines 和 Connections
• SQL 表達式語言入門
• SQLAlchemy - 定義約束
• SQLAlchemy個人學習筆記完整匯總