SQLAlchemy 2.0 查詢使用指南

1. 環境設置

首先，需要安裝 SQLAlchemy 2.0 版本。假設你使用的是 SQLite 數據庫，可以通過以下命令安裝 SQLAlchemy：

pip install sqlalchemy

接著，我們創建數據庫連接并初始化會話：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker# 創建數據庫引擎
engine = create_engine('sqlite:///example.db', echo=True)# 創建會話
Session = sessionmaker(bind=engine)
session = Session()

2. 定義數據模型

SQLAlchemy 使用 ORM (對象關系映射) 使得 Python 對象和數據庫表之間實現映射。以下是定義 User? 和 Address? 表的模型示例：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationshipBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String)age = Column(Integer)# 一對多關系，用戶可以有多個地址addresses = relationship('Address', back_populates='user', lazy='select')class Address(Base):__tablename__ = 'addresses'id = Column(Integer, primary_key=True)user_id = Column(Integer, ForeignKey('users.id'))email = Column(String)# 反向關系user = relationship('User', back_populates='addresses')

在這個模型中，User? 表和 Address? 表通過 user_id? 建立了外鍵關聯，用戶可以有多個地址。

3. 基本查詢操作

SQLAlchemy 2.0 提供了一個更簡潔和一致的查詢接口。以下是一些常見的查詢操作。

3.1 查詢所有記錄

from sqlalchemy import selectstmt = select(User)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查詢：

SELECT * FROM users;

3.2 查詢特定字段

查詢 name? 和 age? 字段：

stmt = select(User.name, User.age)
result = session.execute(stmt).all()for name, age in result:print(name, age)

SQL 原生查詢：

SELECT name, age FROM users;

3.3 查詢帶條件的記錄

查詢年齡大于 30 的用戶：

stmt = select(User).where(User.age > 30)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查詢：

SELECT * FROM users WHERE age > 30;

3.4 排序查詢

根據年齡降序排列查詢用戶：

stmt = select(User).order_by(User.age.desc())
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查詢：

SELECT * FROM users ORDER BY age DESC;

3.5 限制查詢結果

限制返回的記錄數為 5：

stmt = select(User).limit(5)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查詢：

SELECT * FROM users LIMIT 5;

4. 關聯查詢（Join）

4.1 懶查詢 (Lazy Loading)

懶查詢是 SQLAlchemy 默認的加載方式，只有在訪問關聯屬性時才會執行查詢。它通過延遲加載來避免不必要的查詢，但是可能導致 “N+1 查詢問題”。

# 查詢用戶，并訪問用戶的地址
user = session.query(User).first()  # 執行查詢，不查詢 addresses
print(user.name)  # 輸出用戶名稱addresses = user.addresses  # 當訪問 addresses 時，SQLAlchemy 會查詢 addresses 表
for address in addresses:print(address.email)

SQL 原生查詢：

SELECT * FROM users WHERE id = 1;
-- 查詢用戶的地址
SELECT * FROM addresses WHERE user_id = 1;

4.2 預加載（Eager Loading）

預加載是通過 JOIN 或子查詢一次性加載所有相關數據。常見的預加載方法有 joinedload?、subqueryload? 和 selectinload?。

4.2.1 joinedload?

?joinedload? 會通過 JOIN 一次性加載所有相關數據，適用于關聯數據量少的情況。

from sqlalchemy.orm import joinedloadstmt = select(User).options(joinedload(User.addresses))
result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

SQL 原生查詢：

SELECT * FROM users
JOIN addresses ON users.id = addresses.user_id;

4.2.2 subqueryload?

?subqueryload? 使用子查詢來加載關聯數據，適用于關聯數據較多的情況。

from sqlalchemy.orm import subqueryloadstmt = select(User).options(subqueryload(User.addresses))
result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

SQL 原生查詢：

SELECT * FROM users;
-- 查詢 addresses 表的所有數據
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);

4.2.3 selectinload?

?selectinload? 適用于多對一和一對多的關系，能夠通過一次查詢批量加載所有關聯數據。

from sqlalchemy.orm import selectinloadstmt = select(User).options(selectinload(User.addresses))
result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

SQL 原生查詢：

SELECT * FROM users;
-- 查詢 addresses 表
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);

4.3 延遲加載與預加載的比較

特性	懶查詢 (Lazy Loading)	預加載 (Eager Loading)
數據加載方式	延遲加載，直到訪問關聯屬性時才查詢	一次性加載所有關聯數據
查詢次數	可能多次查詢（N+1 查詢問題）	一次性查詢，通常只發出少量查詢
性能	對于小數據量高效	對于復雜查詢避免 N+1 問題，適用于大量關聯數據
常用方法	無	?joinedload?、subqueryload?、selectinload?

