SQLAlchemy 在 Python 中的同步与异步操作及CRUD实现指南
在 Python 开发中，数据库操作是构建应用不可或缺的一部分。SQLAlchemy 是一个强大的 SQL 工具包和对象关系映射（ORM）库，它提供了丰富的接口来操作数据库，既支持同步操作也支持异步操作。本文将详细介绍如何使用 SQLAlchemy 进行同步和异步数据库操作，并展示如何使用 SQLAlchemy 进行基本的 CRUD（创建、读取、更新、删除）操作。
SQLAlchemy 简介
SQLAlchemy 提供了 SQL 工具和对象关系映射（ORM），能够将数据库表映射为 Python 类，表中的行映射为类实例，列映射为实例属性。这样，开发者就可以使用 Python 代码来操作数据库，而无需编写大量的 SQL 语句。
安装 SQLAlchemy
首先，你需要安装 SQLAlchemy。可以通过 pip 安装：
bash复制代码
pip install sqlalchemy |
对于异步操作，还需要安装 sqlalchemy[asyncio] 额外包：
bash复制代码
pip install sqlalchemy[asyncio] |
同步操作
定义模型
在 SQLAlchemy 中，你需要定义一个或多个模型（Model），这些模型对应于数据库中的表。
python复制代码
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey | |
from sqlalchemy.ext.declarative import declarative_base | |
from sqlalchemy.orm import sessionmaker | |
Base = declarative_base() | |
class User(Base): | |
__tablename__ = 'users' | |
id = Column(Integer, primary_key=True) | |
name = Column(String) | |
age = Column(Integer) | |
# 创建数据库引擎（以 SQLite 为例） | |
engine = create_engine('sqlite:///example.db', echo=True) | |
Base.metadata.create_all(engine) | |
# 创建会话 | |
Session = sessionmaker(bind=engine) | |
session = Session() |
CRUD 操作
创建（Create）
python复制代码
new_user = User(name='Alice', age=30) | |
session.add(new_user) | |
session.commit() |
读取（Read）
python复制代码
user = session.query(User).filter_by(name='Alice').first() | |
print(user.name, user.age) |
更新（Update）
python复制代码
user.age = 31 | |
session.commit() |
删除（Delete）
python复制代码
session.delete(user) | |
session.commit() |
异步操作
SQLAlchemy 1.4 及以上版本支持异步操作。使用异步时，需要引入 asyncio 和异步的数据库引擎。
异步定义模型
异步模型定义与同步相同，但操作方式不同。
异步 CRUD 操作
首先，创建一个异步的数据库引擎和会话。
python复制代码
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine | |
# 创建异步数据库引擎 | |
async_engine = create_async_engine('sqlite+aiosqlite:///example.db', echo=True) | |
# 创建异步会话 | |
async_session = AsyncSession(bind=async_engine) | |
# 注意：以下操作需要在异步函数中执行 | |
async def async_crud(): | |
# 异步创建 | |
async with async_session.begin(): | |
new_user = User(name='Bob', age=25) | |
await async_session.add(new_user) | |
# 异步读取 | |
async with async_session.begin(): | |
user = await async_session.execute(select(User).filter_by(name='Bob')).scalars().first() | |
print(user.name, user.age) | |
# 异步更新 | |
async with async_session.begin(): | |
user.age = 26 | |
await async_session.commit() | |
# 异步删除 | |
async with async_session.begin(): | |
await async_session.delete(user) | |
await async_session.commit() | |
# 运行异步函数 | |
import asyncio | |
asyncio.run(async_crud()) |
结论
通过 SQLAlchemy，无论是同步还是异步操作，都可以非常方便地进行数据库的 CRUD 操作。同步操作适用于传统的阻塞式 I/O 场景，而异步操作则更适合需要高并发处理能力的现代 Web 应用和微服务。在实际开发中，可以根据应用的需求和性能要求来选择合适的操作方式。