在 Python 开发中，数据库操作是构建应用不可或缺的一部分。SQLAlchemy 是一个强大的 SQL 工具包和对象关系映射（ORM）库，它提供了丰富的接口来操作数据库，既支持同步操作也支持异步操作。本文将详细介绍如何使用 SQLAlchemy 进行同步和异步数据库操作，并展示如何使用 SQLAlchemy 进行基本的 CRUD（创建、读取、更新、删除）操作。

SQLAlchemy 简介 SQLAlchemy 提供了 SQL 工具和对象关系映射（ORM），能够将数据库表映射为 Python 类，表中的行映射为类实例，列映射为实例属性。这样，开发者就可以使用 Python 代码来操作数据库，而无需编写大量的 SQL 语句。

安装 SQLAlchemy 首先，你需要安装 SQLAlchemy。可以通过 pip 安装：

bashpip install sqlalchemy 对于异步操作，还需要安装 sqlalchemy[asyncio] 额外包：

bashpip install sqlalchemy[asyncio]同步操作定义模型在 SQLAlchemy 中，你需要定义一个或多个模型（Model），这些模型对应于数据库中的表。

pythonfrom 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）pythonnew_user = User(name='Alice', age=30)

session.add(new_user)

session.commit()读取（Read）pythonuser = session.query(User).filter_by(name='Alice').first()

print(user.name, user.age)更新（Update）pythonuser.age = 31

session.commit()删除（Delete）pythonsession.delete(user)

session.commit()异步操作 SQLAlchemy 1.4 及以上版本支持异步操作。使用异步时，需要引入 asyncio 和异步的数据库引擎。

异步定义模型异步模型定义与同步相同，但操作方式不同。

异步 CRUD 操作首先，创建一个异步的数据库引擎和会话。

pythonfrom 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 应用和微服务。在实际开发中，可以根据应用的需求和性能要求来选择合适的操作方式。