SQLAlchemy - 快速指南

---

---

SQLAlchemy - 簡介

SQLAlchemy 是一個流行的 SQL 工具包和物件關係對映器。它是用Python編寫的，併為應用程式開發人員提供了 SQL 的全部功能和靈活性。它是一個開源且跨平臺的軟體，根據 MIT 許可證釋出。

SQLAlchemy 以其物件關係對映器 (ORM) 而聞名，使用它，類可以對映到資料庫，從而允許從一開始就以乾淨解耦的方式開發物件模型和資料庫模式。

隨著 SQL 資料庫的大小和效能開始變得重要，它們的行為越來越不像物件集合。另一方面，隨著物件集合中的抽象開始變得重要，它們的行為越來越不像表和行。SQLAlchemy 旨在兼顧這兩個原則。

出於這個原因，它採用了資料對映器模式（如 Hibernate）而不是許多其他 ORM 使用的活動記錄模式。使用 SQLAlchemy 將以不同的視角看待資料庫和 SQL。

Michael Bayer 是 SQLAlchemy 的最初作者。其初始版本於 2006 年 2 月釋出。最新版本編號為 1.2.7，最近於 2018 年 4 月釋出。

什麼是 ORM？

ORM（物件關係對映）是一種程式設計技術，用於在面向物件程式語言中轉換不相容型別系統之間的資料。通常，面向物件 (OO) 語言（如 Python）中使用的型別系統包含非標量型別。這些不能表示為整數和字串等基本型別。因此，OO 程式設計師必須將物件轉換為標量資料以與後端資料庫互動。但是，大多數資料庫產品（如 Oracle、MySQL 等）中的資料型別都是基本型別。

在 ORM 系統中，每個類都對映到底層資料庫中的一個表。ORM 代替您編寫乏味的資料庫介面程式碼，為您處理這些問題，而您可以專注於系統邏輯的程式設計。

SQLAlchemy - 環境設定

讓我們討論使用 SQLAlchemy 所需的環境設定。

需要 Python 2.7 或更高版本才能安裝 SQLAlchemy。最簡單的安裝方法是使用 Python 包管理器pip。此實用程式與 Python 的標準分發版捆綁在一起。

pip install sqlalchemy

使用上述命令，我們可以從python.org下載 SQLAlchemy 的最新發布版本並將其安裝到您的系統中。

對於 Anaconda 發行的 Python，可以使用以下命令從conda 終端安裝 SQLAlchemy：

conda install -c anaconda sqlalchemy

也可以從以下原始碼安裝 SQLAlchemy：

python setup.py install

SQLAlchemy 旨在與為特定資料庫構建的 DBAPI 實現一起使用。它使用方言系統與各種型別的 DBAPI 實現和資料庫進行通訊。所有方言都需要安裝相應的 DBAPI 驅動程式。

以下是被包含的方言：

• Firebird
• Microsoft SQL Server
• MySQL
• Oracle
• PostgreSQL
• SQLite
• Sybase

要檢查 SQLAlchemy 是否已正確安裝並瞭解其版本，請在 Python 提示符下輸入以下命令：

>>> import sqlalchemy
>>>sqlalchemy.__version__
'1.2.7'

SQLAlchemy Core – 表示式語言

SQLAlchemy core 包括SQL 渲染引擎、DBAPI 整合、事務整合和模式描述服務。SQLAlchemy core 使用 SQL 表示式語言，該語言提供了一種以模式為中心的用法範例，而 SQLAlchemy ORM 是一種以領域為中心的用法模式。

SQL 表示式語言提供了一個使用 Python 結構表示關係資料庫結構和表示式的系統。它提供了一個表示關係資料庫的基本結構的系統，而無需任何意見，這與 ORM 形成對比，ORM 提供了一種高階且抽象的用法模式，這本身就是表示式語言應用用法的示例。

表示式語言是 SQLAlchemy 的核心元件之一。它允許程式設計師在 Python 程式碼中指定 SQL 語句，並將其直接用於更復雜的查詢中。表示式語言獨立於後端，並全面涵蓋了原始 SQL 的各個方面。它比 SQLAlchemy 中的任何其他元件都更接近原始 SQL。

表示式語言直接表示關係資料庫的基本結構。由於 ORM 基於表示式語言之上，因此典型的 Python 資料庫應用程式可能同時使用兩者。應用程式可以單獨使用表示式語言，儘管它必須定義自己的系統來將應用程式概念轉換為單個數據庫查詢。

SQLAlchemy 引擎會將表示式語言的語句轉換為相應的原始 SQL 查詢。我們現在將學習如何建立引擎並藉助其幫助執行各種 SQL 查詢。

SQLAlchemy Core - 連線資料庫

在上一章中，我們討論了 SQLAlchemy 中的表示式語言。現在讓我們繼續討論連線到資料庫所涉及的步驟。

Engine 類將池和方言組合在一起，以提供資料庫連線和行為的來源。使用create_engine()函式例項化 Engine 類的物件。

create_engine() 函式將資料庫作為引數之一。資料庫不需要在任何地方定義。標準呼叫表單必須將 URL 作為第一個位置引數傳送，通常是一個字串，表示資料庫方言和連線引數。使用下面給出的程式碼，我們可以建立一個數據庫。

>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite:///college.db', echo = True)

對於MySQL 資料庫，請使用以下命令：

engine = create_engine("mysql://user:pwd@localhost/college",echo = True)

要專門提及用於連線的DB-API，URL 字串採用以下形式：

dialect[+driver]://user:password@host/dbname

例如，如果您使用PyMySQL 驅動程式與 MySQL，請使用以下命令：

mysql+pymysql://<username>:<password>@<host>/<dbname>

echo 標誌是設定 SQLAlchemy 日誌記錄的快捷方式，這是透過 Python 的標準日誌記錄模組完成的。在隨後的章節中，我們將學習所有生成的 SQL。要隱藏詳細輸出，請將 echo 屬性設定為None。create_engine() 函式的其他引數可能是特定於方言的。

create_engine() 函式返回一個Engine 物件。Engine 類的一些重要方法是：

SQLAlchemy Core - 建立表

現在讓我們討論如何使用建立表函式。

SQL 表示式語言針對表列構建其表示式。SQLAlchemy Column 物件表示資料庫表中的一列，該列又由Tableobject表示。元資料包含表和相關物件的定義，例如索引、檢視、觸發器等。

因此，來自 SQLAlchemy 元資料的 MetaData 類的物件是 Table 物件及其關聯的模式結構的集合。它儲存 Table 物件的集合以及對 Engine 或 Connection 的可選繫結。

from sqlalchemy import MetaData
meta = MetaData()

MetaData 類的建構函式可以具有 bind 和 schema 引數，它們預設為None。

接下來，我們使用Table 結構在上述元資料目錄中定義所有表，這類似於常規 SQL CREATE TABLE 語句。

Table 類的一個物件表示資料庫中相應的表。建構函式採用以下引數：

Column 物件表示資料庫表中的一列。建構函式採用名稱、型別以及其他引數，例如 primary_key、autoincrement 和其他約束。

SQLAlchemy 將 Python 資料匹配到其中定義的最佳通用列資料型別。一些通用資料型別是：

• BigInteger
• Boolean
• Date
• DateTime
• Float
• Integer
• Numeric
• SmallInteger
• String
• Text
• Time

要在 college 資料庫中建立students 表，請使用以下程式碼段：

from sqlalchemy import Table, Column, Integer, String, MetaData
meta = MetaData()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

create_all() 函式使用 engine 物件建立所有定義的表物件並將資訊儲存在元資料中。

meta.create_all(engine)

下面給出了完整的程式碼，它將在其中建立一個 SQLite 資料庫 college.db，並在其中建立一個 students 表。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String),
)
meta.create_all(engine)

因為 create_engine() 函式的 echo 屬性設定為True，所以控制檯將顯示用於建立表的實際 SQL 查詢，如下所示：

CREATE TABLE students (
   id INTEGER NOT NULL,
   name VARCHAR,
   lastname VARCHAR,
   PRIMARY KEY (id)
)

college.db 將在當前工作目錄中建立。要檢查是否已建立 students 表，您可以使用任何 SQLite GUI 工具（例如SQLiteStudio）開啟資料庫。

下圖顯示了在資料庫中建立的 students 表：

SQLAlchemy Core - SQL 表示式

在本章中，我們將簡要介紹 SQL 表示式及其功能。

SQL 表示式是使用相對於目標表物件的相應方法構建的。例如，INSERT 語句是透過執行 insert() 方法建立的，如下所示：

ins = students.insert()

上述方法的結果是一個插入物件，可以使用str()函式進行驗證。以下程式碼插入學生 ID、姓名、姓氏等詳細資訊。

'INSERT INTO students (id, name, lastname) VALUES (:id, :name, :lastname)'

可以使用values()方法將值插入特定欄位以插入物件。以下給出了相同的程式碼：

>>> ins = users.insert().values(name = 'Karan')
>>> str(ins)
'INSERT INTO users (name) VALUES (:name)'

Python 控制檯上回顯的 SQL 未顯示實際值（在本例中為“Karan”）。相反，SQLALchemy 生成一個繫結引數，該引數在語句的編譯形式中可見。

ins.compile().params
{'name': 'Karan'}

類似地，update()、delete()和select()等方法分別建立 UPDATE、DELETE 和 SELECT 表示式。我們將在後面的章節中學習它們。

SQLAlchemy Core - 執行表示式

在上一章中，我們學習了 SQL 表示式。在本章中，我們將深入研究這些表示式的執行。

為了執行生成的 SQL 表示式，我們必須獲取表示已積極簽出的 DBAPI 連線資源的連線物件，然後提供表示式物件，如下面的程式碼所示。

conn = engine.connect()

以下 insert() 物件可用於 execute() 方法 -

ins = students.insert().values(name = 'Ravi', lastname = 'Kapoor')
result = conn.execute(ins)

控制檯顯示 SQL 表示式執行結果如下 -

INSERT INTO students (name, lastname) VALUES (?, ?)
('Ravi', 'Kapoor')
COMMIT

以下是使用 SQLAlchemy 的核心技術執行 INSERT 查詢的完整程式碼片段 -

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

ins = students.insert()
ins = students.insert().values(name = 'Ravi', lastname = 'Kapoor')
conn = engine.connect()
result = conn.execute(ins)

可以透過使用 SQLite Studio 開啟資料庫來驗證結果，如下面的螢幕截圖所示 -