5. 高級查詢

5.1 聚合查詢

SQLAlchemy 支持聚合函數，如 count?、sum?、avg? 等。以下是計算用戶數量的例子：

from sqlalchemy import funcstmt = select(func.count(User.id))
result = session.execute(stmt).scalar()
print(f"User Count: {result}")

SQL 原生查詢：

SELECT COUNT(id) FROM users;

5.2 分組查詢

通過 group_by? 方法進行分組查詢：

stmt = select(User.age, func.count(User.id)).group_by(User.age)
result = session.execute(stmt).all()for age, count in result:print(f"Age: {age}, Count: {count}")

SQL 原生查詢：

SELECT age, COUNT(id) FROM users GROUP BY age;

5.3 子查詢

通過子查詢來執行更復雜的查詢：

subquery = select(func.max(User.age)).scalar_subquery()
stmt = select(User).where(User.age == subquery)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查詢：

SELECT * FROM users WHERE age = (SELECT MAX(age) FROM users);

6. 事務管理

SQLAlchemy 2.0 自動管理事務，但你也可以顯式地管理事務。例如：

from sqlalchemy.exc import SQLAlchemyErrortry:new_user = User(name="David", age=28)session.add(new_user)session.commit()  # 提交事務
except SQLAlchemyError:session.rollback()  # 回滾事務print("事務失敗，已回滾")

7. 總結

SQLAlchemy 2.0 提供了強大的 ORM 功能和靈活的查詢接口。合理選擇懶查詢和預加載策略可以有效避免性能問題，特別是對于關系復雜的數據模型，預加載能幫助避免 N+1 查詢問題。通過使用 select()?、joinedload?、subqueryload? 等方法，我們可以優化查詢性能，提高數據操作的效率。

當然，以下是更詳細的 SQLAlchemy 2.0 查詢使用指南，包括懶查詢（Lazy Loading）、預加載（Eager Loading）以及異步 session? 使用示例。我們將通過詳細的案例來解釋這些概念，幫助團隊成員全面理解 SQLAlchemy 2.0 的使用方式。

SQLAlchemy 2.0 查詢使用指南

1. 環境設置

首先，確保安裝了 SQLAlchemy 2.0：

pip install sqlalchemy

對于異步支持，確保你安裝了 asyncpg?（適用于 PostgreSQL）或 aiomysql?（適用于 MySQL）等異步數據庫驅動。

pip install sqlalchemy[asyncio] asyncpg

1.1 創建數據庫引擎和會話

數據庫連接使用 create_engine()? 來創建數據庫引擎，會話使用 sessionmaker()? 來創建。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker# 創建數據庫引擎
engine = create_engine('sqlite+aiosqlite:///example.db', echo=True, future=True)# 創建異步會話
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine# 異步引擎
engine = create_async_engine('sqlite+aiosqlite:///example.db', echo=True, future=True)# 異步會話
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

2. 定義數據模型

SQLAlchemy ORM 通過 declarative_base()? 定義數據庫模型。這里我們定義 User? 和 Address? 模型。

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationshipBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String)age = Column(Integer)# 一對多關系addresses = relationship('Address', back_populates='user', lazy='select')class Address(Base):__tablename__ = 'addresses'id = Column(Integer, primary_key=True)user_id = Column(Integer, ForeignKey('users.id'))email = Column(String)user = relationship('User', back_populates='addresses')

2.1 解釋模型

在這個模型中，User? 表有一個一對多關系與 Address? 表關聯，User? 表的 addresses? 屬性代表用戶的多個地址，而 Address? 表的 user? 屬性反向關聯用戶。

3. 基本查詢操作

SQLAlchemy 2.0 提供了更加直觀的查詢接口。以下是一些常見的查詢操作。

3.1 查詢所有記錄

查詢 User? 表中的所有用戶數據：

from sqlalchemy import select# 創建查詢語句
stmt = select(User)# 執行查詢
with session.begin():result = session.execute(stmt).scalars().all()# 輸出結果
for user in result:print(user.name, user.age)

原生 SQL 查詢：

SELECT * FROM users;

3.2 查詢特定字段

查詢 name? 和 age? 字段的用戶數據：

stmt = select(User.name, User.age)with session.begin():result = session.execute(stmt).all()for name, age in result:print(name, age)

原生 SQL 查詢：

SELECT name, age FROM users;

3.3 查詢帶條件的記錄

查詢年齡大于 30 的用戶：

