sqlalchemy如何使用

SQLAlchemy 如何使用

目錄

1. 簡介
2. 安裝與配置
3. 核心概念
   • 引擎與連接
   • 會話
   • 模型與表
   • 查詢
4. ORM 使用
   • 定義模型
   • 創建與刪除表
   • 增刪改查
   • 關系與關聯
5. 表達式語言
   • 選擇
   • 插入
   • 更新
   • 刪除
6. 高級特性
   • 事務管理
   • 事件與鉤子
   • 連接池
   • 性能優化
7. 常見問題與解決方案
8. 總結

簡介

SQLAlchemy 是一個功能強大的 Python SQL 工具包和對象關系映射（ORM）框架。它提供了完整的 SQL 表達式語言和 ORM 功能，使得開發者能夠以 Pythonic 的方式與數據庫進行交互。SQLAlchemy 的設計目標是提供靈活性和高性能，同時保持代碼的可讀性和可維護性。

SQLAlchemy 的核心組件包括：

• 引擎（Engine）：負責管理與數據庫的連接。
• 會話（Session）：用于管理 ORM 對象的生命周期和事務。
• 模型（Model）：用于定義數據庫表的結構和關系。
• 查詢（Query）：用于構建和執行數據庫查詢。

SQLAlchemy 支持多種數據庫后端，包括 PostgreSQL、MySQL、SQLite、Oracle 等。

安裝與配置

安裝

使用 pip 安裝 SQLAlchemy：

pip install sqlalchemy

配置

在使用 SQLAlchemy 之前，需要配置數據庫連接。通常，這涉及到創建一個引擎（Engine）實例，該實例將管理與數據庫的連接。

from sqlalchemy import create_engine

# 創建引擎
engine = create_engine('sqlite:///example.db', echo=True)

create_engine 函數的第一個參數是數據庫的 URL，格式為 dialect+driver://username:password@host:port/database。echo=True 參數用于啟用 SQL 語句的日志輸出，便于調試。

核心概念

引擎與連接

引擎（Engine） 是 SQLAlchemy 的核心組件之一，負責管理與數據庫的連接。引擎本身并不直接與數據庫通信，而是通過連接池（Connection Pool）來管理連接。

from sqlalchemy import create_engine

# 創建引擎
engine = create_engine('sqlite:///example.db')

# 獲取連接
connection = engine.connect()

# 執行 SQL 語句
result = connection.execute("SELECT * FROM users")
for row in result:
    print(row)

# 關閉連接
connection.close()

會話

會話（Session） 是 ORM 的核心組件之一，用于管理 ORM 對象的生命周期和事務。會話提供了一個工作單元（Unit of Work）模式，使得開發者能夠以面向對象的方式與數據庫進行交互。

from sqlalchemy.orm import sessionmaker

# 創建會話類
Session = sessionmaker(bind=engine)

# 創建會話實例
session = Session()

# 使用會話進行數據庫操作
new_user = User(name='Alice')
session.add(new_user)
session.commit()

# 關閉會話
session.close()

模型與表

模型（Model） 是 ORM 的核心概念之一，用于定義數據庫表的結構和關系。模型類通常繼承自 Base 類，并使用 Column 類定義表的列。

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

# 創建基類
Base = declarative_base()

# 定義模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

# 創建表
Base.metadata.create_all(engine)

查詢

查詢（Query） 是 ORM 的核心功能之一，用于構建和執行數據庫查詢。查詢對象通常通過會話實例創建，并支持鏈式調用。

# 查詢所有用戶
users = session.query(User).all()
for user in users:
    print(user.name)

# 條件查詢
user = session.query(User).filter_by(name='Alice').first()
print(user.age)

# 排序查詢
users = session.query(User).order_by(User.age).all()
for user in users:
    print(user.name, user.age)

ORM 使用

定義模型

在 SQLAlchemy 中，模型類用于定義數據庫表的結構和關系。模型類通常繼承自 Base 類，并使用 Column 類定義表的列。

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = 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")

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="addresses")

創建與刪除表

使用 Base.metadata.create_all() 方法可以創建所有定義的表，使用 Base.metadata.drop_all() 方法可以刪除所有定義的表。

# 創建表
Base.metadata.create_all(engine)

# 刪除表
Base.metadata.drop_all(engine)

增刪改查

增加記錄

# 創建新用戶
new_user = User(name='Alice', age=25)
session.add(new_user)
session.commit()

刪除記錄

# 刪除用戶
user = session.query(User).filter_by(name='Alice').first()
session.delete(user)
session.commit()

修改記錄