結果變數被稱為 ResultProxy 物件。它類似於 DBAPI 遊標物件。我們可以使用 ResultProxy.inserted_primary_key 獲取從我們的語句中生成的 primary key 值的資訊，如下所示 -

result.inserted_primary_key
[1]

要使用 DBAPI 的 execute many() 方法發出多個插入操作，我們可以傳送一個字典列表，每個字典包含一組要插入的不同引數。

conn.execute(students.insert(), [
   {'name':'Rajiv', 'lastname' : 'Khanna'},
   {'name':'Komal','lastname' : 'Bhandari'},
   {'name':'Abdul','lastname' : 'Sattar'},
   {'name':'Priya','lastname' : 'Rajhans'},
])

這反映在表的 data view 中，如下圖所示 -

SQLAlchemy Core - 選擇行

在本章中，我們將討論表物件中選擇行的概念。

表物件的 select() 方法使我們能夠構建 SELECT 表示式。

s = students.select()

select 物件透過str(s) 函式轉換為 SELECT 查詢，如下所示 -

'SELECT students.id, students.name, students.lastname FROM students'

我們可以將此 select 物件作為引數傳遞給連線物件的 execute() 方法，如下面的程式碼所示 -

result = conn.execute(s)

當執行上述語句時，Python shell 會回顯以下等效的 SQL 表示式 -

SELECT students.id, students.name, students.lastname
FROM students

結果變數等同於 DBAPI 中的遊標。我們現在可以使用fetchone() 方法獲取記錄。

row = result.fetchone()

可以透過for 迴圈打印表中所有選定的行，如下所示 -

for row in result:
   print (row)

列印 students 表中所有行的完整程式碼如下所示 -

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

s = students.select()
conn = engine.connect()
result = conn.execute(s)

for row in result:
   print (row)

Python shell 中顯示的輸出如下 -

(1, 'Ravi', 'Kapoor')
(2, 'Rajiv', 'Khanna')
(3, 'Komal', 'Bhandari')
(4, 'Abdul', 'Sattar')
(5, 'Priya', 'Rajhans')

可以使用Select.where()應用 SELECT 查詢的 WHERE 子句。例如，如果我們想顯示 id > 2 的行

s = students.select().where(students.c.id>2)
result = conn.execute(s)

for row in result:
   print (row)

這裡c 屬性是列的別名。shell 上將顯示以下輸出 -

(3, 'Komal', 'Bhandari')
(4, 'Abdul', 'Sattar')
(5, 'Priya', 'Rajhans')

這裡，我們需要注意的是，select 物件也可以透過 sqlalchemy.sql 模組中的 select() 函式獲得。select() 函式需要表物件作為引數。

from sqlalchemy.sql import select
s = select([users])
result = conn.execute(s)

SQLAlchemy Core - 使用文字 SQL

對於 SQL 已經知道並且不需要語句支援動態功能的情況，SQLAlchemy 允許您只使用字串。text() 構造用於組合一個文字語句，該語句基本上不加修改地傳遞給資料庫。

它構建一個新的TextClause，直接表示文字 SQL 字串，如下面的程式碼所示 -

from sqlalchemy import text
t = text("SELECT * FROM students")
result = connection.execute(t)

text() 提供的優勢超過普通字串 -

• 對繫結引數的後端中立支援
• 每個語句的執行選項
• 結果列型別行為

text() 函式需要以命名冒號格式進行繫結引數。無論資料庫後端如何，它們都保持一致。要傳送引數的值，我們將它們作為附加引數傳遞給 execute() 方法。

以下示例在文字 SQL 中使用繫結引數 -

from sqlalchemy.sql import text
s = text("select students.name, students.lastname from students where students.name between :x and :y")
conn.execute(s, x = 'A', y = 'L').fetchall()

text() 函式構建 SQL 表示式如下 -

select students.name, students.lastname from students where students.name between ? and ?

x = 'A' 和 y = 'L' 的值作為引數傳遞。結果是一個包含名稱在 'A' 和 'L' 之間的行的列表 -

[('Komal', 'Bhandari'), ('Abdul', 'Sattar')]

text() 構造支援使用 TextClause.bindparams() 方法預先建立的繫結值。引數也可以顯式地型別化，如下所示 -

stmt = text("SELECT * FROM students WHERE students.name BETWEEN :x AND :y")

stmt = stmt.bindparams(
   bindparam("x", type_= String), 
   bindparam("y", type_= String)
)

result = conn.execute(stmt, {"x": "A", "y": "L"})

The text() function also be produces fragments of SQL within a select() object that 
accepts text() objects as an arguments. The “geometry” of the statement is provided by 
select() construct , and the textual content by text() construct. We can build a statement 
without the need to refer to any pre-established Table metadata. 

from sqlalchemy.sql import select
s = select([text("students.name, students.lastname from students")]).where(text("students.name between :x and :y"))
conn.execute(s, x = 'A', y = 'L').fetchall()

您還可以使用and_() 函式組合使用 text() 函式建立的 WHERE 子句中的多個條件。

from sqlalchemy import and_
from sqlalchemy.sql import select
s = select([text("* from students")]) \
.where(
   and_(
      text("students.name between :x and :y"),
      text("students.id>2")
   )
)
conn.execute(s, x = 'A', y = 'L').fetchall()

以上程式碼獲取名稱在“A”和“L”之間且 id 大於 2 的行。程式碼的輸出如下所示 -

[(3, 'Komal', 'Bhandari'), (4, 'Abdul', 'Sattar')]

SQLAlchemy Core - 使用別名

SQL 中的別名對應於表的“重新命名”版本或 SELECT 語句，只要你說“SELECT * FROM table1 AS a”就會發生這種情況。AS 為表建立了一個新名稱。別名允許任何表或子查詢由唯一的名稱引用。

對於表，這允許在 FROM 子句中多次命名同一個表。它為語句表示的列提供了一個父名稱，允許它們相對於此名稱進行引用。

在 SQLAlchemy 中，任何 Table、select() 構造或其他可選擇物件都可以使用From Clause.alias() 方法轉換為別名，該方法會生成一個 Alias 構造。sqlalchemy.sql 模組中的 alias() 函式表示別名，通常使用 AS 關鍵字應用於 SQL 語句中的任何表或子選擇。

from sqlalchemy.sql import alias
st = students.alias("a")

此別名現在可以在 select() 構造中用於引用 students 表 -

s = select([st]).where(st.c.id>2)

這轉換為以下 SQL 表示式 -

SELECT a.id, a.name, a.lastname FROM students AS a WHERE a.id > 2

我們現在可以使用連線物件的 execute() 方法執行此 SQL 查詢。完整的程式碼如下 -

from sqlalchemy.sql import alias, select
st = students.alias("a")
s = select([st]).where(st.c.id > 2)
conn.execute(s).fetchall()

當執行以上程式碼行時，它會生成以下輸出 -

[(3, 'Komal', 'Bhandari'), (4, 'Abdul', 'Sattar'), (5, 'Priya', 'Rajhans')]

使用 UPDATE 表示式

目標表物件上的update() 方法構建等效的 UPDATE SQL 表示式。

table.update().where(conditions).values(SET expressions)

結果 update 物件上的values() 方法用於指定 UPDATE 的 SET 條件。如果保留為 None，則 SET 條件由在語句執行和/或編譯期間傳遞給語句的引數確定。

where 子句是描述 UPDATE 語句的 WHERE 條件的可選表示式。

以下程式碼片段將 students 表中 'lastname' 列的值從 'Khanna' 更改為 'Kapoor' -

stmt = students.update().where(students.c.lastname == 'Khanna').values(lastname = 'Kapoor')

stmt 物件是一個 update 物件，它轉換為 -

'UPDATE students SET lastname = :lastname WHERE students.lastname = :lastname_1'

當呼叫execute() 方法時，將替換繫結引數lastname_1。完整的更新程式碼如下 -

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()

students = Table(
   'students', 
   meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

conn = engine.connect()
stmt=students.update().where(students.c.lastname=='Khanna').values(lastname='Kapoor')
conn.execute(stmt)
s = students.select()
conn.execute(s).fetchall()

以上程式碼顯示以下輸出，其中第二行顯示更新操作的效果，如給出的螢幕截圖所示 -

[
   (1, 'Ravi', 'Kapoor'),
   (2, 'Rajiv', 'Kapoor'),
   (3, 'Komal', 'Bhandari'),
   (4, 'Abdul', 'Sattar'),
   (5, 'Priya', 'Rajhans')
]

請注意，類似的功能也可以透過使用 sqlalchemy.sql.expression 模組中的update() 函式來實現，如下所示 -

from sqlalchemy.sql.expression import update
stmt = update(students).where(students.c.lastname == 'Khanna').values(lastname = 'Kapoor')

使用 DELETE 表示式

在上一章中，我們已經瞭解了Update 表示式的作用。接下來我們要學習的表示式是Delete。

刪除操作可以透過在目標表物件上執行 delete() 方法來實現，如下面的語句所示 -

stmt = students.delete()

對於 students 表，以上程式碼行構建的 SQL 表示式如下 -

'DELETE FROM students'

但是，這將刪除 students 表中的所有行。通常 DELETE 查詢與 WHERE 子句指定的邏輯表示式相關聯。以下語句顯示 where 引數 -

stmt = students.delete().where(students.c.id > 2)

結果 SQL 表示式將包含一個繫結引數，該引數將在語句執行時在執行時替換。

'DELETE FROM students WHERE students.id > :id_1'

以下程式碼示例將刪除 students 表中 lastname 為 'Khanna' 的行 -

from sqlalchemy.sql.expression import update
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)

meta = MetaData()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

conn = engine.connect()
stmt = students.delete().where(students.c.lastname == 'Khanna')
conn.execute(stmt)
s = students.select()
conn.execute(s).fetchall()

要驗證結果，請在 SQLiteStudio 中重新整理 students 表的 data view。

SQLAlchemy Core - 使用多個表

RDBMS 的重要功能之一是在表之間建立關係。可以在相關表上執行 SELECT、UPDATE 和 DELETE 等 SQL 操作。本節使用 SQLAlchemy 描述這些操作。

為此，在我們的 SQLite 資料庫 (college.db) 中建立了兩個表。students 表與上一節中給出的結構相同；而 addresses 表具有st_id 列，該列使用外部索引鍵約束對映到students 表中的 id 列。

以下程式碼將在 college.db 中建立兩個表 -

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
engine = create_engine('sqlite:///college.db', echo=True)
meta = MetaData()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

addresses = Table(
   'addresses', meta, 
   Column('id', Integer, primary_key = True), 
   Column('st_id', Integer, ForeignKey('students.id')), 
   Column('postal_add', String), 
   Column('email_add', String))

