当前位置：首页 > news >正文

SQLAlchemy 在 Python 中的同步与异步操作及CRUD实现指南

news 2025/8/19 15:47:27

在 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())`