# 修改用戶年齡
user = session.query(User).filter_by(name='Alice').first()
user.age = 30
session.commit()

查詢記錄

# 查詢所有用戶
users = session.query(User).all()
for user in users:
    print(user.name, user.age)

# 條件查詢
user = session.query(User).filter_by(name='Alice').first()
print(user.age)

# 排序查詢
users = session.query(User).order_by(User.age).all()
for user in users:
    print(user.name, user.age)

關系與關聯

SQLAlchemy 支持多種關系類型，包括一對一、一對多、多對一和多對多關系。通過 relationship 函數可以定義模型之間的關系。

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="addresses")

表達式語言

SQLAlchemy 的表達式語言（Expression Language）提供了一種靈活的方式來構建 SQL 查詢。與 ORM 不同，表達式語言直接操作數據庫表和列，適用于復雜的查詢場景。

選擇

from sqlalchemy import select

# 構建查詢
stmt = select([User.name, User.age]).where(User.age > 20)

# 執行查詢
result = connection.execute(stmt)
for row in result:
    print(row)

插入

from sqlalchemy import insert

# 構建插入語句
stmt = insert(User).values(name='Bob', age=30)

# 執行插入
connection.execute(stmt)

更新

from sqlalchemy import update

# 構建更新語句
stmt = update(User).where(User.name == 'Alice').values(age=35)

# 執行更新
connection.execute(stmt)

刪除

from sqlalchemy import delete

# 構建刪除語句
stmt = delete(User).where(User.name == 'Bob')

# 執行刪除
connection.execute(stmt)

高級特性

事務管理

SQLAlchemy 提供了強大的事務管理功能，支持手動和自動事務管理。

# 手動事務管理
transaction = connection.begin()
try:
    connection.execute("INSERT INTO users (name, age) VALUES ('Alice', 25)")
    connection.execute("INSERT INTO users (name, age) VALUES ('Bob', 30)")
    transaction.commit()
except:
    transaction.rollback()
    raise

# 自動事務管理
with engine.begin() as connection:
    connection.execute("INSERT INTO users (name, age) VALUES ('Alice', 25)")
    connection.execute("INSERT INTO users (name, age) VALUES ('Bob', 30)")

事件與鉤子

SQLAlchemy 提供了事件系統，允許開發者在特定事件發生時執行自定義邏輯。

from sqlalchemy import event

# 定義事件處理函數
def before_insert_listener(mapper, connection, target):
    print("Before insert:", target.name)

# 注冊事件
event.listen(User, 'before_insert', before_insert_listener)

連接池

SQLAlchemy 默認使用連接池來管理數據庫連接，以提高性能和資源利用率。

from sqlalchemy.pool import QueuePool

# 創建帶連接池的引擎
engine = create_engine('sqlite:///example.db', poolclass=QueuePool, pool_size=5, max_overflow=10)

性能優化

SQLAlchemy 提供了多種性能優化手段，包括批量插入、延遲加載、預加載等。

# 批量插入
session.bulk_save_objects([
    User(name='Alice', age=25),
    User(name='Bob', age=30),
    User(name='Charlie', age=35)
])
session.commit()

# 延遲加載
user = session.query(User).first()
print(user.addresses)  # 延遲加載 addresses

# 預加載
user = session.query(User).options(joinedload(User.addresses)).first()
print(user.addresses)  # 預加載 addresses

常見問題與解決方案

1. 如何處理數據庫遷移？

SQLAlchemy 本身不提供數據庫遷移工具，但可以與 Alembic 結合使用來實現數據庫遷移。

pip install alembic
alembic init migrations

2. 如何處理多數據庫連接？

可以通過創建多個引擎實例來管理多個數據庫連接。

engine1 = create_engine('sqlite:///db1.db')
engine2 = create_engine('sqlite:///db2.db')

3. 如何處理復雜的查詢？

可以使用 SQLAlchemy 的表達式語言來構建復雜的查詢，或者使用原生 SQL 查詢。

from sqlalchemy.sql import text

stmt = text("SELECT * FROM users WHERE age > :age")
result = connection.execute(stmt, age=20)
for row in result:
    print(row)

總結

SQLAlchemy 是一個功能強大且靈活的 Python SQL 工具包和 ORM 框架。通過本文的介紹，你應該已經掌握了 SQLAlchemy 的核心概念、基本用法以及一些高級特性。無論是簡單的 CRUD 操作，還是復雜的查詢和事務管理，SQLAlchemy 都能提供強大的支持。希望本文能幫助你在實際項目中更好地使用 SQLAlchemy，提升開發效率和代碼質量。