meta.create_all(engine)

以上程式碼將轉換為 students 和 addresses 表的 CREATE TABLE 查詢，如下所示 -

CREATE TABLE students (
   id INTEGER NOT NULL,
   name VARCHAR,
   lastname VARCHAR,
   PRIMARY KEY (id)
)

CREATE TABLE addresses (
   id INTEGER NOT NULL,
   st_id INTEGER,
   postal_add VARCHAR,
   email_add VARCHAR,
   PRIMARY KEY (id),
   FOREIGN KEY(st_id) REFERENCES students (id)
)

以下螢幕截圖非常清楚地展示了以上程式碼 -

透過執行表物件的insert() 方法來填充這些表中的資料。要在 students 表中插入 5 行，您可以使用以下程式碼 -

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()

conn = engine.connect()
students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

conn.execute(students.insert(), [
   {'name':'Ravi', 'lastname':'Kapoor'},
   {'name':'Rajiv', 'lastname' : 'Khanna'},
   {'name':'Komal','lastname' : 'Bhandari'},
   {'name':'Abdul','lastname' : 'Sattar'},
   {'name':'Priya','lastname' : 'Rajhans'},
])

使用以下程式碼在 addresses 表中新增行 -

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()

addresses = Table(
   'addresses', meta, 
   Column('id', Integer, primary_key = True), 
   Column('st_id', Integer), 
   Column('postal_add', String), 
   Column('email_add', String)
)

conn.execute(addresses.insert(), [
   {'st_id':1, 'postal_add':'Shivajinagar Pune', 'email_add':'ravi@gmail.com'},
   {'st_id':1, 'postal_add':'ChurchGate Mumbai', 'email_add':'kapoor@gmail.com'},
   {'st_id':3, 'postal_add':'Jubilee Hills Hyderabad', 'email_add':'komal@gmail.com'},
   {'st_id':5, 'postal_add':'MG Road Bangaluru', 'email_add':'as@yahoo.com'},
   {'st_id':2, 'postal_add':'Cannought Place new Delhi', 'email_add':'admin@khanna.com'},
])

請注意，addresses 表中的 st_id 列引用 students 表中的 id 列。我們現在可以使用此關係從兩個表中獲取資料。我們希望從 students 表中獲取與 addresses 表中的 st_id 對應的name 和lastname。

from sqlalchemy.sql import select
s = select([students, addresses]).where(students.c.id == addresses.c.st_id)
result = conn.execute(s)

for row in result:
   print (row)

select 物件將有效地轉換為以下 SQL 表示式，該表示式在公共關係上連線兩個表 -

SELECT students.id, 
   students.name, 
   students.lastname, 
   addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM students, addresses
WHERE students.id = addresses.st_id

這將生成輸出，從兩個表中提取相應的資料，如下所示 -

(1, 'Ravi', 'Kapoor', 1, 1, 'Shivajinagar Pune', 'ravi@gmail.com')
(1, 'Ravi', 'Kapoor', 2, 1, 'ChurchGate Mumbai', 'kapoor@gmail.com')
(3, 'Komal', 'Bhandari', 3, 3, 'Jubilee Hills Hyderabad', 'komal@gmail.com')
(5, 'Priya', 'Rajhans', 4, 5, 'MG Road Bangaluru', 'as@yahoo.com')
(2, 'Rajiv', 'Khanna', 5, 2, 'Cannought Place new Delhi', 'admin@khanna.com')

使用多表更新

在上一章中，我們討論瞭如何使用多個表。因此，我們在本章中更進一步，學習多表更新。

使用 SQLAlchemy 的表物件，可以在 update() 方法的 WHERE 子句中指定多個表。PostgreSQL 和 Microsoft SQL Server 支援引用多個表的 UPDATE 語句。這實現了“UPDATE FROM”語法，該語法一次更新一個表。但是，可以在 WHERE 子句中直接的附加“FROM”子句中引用其他表。以下程式碼行清楚地解釋了多表更新的概念。

stmt = students.update().\
values({
   students.c.name:'xyz',
   addresses.c.email_add:'abc@xyz.com'
}).\
where(students.c.id == addresses.c.id)

update 物件等效於以下 UPDATE 查詢 -

UPDATE students 
SET email_add = :addresses_email_add, name = :name 
FROM addresses 
WHERE students.id = addresses.id

就 MySQL 方言而言，多個表可以嵌入到由逗號分隔的單個 UPDATE 語句中，如下所示 -

stmt = students.update().\
   values(name = 'xyz').\
   where(students.c.id == addresses.c.id)

以下程式碼描述了結果 UPDATE 查詢 -

'UPDATE students SET name = :name 
FROM addresses 
WHERE students.id = addresses.id'

但是，SQLite 方言不支援 UPDATE 中的多表條件，並顯示以下錯誤 -

NotImplementedError: This backend does not support multiple-table criteria within UPDATE

引數有序更新

原始 SQL 的 UPDATE 查詢具有 SET 子句。它由 update() 構造使用源 Table 物件中給定的列順序呈現。因此，具有特定列的特定 UPDATE 語句每次都會以相同的方式呈現。由於引數本身作為 Python 字典鍵傳遞到 Update.values() 方法，因此沒有其他可用的固定順序。

在某些情況下，SET 子句中呈現的引數順序很重要。在 MySQL 中，提供對列值的更新是基於其他列值的。

以下語句的結果 -

UPDATE table1 SET x = y + 10, y = 20

將與以下結果不同 -

UPDATE table1 SET y = 20, x = y + 10

MySQL 中的 SET 子句是基於每個值評估的，而不是基於每行評估的。為此，使用preserve_parameter_order。Python 2 元組列表作為引數傳遞給Update.values() 方法 -

stmt = table1.update(preserve_parameter_order = True).\
   values([(table1.c.y, 20), (table1.c.x, table1.c.y + 10)])

List 物件類似於字典，但它是按順序排列的。這確保“y”列的 SET 子句將首先呈現，然後是“x”列的 SET 子句。

SQLAlchemy Core - 多表刪除

在本章中，我們將瞭解多表刪除表示式，它類似於使用多表更新函式。

在許多 DBMS 方言中，可以在 DELETE 語句的 WHERE 子句中引用多個表。對於 PG 和 MySQL，使用“DELETE USING”語法；對於 SQL Server，使用“DELETE FROM”表示式引用多個表。SQLAlchemy 的delete() 構造隱式支援這兩種模式，方法是在 WHERE 子句中指定多個表，如下所示 -

stmt = users.delete().\
   where(users.c.id == addresses.c.id).\
   where(addresses.c.email_address.startswith('xyz%'))
conn.execute(stmt)

在 PostgreSQL 後端上，上述語句的結果 SQL 將呈現為 -

DELETE FROM users USING addresses
WHERE users.id = addresses.id
AND (addresses.email_address LIKE %(email_address_1)s || '%%')

如果將此方法與不支援此行為的資料庫一起使用，編譯器將引發 NotImplementedError。

SQLAlchemy Core - 使用連線

在本章中，我們將學習如何在 SQLAlchemy 中使用連線。

連線效果可以透過簡單地將兩個表放在 select() 建構函式的列子句或where 子句中來實現。現在我們使用 join() 和 outerjoin() 方法。

join() 方法從一個表物件到另一個表物件返回一個連線物件。

join(right, onclause = None, isouter = False, full = False)

上述程式碼中提到的引數的功能如下：

• right - 連線的右側；這是任何 Table 物件

• onclause - 表示連線的 ON 子句的 SQL 表示式。如果保留為 None，它會嘗試根據外部索引鍵關係連線這兩個表

• isouter - 如果為 True，則呈現 LEFT OUTER JOIN，而不是 JOIN

• full - 如果為 True，則呈現 FULL OUTER JOIN，而不是 LEFT OUTER JOIN

例如，以下 join() 方法的使用將自動導致基於外部索引鍵的連線。

>>> print(students.join(addresses))

這等效於以下 SQL 表示式：

students JOIN addresses ON students.id = addresses.st_id

您可以明確地指定連線條件，如下所示：

j = students.join(addresses, students.c.id == addresses.c.st_id)

如果我們現在使用此連線構建以下 select 建構函式：

stmt = select([students]).select_from(j)

這將導致以下 SQL 表示式：

SELECT students.id, students.name, students.lastname
FROM students JOIN addresses ON students.id = addresses.st_id

如果使用表示引擎的連線執行此語句，則將顯示屬於所選列的資料。完整程式碼如下：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
engine = create_engine('sqlite:///college.db', echo = True)

meta = MetaData()
conn = engine.connect()
students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

addresses = Table(
   'addresses', meta, 
   Column('id', Integer, primary_key = True), 
   Column('st_id', Integer,ForeignKey('students.id')), 
   Column('postal_add', String), 
   Column('email_add', String)
)

from sqlalchemy import join
from sqlalchemy.sql import select
j = students.join(addresses, students.c.id == addresses.c.st_id)
stmt = select([students]).select_from(j)
result = conn.execute(stmt)
result.fetchall()

以下是上述程式碼的輸出：

[
   (1, 'Ravi', 'Kapoor'),
   (1, 'Ravi', 'Kapoor'),
   (3, 'Komal', 'Bhandari'),
   (5, 'Priya', 'Rajhans'),
   (2, 'Rajiv', 'Khanna')
]

SQLAlchemy Core - 使用連線

連線是 SQLAlchemy 模組中的函式，它們實現 SQL 表示式 WHERE 子句中使用的關係運算符。運算子 AND、OR、NOT 等用於形成組合表示式，組合兩個單獨的邏輯表示式。以下是在 SELECT 語句中使用 AND 的一個簡單示例：

SELECT * from EMPLOYEE WHERE salary>10000 AND age>30

SQLAlchemy 函式 and_()、or_() 和 not_() 分別實現 AND、OR 和 NOT 運算子。

and_() 函式

它生成由 AND 連線的表示式的連線。以下提供了一個示例以更好地理解：

from sqlalchemy import and_

print(
   and_(
      students.c.name == 'Ravi',
      students.c.id <3
   )
)

這轉換為：

students.name = :name_1 AND students.id < :id_1

要在 students 表上的 select() 建構函式中使用 and_()，請使用以下程式碼行：

stmt = select([students]).where(and_(students.c.name == 'Ravi', students.c.id <3))

將構建以下性質的 SELECT 語句：

SELECT students.id, 
   students.name, 
   students.lastname
FROM students
WHERE students.name = :name_1 AND students.id < :id_1

顯示上述 SELECT 查詢輸出的完整程式碼如下：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey, select
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