stmt = select(User).where(User.age > 30)with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

原生 SQL 查詢：

SELECT * FROM users WHERE age > 30;

3.4 排序查詢

按年齡降序排列查詢：

stmt = select(User).order_by(User.age.desc())with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

原生 SQL 查詢：

SELECT * FROM users ORDER BY age DESC;

3.5 限制查詢結果

限制返回前 5 條記錄：

stmt = select(User).limit(5)with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

原生 SQL 查詢：

SELECT * FROM users LIMIT 5;

4. 關聯查詢（Join）

SQLAlchemy 提供了強大的關聯查詢功能。我們可以使用 joinedload?、subqueryload? 等方法來優化查詢。

4.1 懶查詢 (Lazy Loading)

懶查詢是 SQLAlchemy 默認的加載方式。只有當你訪問某個關聯屬性時，SQLAlchemy 會延遲加載數據。

user = session.query(User).first()  # 執行查詢，不查詢 addresses
print(user.name)  # 輸出用戶名稱
addresses = user.addresses  # 此時會查詢 addresses 表
for address in addresses:print(address.email)

原生 SQL 查詢：

SELECT * FROM users WHERE id = 1;
-- 查詢地址
SELECT * FROM addresses WHERE user_id = 1;

4.2 預加載（Eager Loading）

為了避免 N+1 查詢問題，可以使用預加載。以下是幾種常用的預加載方式。

4.2.1 joinedload?

?joinedload? 使用 SQL JOIN 一次性加載所有關聯數據，適用于數據量較小的場景。

from sqlalchemy.orm import joinedloadstmt = select(User).options(joinedload(User.addresses))with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

原生 SQL 查詢：

SELECT * FROM users
JOIN addresses ON users.id = addresses.user_id;

4.2.2 subqueryload?

?subqueryload? 使用子查詢加載關聯數據，適用于關聯數據較多的情況。

from sqlalchemy.orm import subqueryloadstmt = select(User).options(subqueryload(User.addresses))with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

原生 SQL 查詢：

SELECT * FROM users;
-- 查詢 addresses 表的所有數據
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);

4.2.3 selectinload?

?selectinload? 通過批量查詢一次性加載關聯數據，適用于多對一和一對多的關系。

from sqlalchemy.orm import selectinloadstmt = select(User).options(selectinload(User.addresses))with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

原生 SQL 查詢：

SELECT * FROM users;
-- 查詢 addresses 表
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);

4.3 延遲加載與預加載的比較

特性	懶查詢 (Lazy Loading)	預加載 (Eager Loading)
數據加載方式	延遲加載，直到訪問關聯屬性時才查詢	一次性加載所有關聯數據
查詢次數	可能多次查詢（N+1 查詢問題）	一次性查詢，通常只發出少量查詢
性能	對于小數據量高效	對于復雜查詢避免 N+1 問題，適用于大量關聯數據
常用方法	無	?joinedload?、subqueryload?、selectinload?

5. 高級查詢

5.1 聚合查詢

SQLAlchemy 支持聚合函數，如 count?、sum?、avg? 等。以下是計算用戶數量的例子：

from sqlalchemy import funcstmt = select(func.count(User.id))with session.begin():result = session.execute(stmt).scalar()print(f"User Count: {result}")

SQL 原生查詢：

SELECT COUNT(id) FROM users;

5.2 分組查詢

通過 group_by? 方法進行分組查詢：

stmt = select(User.age, func.count(User.id)).group_by(User.age)with session.begin():result = session.execute(stmt).all()for age, count in result:print(f"Age: {age}, Count: {count}")

SQL 原生查詢：

SELECT age, COUNT(id) FROM users GROUP BY age;

5.3 子查詢

通過子查詢來執行更復雜的查詢：

subquery = select(func.max(User.age)).scalar_subquery()stmt = select(User).where(User.age == subquery)with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查詢：

SELECT * FROM users WHERE age = (SELECT MAX(age) FROM users);

6. 異步查詢使用示例

SQLAlchemy 支持異步查詢，適用于需要處理高并發操作的場景。以下是使用異步 session? 進行查詢的示例。

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select# 創建異步引擎和會話
engine = create_async_engine('sqlite+aiosqlite:///example.db', echo=True, future=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)async def async_query():async with async_session() as session:stmt = select(User)result = await session.execute(stmt)users = result.scalars().all()for user in users:print(user.name)# 執行異步查詢
import asyncio
asyncio.run(async_query())

在異步查詢中，我們使用 await? 來執行查詢，確保數據庫操作不會阻塞主線程。