from sqlalchemy import and_, or_
stmt = select([students]).where(and_(students.c.name == 'Ravi', students.c.id <3))
result = conn.execute(stmt)
print (result.fetchall())

假設 students 表已填充了先前示例中使用的資料，則將選擇以下行：

[(1, 'Ravi', 'Kapoor')]

or_() 函式

它生成由 OR 連線的表示式的連線。我們將使用 or_() 將上述示例中的 stmt 物件替換為以下物件

stmt = select([students]).where(or_(students.c.name == 'Ravi', students.c.id <3))

這將有效地等效於以下 SELECT 查詢：

SELECT students.id, 
   students.name, 
   students.lastname
FROM students
WHERE students.name = :name_1 
OR students.id < :id_1

一旦您進行替換並執行上述程式碼，結果將是兩行符合 OR 條件的行：

[(1, 'Ravi', 'Kapoor'),
(2, 'Rajiv', 'Khanna')]

asc() 函式

它生成一個升序 ORDER BY 子句。該函式將要應用函式的列作為引數。

from sqlalchemy import asc
stmt = select([students]).order_by(asc(students.c.name))

該語句實現以下 SQL 表示式：

SELECT students.id, 
   students.name, 
   students.lastname
FROM students 
ORDER BY students.name ASC

以下程式碼按 name 列的升序列出 students 表中的所有記錄：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey, select
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()

students = Table(
   'students', meta, 
   Column('id', Integer, primary_key = True), 
   Column('name', String), 
   Column('lastname', String), 
)

from sqlalchemy import asc
stmt = select([students]).order_by(asc(students.c.name))
result = conn.execute(stmt)

for row in result:
   print (row)

以上程式碼產生以下輸出：

(4, 'Abdul', 'Sattar')
(3, 'Komal', 'Bhandari')
(5, 'Priya', 'Rajhans')
(2, 'Rajiv', 'Khanna')
(1, 'Ravi', 'Kapoor')

desc() 函式

類似地，desc() 函式生成降序 ORDER BY 子句，如下所示：

from sqlalchemy import desc
stmt = select([students]).order_by(desc(students.c.lastname))

等效的 SQL 表示式為：

SELECT students.id, 
   students.name, 
   students.lastname
FROM students 
ORDER BY students.lastname DESC

以上程式碼行的輸出為：

(4, 'Abdul', 'Sattar')
(5, 'Priya', 'Rajhans')
(2, 'Rajiv', 'Khanna')
(1, 'Ravi', 'Kapoor')
(3, 'Komal', 'Bhandari')

between() 函式

它生成一個 BETWEEN 謂詞子句。這通常用於驗證特定列的值是否落在某個範圍內。例如，以下程式碼選擇 id 列介於 2 和 4 之間的行：

from sqlalchemy import between
stmt = select([students]).where(between(students.c.id,2,4))
print (stmt)

生成的 SQL 表示式類似於：

SELECT students.id, 
   students.name, 
   students.lastname
FROM students
WHERE students.id 
BETWEEN :id_1 AND :id_2

結果如下：

(2, 'Rajiv', 'Khanna')
(3, 'Komal', 'Bhandari')
(4, 'Abdul', 'Sattar')

SQLAlchemy Core - 使用函式

本章討論了 SQLAlchemy 中使用的一些重要函式。

標準 SQL 建議了許多函式，這些函式由大多數方言實現。它們根據傳遞給它的引數返回單個值。一些 SQL 函式將列作為引數，而一些是通用的。SQLAlchemy API 中的 thefunc 關鍵字用於生成這些函式。

在 SQL 中，now() 是一個通用函式。以下語句使用 func 呈現 now() 函式：

from sqlalchemy.sql import func
result = conn.execute(select([func.now()]))
print (result.fetchone())

以上程式碼的示例結果可能如下所示：

(datetime.datetime(2018, 6, 16, 6, 4, 40),)

另一方面，count() 函式返回從表中選擇的行數，由以下 func 用法呈現：

from sqlalchemy.sql import func
result = conn.execute(select([func.count(students.c.id)]))
print (result.fetchone())

從以上程式碼中，將獲取 students 表中行數的計數。

一些內建的 SQL 函式使用 Employee 表演示，該表包含以下資料：

max() 函式透過以下 SQLAlchemy 中 func 的用法實現，這將導致 85，即獲得的總最高分：

from sqlalchemy.sql import func
result = conn.execute(select([func.max(employee.c.marks)]))
print (result.fetchone())

類似地，將返回 56（最低分數）的 min() 函式將由以下程式碼呈現：

from sqlalchemy.sql import func
result = conn.execute(select([func.min(employee.c.marks)]))
print (result.fetchone())

因此，AVG() 函式也可以透過使用以下程式碼實現：

from sqlalchemy.sql import func
result = conn.execute(select([func.avg(employee.c.marks)]))
print (result.fetchone())

Functions are normally used in the columns clause of a select statement. 
They can also be given label as well as a type. A label to function allows the result 
to be targeted in a result row based on a string name, and a type is required when 
you need result-set processing to occur.from sqlalchemy.sql import func

result = conn.execute(select([func.max(students.c.lastname).label('Name')]))

print (result.fetchone())

SQLAlchemy Core - 使用集合運算

在上一章中，我們學習了 max()、min()、count() 等各種函式，在這裡，我們將學習集合運算及其用途。

標準 SQL 及其大多數方言都支援 UNION 和 INTERSECT 等集合運算。SQLAlchemy 透過以下函式來實現它們：

union()

在組合兩個或多個 SELECT 語句的結果時，UNION 會從結果集中消除重複項。兩個表中的列數和資料型別必須相同。

union() 函式從多個表返回一個 CompoundSelect 物件。以下示例演示了其用法：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, union
engine = create_engine('sqlite:///college.db', echo = True)

meta = MetaData()
conn = engine.connect()
addresses = Table(
   'addresses', meta, 
   Column('id', Integer, primary_key = True), 
   Column('st_id', Integer), 
   Column('postal_add', String), 
   Column('email_add', String)
)

u = union(addresses.select().where(addresses.c.email_add.like('%@gmail.com addresses.select().where(addresses.c.email_add.like('%@yahoo.com'))))

result = conn.execute(u)
result.fetchall()

union 構造轉換為以下 SQL 表示式：

SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.email_add LIKE ? UNION SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.email_add LIKE ?

從我們的 addresses 表中，以下行表示 union 操作：

[
   (1, 1, 'Shivajinagar Pune', 'ravi@gmail.com'),
   (2, 1, 'ChurchGate Mumbai', 'kapoor@gmail.com'),
   (3, 3, 'Jubilee Hills Hyderabad', 'komal@gmail.com'),
   (4, 5, 'MG Road Bangaluru', 'as@yahoo.com')
]

union_all()

UNION ALL 操作無法刪除重複項，也無法對結果集中的資料進行排序。例如，在上述查詢中，UNION 被替換為 UNION ALL 以檢視效果。

u = union_all(addresses.select().where(addresses.c.email_add.like('%@gmail.com')), addresses.select().where(addresses.c.email_add.like('%@yahoo.com')))

相應的 SQL 表示式如下：

SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.email_add LIKE ? UNION ALL SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.email_add LIKE ?

except_()

SQL EXCEPT 子句/運算子用於組合兩個 SELECT 語句，並返回第一個 SELECT 語句中未由第二個 SELECT 語句返回的行。except_() 函式生成帶有 EXCEPT 子句的 SELECT 表示式。

在以下示例中，except_() 函式僅返回 addresses 表中在 email_add 欄位中包含“gmail.com”的記錄，但排除在 postal_add 欄位中包含“Pune”的記錄。

u = except_(addresses.select().where(addresses.c.email_add.like('%@gmail.com')), addresses.select().where(addresses.c.postal_add.like('%Pune')))

上述程式碼的結果是以下 SQL 表示式：

SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.email_add LIKE ? EXCEPT SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.postal_add LIKE ?

假設 addresses 表包含前面示例中使用的資料，它將顯示以下輸出：

[(2, 1, 'ChurchGate Mumbai', 'kapoor@gmail.com'),
   (3, 3, 'Jubilee Hills Hyderabad', 'komal@gmail.com')]

intersect()

使用 INTERSECT 運算子，SQL 顯示兩個 SELECT 語句的公共行。intersect() 函式實現了此行為。

在以下示例中，兩個 SELECT 建構函式是 intersect() 函式的引數。一個返回在 email_add 列中包含“gmail.com”的行，另一個返回在 postal_add 列中包含“Pune”的行。結果將是兩個結果集的公共行。

u = intersect(addresses.select().where(addresses.c.email_add.like('%@gmail.com')), addresses.select().where(addresses.c.postal_add.like('%Pune')))

實際上，這等效於以下 SQL 語句：

SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.email_add LIKE ? INTERSECT SELECT addresses.id, 
   addresses.st_id, 
   addresses.postal_add, 
   addresses.email_add
FROM addresses
WHERE addresses.postal_add LIKE ?

兩個繫結引數“%gmail.com”和“%Pune”從 addresses 表中的原始資料生成一行，如下所示：

[(1, 1, 'Shivajinagar Pune', 'ravi@gmail.com')]

SQLAlchemy ORM - 宣告對映

SQLAlchemy 的物件關係對映 API 的主要目標是促進將使用者定義的 Python 類與資料庫表關聯起來，並將這些類的物件與它們對應表中的行關聯起來。物件和行的狀態更改會同步匹配。SQLAlchemy 使能夠根據使用者定義的類及其定義的關係來表達資料庫查詢。

ORM 建立在 SQL 表示式語言之上。這是一種高階且抽象的用法模式。事實上，ORM 是表示式語言的應用用法。

雖然可以使用物件關係對映器單獨構建成功的應用程式，但有時使用 ORM 構建的應用程式可能會直接使用表示式語言，在需要特定資料庫互動的情況下。

宣告對映

首先，呼叫 create_engine() 函式來設定一個引擎物件，該物件隨後用於執行 SQL 操作。該函式有兩個引數，一個是資料庫的名稱，另一個是 echo 引數，當設定為 True 時將生成活動日誌。如果不存在，則將建立資料庫。在以下示例中，建立了一個 SQLite 資料庫。

from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)

當呼叫 Engine.execute() 或 Engine.connect() 等方法時，引擎會建立與資料庫的真實 DBAPI 連線。然後它用於發出 SQLORM，它不直接使用引擎；相反，它在幕後由 ORM 使用。

在 ORM 的情況下，配置過程從描述資料庫表開始，然後定義將對映到這些表的類。在 SQLAlchemy 中，這兩項任務一起執行。這是透過使用宣告式系統完成的；建立的類包括描述它們對映到的實際資料庫表的指令。

一個基類在宣告式系統中儲存類的目錄和對映表。這稱為宣告式基類。在通常匯入的模組中，通常只有一個此基類的例項。declarative_base() 函式用於建立基類。此函式在 sqlalchemy.ext.declarative 模組中定義。

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

宣告基類後，可以根據它定義任意數量的對映類。以下程式碼定義了一個 Customer 的類。它包含要對映到的表，以及其中列的名稱和資料型別。

class Customers(Base):
   __tablename__ = 'customers'
   
   id = Column(Integer, primary_key = True)
   name = Column(String)
   address = Column(String)
   email = Column(String)

宣告式中的類必須具有__tablename__屬性，並且至少有一個Column是主鍵的一部分。Declarative 將所有Column物件替換為稱為描述符的特殊 Python 訪問器。此過程稱為檢測，它提供了在 SQL 上下文中引用表的方法，並能夠從資料庫中持久化和載入列的值。

此對映類就像一個普通的 Python 類，根據需要具有屬性和方法。

關於宣告式系統中類的資訊稱為表元資料。SQLAlchemy 使用 Table 物件來表示 Declarative 建立的特定表的此資訊。Table 物件根據規範建立，並透過構造 Mapper 物件與類關聯。此對映器物件不用於直接使用，而是在內部用作對映類和表之間的介面。

每個 Table 物件都是稱為 MetaData 的較大集合的成員，並且此物件可透過宣告式基類的.metadata屬性獲得。MetaData.create_all()方法是，將我們的 Engine 作為資料庫連線的來源傳遞進去。對於尚未建立的所有表，它都會向資料庫發出 CREATE TABLE 語句。

Base.metadata.create_all(engine)

建立資料庫和表以及對映 Python 類的完整指令碼如下所示：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class Customers(Base):
   __tablename__ = 'customers'
   id = Column(Integer, primary_key=True)

   name = Column(String)
   address = Column(String)
   email = Column(String)
Base.metadata.create_all(engine)

執行後，Python 控制檯將回顯以下正在執行的 SQL 表示式：

CREATE TABLE customers (
   id INTEGER NOT NULL,
   name VARCHAR,
   address VARCHAR,
   email VARCHAR,
   PRIMARY KEY (id)
)

如果我們使用 SQLiteStudio 圖形工具開啟 Sales.db，它將在其中顯示 customers 表，並具有上述結構。

SQLAlchemy ORM - 建立會話

為了與資料庫互動，我們需要獲取其控制代碼。會話物件是資料庫的控制代碼。Session 類使用 sessionmaker() 定義——一個可配置的會話工廠方法，它繫結到前面建立的引擎物件。

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)

然後使用其預設建構函式設定會話物件，如下所示：

session = Session()

下面列出了會話類的一些常用方法：

SQLAlchemy ORM - 新增物件

在 SQLAlchemy ORM 的前幾章中，我們學習瞭如何宣告對映和建立會話。在本章中，我們將學習如何將物件新增到表中。

我們已宣告 Customer 類，該類已對映到 customers 表。我們必須宣告此類的物件並透過會話物件的 add() 方法將其持久地新增到表中。

c1 = Sales(name = 'Ravi Kumar', address = 'Station Road Nanded', email = 'ravi@gmail.com')
session.add(c1)

請注意，此事務將處於掛起狀態，直到使用 commit() 方法重新整理。

session.commit()

以下是將記錄新增到 customers 表中的完整指令碼：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class Customers(Base):
   __tablename__ = 'customers'
   
   id = Column(Integer, primary_key=True)
   name = Column(String)
   address = Column(String)
   email = Column(String)
   
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()

c1 = Customers(name = 'Ravi Kumar', address = 'Station Road Nanded', email = 'ravi@gmail.com')

session.add(c1)
session.commit()

要新增多條記錄，我們可以使用會話類的 **add_all()** 方法。

session.add_all([
   Customers(name = 'Komal Pande', address = 'Koti, Hyderabad', email = 'komal@gmail.com'), 
   Customers(name = 'Rajender Nath', address = 'Sector 40, Gurgaon', email = 'nath@gmail.com'), 
   Customers(name = 'S.M.Krishna', address = 'Budhwar Peth, Pune', email = 'smk@gmail.com')]
)

session.commit()

SQLiteStudio 的表檢視顯示記錄已持久地新增到 customers 表中。下圖顯示了結果：

SQLAlchemy ORM - 使用查詢

SQLAlchemy ORM 生成的所有 SELECT 語句都由 Query 物件構建。它提供了一個生成式介面，因此連續的呼叫會返回一個新的 Query 物件，它是前一個物件的副本，並具有與其關聯的其他條件和選項。

Query 物件最初是使用 Session 的 query() 方法生成的，如下所示：

q = session.query(mapped class)

以下語句也等效於上述語句：

q = Query(mappedClass, session)

query 物件具有 all() 方法，該方法以物件列表的形式返回結果集。如果我們在 customers 表上執行它：

result = session.query(Customers).all()

此語句實際上等效於以下 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers

可以使用 For 迴圈遍歷結果物件以獲取基礎 customers 表中的所有記錄。以下是顯示 Customers 表中所有記錄的完整程式碼：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class Customers(Base):
   __tablename__ = 'customers'
   id = Column(Integer, primary_key =  True)
   name = Column(String)

   address = Column(String)
   email = Column(String)

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
result = session.query(Customers).all()

for row in result:
   print ("Name: ",row.name, "Address:",row.address, "Email:",row.email)

Python 控制檯顯示如下記錄列表：

Name: Ravi Kumar Address: Station Road Nanded Email: ravi@gmail.com
Name: Komal Pande Address: Koti, Hyderabad Email: komal@gmail.com
Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com
Name: S.M.Krishna Address: Budhwar Peth, Pune Email: smk@gmail.com

Query 物件還具有以下有用的方法：

SQLAlchemy ORM - 更新物件

在本章中，我們將瞭解如何修改或更新表中的所需值。

要修改任何物件的某個屬性的資料，我們必須為其分配新值並提交更改以使更改持久化。

讓我們從表中獲取一個物件，其主鍵識別符號在我們的 Customers 表中，ID=2。我們可以使用會話的 get() 方法，如下所示：

x = session.query(Customers).get(2)

我們可以使用以下程式碼顯示所選物件的內容：

print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)

從我們的 customers 表中，應顯示以下輸出：

Name: Komal Pande Address: Koti, Hyderabad Email: komal@gmail.com

現在我們需要透過分配如下所示的新值來更新 Address 欄位：

x.address = 'Banjara Hills Secunderabad'
session.commit()

更改將持久地反映在資料庫中。現在，我們透過使用 **first() 方法**獲取表中第一行的對應物件，如下所示：

x = session.query(Customers).first()

這將執行以下 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
LIMIT ? OFFSET ?

繫結引數將分別為 LIMIT = 1 和 OFFSET = 0，這意味著將選擇第一行。

print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)

現在，上面顯示第一行的程式碼的輸出如下所示：

Name: Ravi Kumar Address: Station Road Nanded Email: ravi@gmail.com

現在更改 name 屬性並使用以下程式碼顯示內容：

x.name = 'Ravi Shrivastava'
print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)

上面程式碼的輸出為：

Name: Ravi Shrivastava Address: Station Road Nanded Email: ravi@gmail.com

即使顯示了更改，但它尚未提交。您可以使用以下程式碼中的 **rollback() 方法**保留以前的持久位置。

session.rollback()

print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)

將顯示第一條記錄的原始內容。

對於批次更新，我們將使用 Query 物件的 update() 方法。讓我們嘗試在每一行（除了 ID = 2）的 name 中新增字首“Mr.”。相應的 update() 語句如下：

session.query(Customers).filter(Customers.id! = 2).
update({Customers.name:"Mr."+Customers.name}, synchronize_session = False)

update() 方法需要兩個引數，如下所示：

• 一個鍵值對字典，其中鍵是要更新的屬性，值是屬性的新內容。

• synchronize_session 屬性，用於說明更新會話中屬性的策略。有效值為 false：不同步會話，fetch：在更新之前執行 select 查詢以查詢與更新查詢匹配的物件；以及 evaluate：評估會話中物件的條件。

表中的四行中有三行將以“Mr.”為字首。但是，更改尚未提交，因此不會反映在 SQLiteStudio 的表檢視中。只有在提交會話時才會重新整理它。

SQLAlchemy ORM - 應用過濾器

在本章中，我們將討論如何應用過濾器以及某些過濾操作及其程式碼。

由 Query 物件表示的結果集可以透過使用 filter() 方法應用某些條件。filter 方法的一般用法如下：

session.query(class).filter(criteria)

在以下示例中，透過 SELECT 查詢在 Customers 表上獲得的結果集透過條件 (ID>2) 進行過濾：

result = session.query(Customers).filter(Customers.id>2)

此語句將轉換為以下 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id > ?

由於繫結引數 (?) 為 2，因此僅顯示 ID 列 > 2 的那些行。完整的程式碼如下所示：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class Customers(Base):
   __tablename__ = 'customers'
   
   id = Column(Integer, primary_key = True)
   name = Column(String)

   address = Column(String)
   email = Column(String)

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
result = session.query(Customers).filter(Customers.id>2)

for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

Python 控制檯顯示的輸出如下所示：

ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com
ID: 4 Name: S.M.Krishna Address: Budhwar Peth, Pune Email: smk@gmail.com

SQLAlchemy ORM - 過濾器運算子

現在，我們將學習過濾器操作及其相應的程式碼和輸出。

等於

通常使用的運算子是 ==，它應用條件來檢查相等性。

result = session.query(Customers).filter(Customers.id == 2)

for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

SQLAlchemy 將傳送以下 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id = ?

上面程式碼的輸出如下所示：

ID: 2 Name: Komal Pande Address: Banjara Hills Secunderabad Email: komal@gmail.com

不等於

不等於運算子使用 !=，它提供不等於條件。

result = session.query(Customers).filter(Customers.id! = 2)

for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

生成的 SQL 表示式為：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id != ?

上面幾行程式碼的輸出如下所示：

ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: ravi@gmail.com
ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com
ID: 4 Name: S.M.Krishna Address: Budhwar Peth, Pune Email: smk@gmail.com

喜歡

like() 方法本身為 SELECT 表示式中的 WHERE 子句生成 LIKE 條件。

result = session.query(Customers).filter(Customers.name.like('Ra%'))
for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

上面的 SQLAlchemy 程式碼等效於以下 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.name LIKE ?

上面程式碼的輸出為：

ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: ravi@gmail.com
ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com

在

此運算子檢查列值是否屬於列表中的一組專案。它由 in_() 方法提供。

result = session.query(Customers).filter(Customers.id.in_([1,3]))
for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

這裡，SQLite 引擎評估的 SQL 表示式將如下所示：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id IN (?, ?)

上面程式碼的輸出如下所示：

ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: ravi@gmail.com
ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com

和

此連線是透過以下方式生成的：**在過濾器中放置多個逗號分隔的條件或使用 and_() 方法**，如下所示：

result = session.query(Customers).filter(Customers.id>2, Customers.name.like('Ra%'))
for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

from sqlalchemy import and_
result = session.query(Customers).filter(and_(Customers.id>2, Customers.name.like('Ra%')))

for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

以上兩種方法都會產生類似的 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id > ? AND customers.name LIKE ?

上面幾行程式碼的輸出為：

ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com

或

此連線由 **or_() 方法**實現。

from sqlalchemy import or_
result = session.query(Customers).filter(or_(Customers.id>2, Customers.name.like('Ra%')))

for row in result:
   print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)

結果，SQLite 引擎獲得以下等效的 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id > ? OR customers.name LIKE ?

上面程式碼的輸出如下所示：

ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: ravi@gmail.com
ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: nath@gmail.com
ID: 4 Name: S.M.Krishna Address: Budhwar Peth, Pune Email: smk@gmail.com

返回列表和標量

Query 物件有許多方法會立即發出 SQL 並返回包含已載入資料庫結果的值。

以下是返回列表和標量的簡要概述：

all()

它返回一個列表。以下是 all() 函式的程式碼行。

session.query(Customers).all()

Python 控制檯顯示以下發出的 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers

first()

它應用一個限制為 1 並返回第一個結果作為標量。

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
LIMIT ? OFFSET ?

LIMIT 的繫結引數為 1，OFFSET 的繫結引數為 0。

one()

此命令完全獲取所有行，如果結果中不存在正好一個物件標識或複合行，則會引發錯誤。

session.query(Customers).one()

找到多行時：

MultipleResultsFound: Multiple rows were found for one()

未找到行時：

NoResultFound: No row was found for one()

one() 方法適用於期望以不同的方式處理“未找到專案”與“找到多個專案”的系統。

scalar()

它呼叫 one() 方法，並在成功時返回行的第一列，如下所示：

session.query(Customers).filter(Customers.id == 3).scalar()

這會生成以下 SQL 語句：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id = ?

SQLAlchemy ORM - 文字 SQL

之前，從 SQLAlchemy 的核心表示式語言的角度解釋了使用 text() 函式的文字 SQL。現在我們將從 ORM 的角度討論它。

透過使用 text() 結構指定其用法，可以在 Query 物件中靈活地使用文字字串。大多數適用的方法都接受它。例如，filter() 和 order_by()。

在下面給出的示例中，filter() 方法將字串“id<3”轉換為 WHERE id<3

from sqlalchemy import text
for cust in session.query(Customers).filter(text("id<3")):
   print(cust.name)

生成的原始 SQL 表示式展示了將過濾器轉換為 WHERE 子句的過程，如下程式碼所示：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE id<3

從我們 Customers 表中的示例資料中，將選擇兩行，並列印 name 列，如下所示：

Ravi Kumar
Komal Pande

要使用基於字串的 SQL 指定繫結引數，請使用冒號，並使用 params() 方法指定值。

cust = session.query(Customers).filter(text("id = :value")).params(value = 1).one()

在 Python 控制檯上顯示的有效 SQL 將如下所示：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE id = ?

要使用完全基於字串的語句，可以將表示完整語句的 text() 結構傳遞給 from_statement()。

session.query(Customers).from_statement(text("SELECT * FROM customers")).all()

上述程式碼的結果將是一個基本的 SELECT 語句，如下所示：

SELECT * FROM customers

顯然，將選擇 customers 表中的所有記錄。

text() 結構允許我們將它的文字 SQL 與 Core 或 ORM 對映的列表達式按位置關聯。我們可以透過將列表達式作為位置引數傳遞給 TextClause.columns() 方法來實現這一點。

stmt = text("SELECT name, id, name, address, email FROM customers")
stmt = stmt.columns(Customers.id, Customers.name)
session.query(Customers.id, Customers.name).from_statement(stmt).all()

即使 SQLite 引擎執行了上述程式碼生成的以下表達式，也將會選擇所有行的 id 和 name 列，該表示式顯示了 text() 方法中的所有列：

SELECT name, id, name, address, email FROM customers

SQLAlchemy ORM - 建立關係

本節描述了建立另一個表，該表與資料庫中已存在的表相關聯。customers 表包含客戶的主資料。我們現在需要建立 invoices 表，該表可能包含屬於某個客戶的任意數量的發票。這是一種一對多關係。

使用宣告式方式，我們定義此表及其對映類 Invoices，如下所示：

from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy.orm import relationship

class Customer(Base):
   __tablename__ = 'customers'

   id = Column(Integer, primary_key = True)
   name = Column(String)
   address = Column(String)
   email = Column(String)

class Invoice(Base):
   __tablename__ = 'invoices'
   
   id = Column(Integer, primary_key = True)
   custid = Column(Integer, ForeignKey('customers.id'))
   invno = Column(Integer)
   amount = Column(Integer)
   customer = relationship("Customer", back_populates = "invoices")

Customer.invoices = relationship("Invoice", order_by = Invoice.id, back_populates = "customer")
Base.metadata.create_all(engine)

這將向 SQLite 引擎傳送一個 CREATE TABLE 查詢，如下所示：

CREATE TABLE invoices (
   id INTEGER NOT NULL,
   custid INTEGER,
   invno INTEGER,
   amount INTEGER,
   PRIMARY KEY (id),
   FOREIGN KEY(custid) REFERENCES customers (id)
)

我們可以使用 SQLiteStudio 工具檢查 sales.db 中是否建立了新表。

Invoices 類在 custid 屬性上應用 ForeignKey 結構。此指令指示此列中的值應約束為 customers 表中 id 列中存在的值。這是關係型資料庫的核心功能，也是將不相關的表集合轉換為具有豐富重疊關係的“粘合劑”。

第二個指令稱為 relationship()，它告訴 ORM Invoice 類應使用屬性 Invoice.customer 與 Customer 類連結。relationship() 使用兩張表之間的外部索引鍵關係來確定此連結的性質，確定它是一對多關係。

在 Customer 對映類下的屬性 Customer.invoices 上放置了一個額外的 relationship() 指令。relationship.back_populates 引數被分配為引用補充屬性名稱，以便每個 relationship() 都可以對相同關係進行智慧決策，如反向表達所示。在一方面，Invoices.customer 指的是 Invoices 例項，另一方面，Customer.invoices 指的是 Customers 例項的列表。

relationship 函式是 SQLAlchemy ORM 包的關係 API 的一部分。它提供了兩個對映類之間的關係。這對應於父子或關聯表關係。

以下是發現的基本關係模式：

一對多

一對多關係指的是父級，它藉助子表上的外部索引鍵進行關聯。然後在父級上指定 relationship()，將其作為對子級表示的專案集合的引用。relationship.back_populates 引數用於在一對多關係中建立雙向關係，其中“反向”端是多對一關係。

多對一

另一方面，多對一關係在父表中放置一個外部索引鍵來引用子表。relationship() 在父級上宣告，其中將建立一個新的標量持有屬性。這裡同樣使用 relationship.back_populates 引數來實現雙向行為。

一對一

一對一關係本質上是雙向關係。uselist 標誌指示在關係的“多”端放置一個標量屬性而不是集合。要將一對多轉換為一對一型別的關係，請將 uselist 引數設定為 false。

多對多

多對多關係是透過新增一個關聯表來建立的，該關聯表透過定義具有其外部索引鍵的屬性來關聯兩個類。它由 relationship() 的 secondary 引數指示。通常，Table 使用與宣告式基類關聯的 MetaData 物件，以便 ForeignKey 指令可以找到要與其連結的遠端表。每個 relationship() 的 relationship.back_populates 引數建立雙向關係。關係的兩端都包含一個集合。

處理相關物件

在本章中，我們將重點關注 SQLAlchemy ORM 中的相關物件。

現在，當我們建立一個 Customer 物件時，一個空白的發票集合將以 Python 列表的形式存在。

c1 = Customer(name = "Gopal Krishna", address = "Bank Street Hydarebad", email = "gk@gmail.com")

c1.invoices 的 invoices 屬性將是一個空列表。我們可以像這樣分配列表中的專案：

c1.invoices = [Invoice(invno = 10, amount = 15000), Invoice(invno = 14, amount = 3850)]

讓我們使用 Session 物件將此物件提交到資料庫，如下所示：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
session.add(c1)
session.commit()

這將自動為 customers 和 invoices 表生成 INSERT 查詢：

INSERT INTO customers (name, address, email) VALUES (?, ?, ?) 
('Gopal Krishna', 'Bank Street Hydarebad', 'gk@gmail.com')
INSERT INTO invoices (custid, invno, amount) VALUES (?, ?, ?)
(2, 10, 15000)
INSERT INTO invoices (custid, invno, amount) VALUES (?, ?, ?)
(2, 14, 3850)

現在讓我們在 SQLiteStudio 的表檢視中檢視 customers 表和 invoices 表的內容：

您可以透過在建構函式本身中提供 invoices 的對映屬性來構建 Customer 物件，使用以下命令：

c2 = [
   Customer(
      name = "Govind Pant", 
      address = "Gulmandi Aurangabad",
      email = "gpant@gmail.com",
      invoices = [Invoice(invno = 3, amount = 10000), 
      Invoice(invno = 4, amount = 5000)]
   )
]

或者使用 session 物件的 add_all() 函式新增物件的列表，如下所示：

rows = [
   Customer(
      name = "Govind Kala", 
      address = "Gulmandi Aurangabad", 
      email = "kala@gmail.com", 
      invoices = [Invoice(invno = 7, amount = 12000), Invoice(invno = 8, amount = 18500)]),

   Customer(
      name = "Abdul Rahman", 
      address = "Rohtak", 
      email = "abdulr@gmail.com",
      invoices = [Invoice(invno = 9, amount = 15000), 
      Invoice(invno = 11, amount = 6000)
   ])
]

session.add_all(rows)
session.commit()

SQLAlchemy ORM - 使用連線

既然我們有兩個表，我們將看看如何同時對這兩個表建立查詢。要構建 Customer 和 Invoice 之間的簡單隱式連線，我們可以使用 Query.filter() 將其相關列等同起來。下面，我們將使用此方法同時載入 Customer 和 Invoice 實體：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()

for c, i in session.query(Customer, Invoice).filter(Customer.id == Invoice.custid).all():
   print ("ID: {} Name: {} Invoice No: {} Amount: {}".format(c.id,c.name, i.invno, i.amount))

SQLAlchemy 發出的 SQL 表示式如下：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email, invoices.id 
AS invoices_id, invoices.custid 
AS invoices_custid, invoices.invno 
AS invoices_invno, invoices.amount 
AS invoices_amount
FROM customers, invoices
WHERE customers.id = invoices.custid

上述程式碼行的結果如下：

ID: 2 Name: Gopal Krishna Invoice No: 10 Amount: 15000
ID: 2 Name: Gopal Krishna Invoice No: 14 Amount: 3850
ID: 3 Name: Govind Pant Invoice No: 3 Amount: 10000
ID: 3 Name: Govind Pant Invoice No: 4 Amount: 5000
ID: 4 Name: Govind Kala Invoice No: 7 Amount: 12000
ID: 4 Name: Govind Kala Invoice No: 8 Amount: 8500
ID: 5 Name: Abdul Rahman Invoice No: 9 Amount: 15000
ID: 5 Name: Abdul Rahman Invoice No: 11 Amount: 6000

可以使用 Query.join() 方法輕鬆實現實際的 SQL JOIN 語法，如下所示：

session.query(Customer).join(Invoice).filter(Invoice.amount == 8500).all()

連線的 SQL 表示式將顯示在控制檯上：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers JOIN invoices ON customers.id = invoices.custid
WHERE invoices.amount = ?

我們可以使用 for 迴圈遍歷結果：

result = session.query(Customer).join(Invoice).filter(Invoice.amount == 8500)
for row in result:
   for inv in row.invoices:
      print (row.id, row.name, inv.invno, inv.amount)

將 8500 作為繫結引數，將顯示以下輸出：

4 Govind Kala 8 8500  

Query.join() 知道如何在這些表之間連線，因為它們之間只有一個外部索引鍵。如果沒有外部索引鍵或多個外部索引鍵，當使用以下其中一種形式時，Query.join() 執行效果更好：

類似地，outerjoin() 函式可用於實現左外連線。

query.outerjoin(Customer.invoices)

subquery() 方法生成一個 SQL 表示式，表示嵌入在別名中的 SELECT 語句。

from sqlalchemy.sql import func

stmt = session.query(
   Invoice.custid, func.count('*').label('invoice_count')
).group_by(Invoice.custid).subquery()

stmt 物件將包含如下 SQL 語句：

SELECT invoices.custid, count(:count_1) AS invoice_count FROM invoices GROUP BY invoices.custid

一旦我們有了語句，它的行為就像 Table 結構一樣。語句上的列可以透過名為 c 的屬性訪問，如下面的程式碼所示：

for u, count in session.query(Customer, stmt.c.invoice_count).outerjoin(stmt, Customer.id == stmt.c.custid).order_by(Customer.id):
   print(u.name, count)

上面的 for 迴圈按名稱顯示發票計數，如下所示：

Arjun Pandit None
Gopal Krishna 2
Govind Pant 2
Govind Kala 2
Abdul Rahman 2

常見關係運算符

在本章中，我們將討論構建在關係上的運算子。

__eq__()

上述運算子是多對一“等於”比較。此運算子的程式碼行如下所示：

s = session.query(Customer).filter(Invoice.invno.__eq__(12))

上述程式碼行的等效 SQL 查詢為：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers, invoices
WHERE invoices.invno = ?

__ne__()

此運算子是多對一“不等於”比較。此運算子的程式碼行如下所示：

s = session.query(Customer).filter(Invoice.custid.__ne__(2))

上述程式碼行的等效 SQL 查詢如下所示：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers, invoices
WHERE invoices.custid != ?

contains()

此運算子用於一對多集合，以下是 contains() 的程式碼：

s = session.query(Invoice).filter(Invoice.invno.contains([3,4,5]))

上述程式碼行的等效 SQL 查詢為：

SELECT invoices.id 
AS invoices_id, invoices.custid 
AS invoices_custid, invoices.invno 
AS invoices_invno, invoices.amount 
AS invoices_amount
FROM invoices
WHERE (invoices.invno LIKE '%' + ? || '%')

any()

any() 運算子用於集合，如下所示：

s = session.query(Customer).filter(Customer.invoices.any(Invoice.invno==11))

上述程式碼行的等效 SQL 查詢如下所示：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE EXISTS (
   SELECT 1
   FROM invoices
   WHERE customers.id = invoices.custid 
   AND invoices.invno = ?)

has()

此運算子用於標量引用，如下所示：

s = session.query(Invoice).filter(Invoice.customer.has(name = 'Arjun Pandit'))

上述程式碼行的等效 SQL 查詢為：

SELECT invoices.id 
AS invoices_id, invoices.custid 
AS invoices_custid, invoices.invno 
AS invoices_invno, invoices.amount 
AS invoices_amount
FROM invoices
WHERE EXISTS (
   SELECT 1
   FROM customers
   WHERE customers.id = invoices.custid 
   AND customers.name = ?)

SQLAlchemy ORM - 提前載入

提前載入減少了查詢次數。SQLAlchemy 提供了透過查詢選項呼叫的提前載入函式，這些函式為 Query 提供了其他說明。這些選項確定如何透過 Query.options() 方法載入各種屬性。

子查詢載入

我們希望 Customer.invoices 能夠提前載入。orm.subqueryload() 選項提供了一個第二個 SELECT 語句，該語句完全載入與剛剛載入的結果關聯的集合。名稱“subquery”導致 SELECT 語句直接透過 Query 重新使用並作為子查詢嵌入到針對相關表的 SELECT 中。

from sqlalchemy.orm import subqueryload
c1 = session.query(Customer).options(subqueryload(Customer.invoices)).filter_by(name = 'Govind Pant').one()

這將產生以下兩個 SQL 表示式：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.name = ?
('Govind Pant',)

SELECT invoices.id 
AS invoices_id, invoices.custid 
AS invoices_custid, invoices.invno 
AS invoices_invno, invoices.amount 
AS invoices_amount, anon_1.customers_id 
AS anon_1_customers_id
FROM (
   SELECT customers.id 
   AS customers_id
   FROM customers
   WHERE customers.name = ?) 
   
AS anon_1 
JOIN invoices 
ON anon_1.customers_id = invoices.custid 
ORDER BY anon_1.customers_id, invoices.id 2018-06-25 18:24:47,479 
INFO sqlalchemy.engine.base.Engine ('Govind Pant',)

要訪問兩個表中的資料，我們可以使用以下程式：

print (c1.name, c1.address, c1.email)

for x in c1.invoices:
   print ("Invoice no : {}, Amount : {}".format(x.invno, x.amount))

上述程式的輸出如下：

Govind Pant Gulmandi Aurangabad gpant@gmail.com
Invoice no : 3, Amount : 10000
Invoice no : 4, Amount : 5000

連線載入

另一個函式稱為 orm.joinedload()。這會發出一個 LEFT OUTER JOIN。引導物件以及相關物件或集合將在一步中載入。

from sqlalchemy.orm import joinedload
c1 = session.query(Customer).options(joinedload(Customer.invoices)).filter_by(name='Govind Pant').one()

這會發出以下表達式，輸出與上面相同：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email, invoices_1.id 
AS invoices_1_id, invoices_1.custid 
AS invoices_1_custid, invoices_1.invno 
AS invoices_1_invno, invoices_1.amount 
AS invoices_1_amount

FROM customers 
LEFT OUTER JOIN invoices 
AS invoices_1 
ON customers.id = invoices_1.custid

WHERE customers.name = ? ORDER BY invoices_1.id
('Govind Pant',)

OUTER JOIN 產生了兩個行，但它只返回一個 Customer 例項。這是因為 Query 根據物件標識對返回的實體應用了“唯一化”策略。可以應用連線提前載入而不會影響查詢結果。

subqueryload() 更適合載入相關集合，而 joinedload() 更適合多對一關係。

SQLAlchemy ORM - 刪除相關物件

在單個表上執行刪除操作很容易。您只需從會話中刪除對映類的物件並提交操作即可。但是，在多個相關表上執行刪除操作有點棘手。

在我們的 sales.db 資料庫中，Customer 和 Invoice 類分別對映到 customer 和 invoice 表，它們之間存在一對多型別的關係。我們將嘗試刪除 Customer 物件並檢視結果。

作為快速參考，以下是 Customer 和 Invoice 類的定義：

from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy.orm import relationship
class Customer(Base):
   __tablename__ = 'customers'

   id = Column(Integer, primary_key = True)
   name = Column(String)
   address = Column(String)
   email = Column(String)
   
class Invoice(Base):
   __tablename__ = 'invoices'

   id = Column(Integer, primary_key = True)
   custid = Column(Integer, ForeignKey('customers.id'))
   invno = Column(Integer)
   amount = Column(Integer)
   customer = relationship("Customer", back_populates = "invoices")
   
Customer.invoices = relationship("Invoice", order_by = Invoice.id, back_populates = "customer")

我們設定一個會話，並透過使用以下程式查詢主鍵 ID 來獲取 Customer 物件：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
x = session.query(Customer).get(2)

在我們的示例表中，x.name 恰好是“Gopal Krishna”。讓我們從會話中刪除此 x 並計算此名稱出現的次數。

session.delete(x)
session.query(Customer).filter_by(name = 'Gopal Krishna').count()

生成的 SQL 表示式將返回 0。

SELECT count(*) 
AS count_1
FROM (
   SELECT customers.id 
   AS customers_id, customers.name 
   AS customers_name, customers.address 
   AS customers_address, customers.email 
   AS customers_email
   FROM customers
   WHERE customers.name = ?) 
AS anon_1('Gopal Krishna',) 0

但是，x 的相關 Invoice 物件仍然存在。可以透過以下程式碼驗證：

session.query(Invoice).filter(Invoice.invno.in_([10,14])).count()

這裡，10 和 14 是屬於客戶 Gopal Krishna 的發票編號。上述查詢的結果為 2，這意味著相關物件尚未刪除。

SELECT count(*) 
AS count_1
FROM (
   SELECT invoices.id 
   AS invoices_id, invoices.custid 
   AS invoices_custid, invoices.invno 
   AS invoices_invno, invoices.amount 
   AS invoices_amount
   FROM invoices
   WHERE invoices.invno IN (?, ?)) 
AS anon_1(10, 14) 2

這是因為 SQLAlchemy 不會假設級聯刪除；我們必須發出刪除它的命令。

要更改行為，我們可以在 User.addresses 關係上配置級聯選項。讓我們關閉正在進行的會話，使用新的 declarative_base() 並重新宣告 User 類，包括級聯配置在內的地址關係。

關係函式中的 cascade 屬性是一個逗號分隔的級聯規則列表，它決定了 Session 操作如何從父級“級聯”到子級。預設情況下，它是 False，這意味著它是“save-update, merge”。

可用的級聯如下：

• save-update
• merge
• expunge
• delete
• delete-orphan
• refresh-expire

常用的選項是“all, delete-orphan”，表示相關物件在所有情況下都應跟隨父物件，並在解除關聯時被刪除。

因此，重新宣告的 Customer 類如下所示：

class Customer(Base): 
   __tablename__ = 'customers'
   
   id = Column(Integer, primary_key = True) 
   name = Column(String) 
   address = Column(String) 
   email = Column(String) 
   invoices = relationship(
      "Invoice", 
      order_by = Invoice.id, 
      back_populates = "customer",
      cascade = "all, 
      delete, delete-orphan" 
   )

讓我們使用下面的程式刪除名為 Gopal Krishna 的 Customer，並檢視其相關 Invoice 物件的數量：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
x = session.query(Customer).get(2)
session.delete(x)
session.query(Customer).filter_by(name = 'Gopal Krishna').count()
session.query(Invoice).filter(Invoice.invno.in_([10,14])).count()

現在計數為 0，上面指令碼發出的 SQL 如下：

SELECT customers.id 
AS customers_id, customers.name 
AS customers_name, customers.address 
AS customers_address, customers.email 
AS customers_email
FROM customers
WHERE customers.id = ?
(2,)
SELECT invoices.id 
AS invoices_id, invoices.custid 
AS invoices_custid, invoices.invno 
AS invoices_invno, invoices.amount
AS invoices_amount
FROM invoices
WHERE ? = invoices.custid 
ORDER BY invoices.id (2,)
DELETE FROM invoices 
WHERE invoices.id = ? ((1,), (2,))
DELETE FROM customers 
WHERE customers.id = ? (2,)
SELECT count(*) 
AS count_1
FROM (
   SELECT customers.id 
   AS customers_id, customers.name 
   AS customers_name, customers.address 
   AS customers_address, customers.email 
   AS customers_email
   FROM customers
   WHERE customers.name = ?) 
AS anon_1('Gopal Krishna',)
SELECT count(*) 
AS count_1
FROM (
   SELECT invoices.id 
   AS invoices_id, invoices.custid 
   AS invoices_custid, invoices.invno 
   AS invoices_invno, invoices.amount 
   AS invoices_amount
   FROM invoices
   WHERE invoices.invno IN (?, ?)) 
AS anon_1(10, 14)
0

多對多關係

兩個表之間的多對多關係是透過新增一個關聯表來實現的，該表有兩個外部索引鍵——一個來自每個表的主鍵。此外，對映到這兩個表的類具有一個屬性，該屬性包含一組來自其他關聯表的物件，這些物件被分配為 relationship() 函式的 secondary 屬性。

為此，我們將建立一個 SQLite 資料庫 (mycollege.db)，其中包含兩個表——department 和 employee。在這裡，我們假設一個員工可以屬於多個部門，一個部門可以有多個員工。這構成了多對多關係。

對映到 department 和 employee 表的 Employee 和 Department 類的定義如下：

from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
engine = create_engine('sqlite:///mycollege.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy.orm import relationship

class Department(Base):
   __tablename__ = 'department'
   id = Column(Integer, primary_key = True)
   name = Column(String)
   employees = relationship('Employee', secondary = 'link')
   
class Employee(Base):
   __tablename__ = 'employee'
   id = Column(Integer, primary_key = True)
   name = Column(String)
   departments = relationship(Department,secondary='link')

現在我們定義一個 Link 類。它連結到 link 表，分別包含 department_id 和 employee_id 屬性，分別引用 department 和 employee 表的主鍵。

class Link(Base):
   __tablename__ = 'link'
   department_id = Column(
      Integer, 
      ForeignKey('department.id'), 
      primary_key = True)

employee_id = Column(
   Integer, 
   ForeignKey('employee.id'), 
   primary_key = True)

這裡，我們需要注意的是，Department 類具有與 Employee 類相關的 employees 屬性。relationship 函式的 secondary 屬性被分配了一個 link 作為其值。

類似地，Employee 類具有與 Department 類相關的 departments 屬性。relationship 函式的 secondary 屬性被分配了一個 link 作為其值。

執行以下語句時，將建立所有這三個表：

Base.metadata.create_all(engine)

Python 控制檯發出以下 CREATE TABLE 查詢：

CREATE TABLE department (
   id INTEGER NOT NULL,
   name VARCHAR,
   PRIMARY KEY (id)
)

CREATE TABLE employee (
   id INTEGER NOT NULL,
   name VARCHAR,
   PRIMARY KEY (id)
)

CREATE TABLE link (
   department_id INTEGER NOT NULL,
   employee_id INTEGER NOT NULL,
   PRIMARY KEY (department_id, employee_id),
   FOREIGN KEY(department_id) REFERENCES department (id),
   FOREIGN KEY(employee_id) REFERENCES employee (id)
)

我們可以透過使用 SQLiteStudio 開啟 mycollege.db 來檢查這一點，如下面的螢幕截圖所示：

接下來，我們建立三個 Department 類的物件和三個 Employee 類的物件，如下所示：

d1 = Department(name = "Accounts")
d2 = Department(name = "Sales")
d3 = Department(name = "Marketing")

e1 = Employee(name = "John")
e2 = Employee(name = "Tony")
e3 = Employee(name = "Graham")

每個表都有一個包含 append() 方法的集合屬性。我們可以將 Employee 物件新增到 Department 物件的 Employees 集合中。類似地，我們可以將 Department 物件新增到 Employee 物件的 departments 集合屬性中。

e1.departments.append(d1)
e2.departments.append(d3)
d1.employees.append(e3)
d2.employees.append(e2)
d3.employees.append(e1)
e3.departments.append(d2)

現在我們要做的就是設定一個 session 物件，將所有物件新增到其中並提交更改，如下所示：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
session.add(e1)
session.add(e2)
session.add(d1)
session.add(d2)
session.add(d3)
session.add(e3)
session.commit()

以下 SQL 語句將在 Python 控制檯上發出：

INSERT INTO department (name) VALUES (?) ('Accounts',)
INSERT INTO department (name) VALUES (?) ('Sales',)
INSERT INTO department (name) VALUES (?) ('Marketing',)
INSERT INTO employee (name) VALUES (?) ('John',)
INSERT INTO employee (name) VALUES (?) ('Graham',)
INSERT INTO employee (name) VALUES (?) ('Tony',)
INSERT INTO link (department_id, employee_id) VALUES (?, ?) ((1, 2), (3, 1), (2, 3))
INSERT INTO link (department_id, employee_id) VALUES (?, ?) ((1, 1), (2, 2), (3, 3))

要檢查上述操作的效果，請使用 SQLiteStudio 並檢視 department、employee 和 link 表中的資料：

要顯示資料，請執行以下查詢語句：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()

for x in session.query( Department, Employee).filter(Link.department_id == Department.id, 
   Link.employee_id == Employee.id).order_by(Link.department_id).all():
   print ("Department: {} Name: {}".format(x.Department.name, x.Employee.name))

根據我們示例中填充的資料，輸出將顯示如下：

Department: Accounts Name: John
Department: Accounts Name: Graham
Department: Sales Name: Graham
Department: Sales Name: Tony
Department: Marketing Name: John
Department: Marketing Name: Tony

SQLAlchemy - 方言

SQLAlchemy 使用方言系統與各種型別的資料庫進行通訊。每個資料庫都有一個相應的 DBAPI 包裝器。所有方言都需要安裝相應的 DBAPI 驅動程式。

SQLAlchemy API 中包含以下方言：

• Firebird
• Microsoft SQL Server
• MySQL
• Oracle
• PostgreSQL
• SQL
• Sybase

create_engine() 函式生成基於 URL 的 Engine 物件。這些 URL 可以包含使用者名稱、密碼、主機名和資料庫名稱。可能存在用於其他配置的可選關鍵字引數。在某些情況下，會接受檔案路徑，而在其他情況下，“資料來源名稱”會替換“主機”和“資料庫”部分。資料庫 URL 的典型形式如下：

dialect+driver://username:password@host:port/database

PostgreSQL

PostgreSQL 方言使用psycopg2作為預設的 DBAPI。pg8000 也可用作純 Python 替代品，如下所示

# default
engine = create_engine('postgresql://scott:tiger@localhost/mydatabase')

# psycopg2
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')

# pg8000
engine = create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')

MySQL

MySQL 方言使用mysql-python作為預設的 DBAPI。有許多可用的 MySQL DBAPI，例如 MySQL-connector-python，如下所示：

# default
engine = create_engine('mysql://scott:tiger@localhost/foo')

# mysql-python
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')

# MySQL-connector-python
engine = create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')

Oracle

Oracle 方言使用cx_oracle作為預設的 DBAPI，如下所示：

engine = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')
engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')

Microsoft SQL Server

SQL Server 方言使用pyodbc作為預設的 DBAPI。pymssql 也可用。

# pyodbc
engine = create_engine('mssql+pyodbc://scott:tiger@mydsn')

# pymssql
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')

SQLite

SQLite 連線到基於檔案的資料庫，預設情況下使用 Python 內建模組sqlite3。由於 SQLite 連線到本地檔案，因此 URL 格式略有不同。“file”部分是資料庫的檔名。對於相對檔案路徑，這需要三個斜槓，如下所示：

engine = create_engine('sqlite:///foo.db')

對於絕對檔案路徑，三個斜槓後跟絕對路徑，如下所示：

engine = create_engine('sqlite:///C:\\path\\to\\foo.db')

要使用 SQLite:memory: 資料庫，請指定一個空 URL，如下所示：

engine = create_engine('sqlite://')

結論

在本教程的第一部分，我們學習瞭如何使用表示式語言執行 SQL 語句。表示式語言將 SQL 結構嵌入到 Python 程式碼中。在第二部分中，我們討論了 SQLAlchemy 的物件關係對映功能。ORM API 將 SQL 表與 Python 類對映。