Python生态数据库ORM：SQLModel、SQLAlchemy、Peewee、Django ORM、Tortoise-ORM、PynamoDB、ormar

本文汇总几款Python生态下的数据库ORM客户端框架（类库）。

选型参考

SQLModel

项目主页即官方文档，可把数据库表映射成 Python 类的开源（GitHub，18.2K Star，882 Fork）小工具，SQLAlchemy+Pydantic，底层靠type hint（类型提示）搞定模型定义，写代码时能得到IDE的自动补全和即时报错，感觉像在写普通的Python类，却能直接对数据库增删改查。

核心特点：

• 统一模型：同一个模型类同时作为ORM模型和Pydantic模型（序列化、校验）
• FastAPI原生集成：与FastAPI的自动请求校验和API文档生成完美配合
• SQLAlchemy底层：底层基于SQLAlchemy，继承其稳定性和功能
• 代码量少：消除ORM模型和Pydantic模型之间的重复定义

传统开发痛点

实战

基于pip安装

python -m venv venv
source venv/bin/activate
pip install sqlmodel

示例

from sqlmodel import Field, SQLModel, Session, create_engine, select

class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None

# SQLite
engine = create_engine("sqlite:///heroes.db")
SQLModel.metadata.create_all(engine)

with Session(engine) as sess:
sess.add_all([
Hero(name="Deadpond", secret_name="Dive Wilson"),
Hero(name="Spider‑Boy", secret_name="Pedro Parqueador"),
Hero(name="Rusty‑Man", secret_name="Tommy Sharp", age=48)
])
sess.commit()

with Session(engine) as sess:
stmt = select(Hero).where(Hero.name == "Spider‑Boy")
hero = sess.exec(stmt).first()
print(hero)

最佳实践

• 字段默认值：用Field(default=…, nullable=True)控制是否可以为NULL
• 分页：直接在select()上使用.offset()、.limit()，和原生SQLAlchemy完全一致
• 事务：推荐使用with Session(engine) as sess:上下文管理器，自动提交或回滚
• 迁移：配合Alembic（SQLAlchemy官方迁移工具）即可管理表结构变更

SQLAlchemy

官网，Python ORM的事实标准，生态最成熟、功能最全面；开源（GitHub，12K Star，1.7K Fork），官方文档。

核心特点：

• 双模式设计：Core（SQL表达式语言）和ORM（对象关系映射）两种使用方式
• 企业级特性：连接池、事务管理、迁移（Alembic）、多数据库支持
• 异步支持：1.4版本开始原生支持asyncio，引入sqlalchemy.ext.asyncio
• 声明式映射：基于类的声明式模型定义

实战

基于pip安装：pip install SQLAlchemy

示例：

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, Session

Base = declarative_base()
class User(Base):
__tablename__ = 'users'

id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
email = Column(String, unique=True)
posts = relationship('Post', back_populates='author')

class Post(Base):
__tablename__ = 'posts'

id = Column(Integer, primary_key=True)
title = Column(String)
user_id = Column(Integer, ForeignKey('users.id'))
author = relationship('User', back_populates='posts')

# 使用
engine = create_engine('sqlite:///blog.db')
Base.metadata.create_all(engine)

with Session(engine) as session:
user = User(name='Alice', email='alice@example.com')
session.add(user)
session.commit()

Peewee

轻量、简洁、学习成本低的开源（GitHub，12K Star，1.4K Fork）ORM，适合小型项目和脚本，官方文档，中文文档。

特性：

• 数据库：支持SQLite、MySQL、MariaDB、PG
• 生态集成：asyncio、Flask、FastAPI、PyDantic、Bottle、Pyramid等
• 零配置：无需复杂的配置，几行代码即可开始使用
• SQLite优先：对SQLite有出色支持，适合嵌入式场景

适用场景：小型Web应用、数据脚本、嵌入式项目、快速原型

缺点：缺乏高级特性（如复杂的连接池管理、异步原生支持弱），不适合大型企业应用

实战

基于pip安装：pip install peewee

示例：

from peewee import SqliteDatabase, Model, CharField, IntegerField, ForeignKeyField

db = SqliteDatabase('blog.db')

class User(Model):
name = CharField()
email = CharField(unique=True)

class Meta:
database = db

class Post(Model):
title = CharField()
user = ForeignKeyField(User, backref='posts')

class Meta:
database = db

db.connect()
db.create_tables([User, Post])

# 使用
alice = User.create(name='Alice', email='alice@example.com')
Post.create(title='Hello World', user=alice)

Django ORM

Django框架内置的ORM，与Django生态深度绑定，Django项目的唯一选择。

核心特点：

• 自动迁移：make migrations + migrate自动管理数据库Schema变更
• Admin后台：基于模型的自动管理界面
• 查询API丰富：链式查询、懒加载、预取，如select_related/prefetch_related
• 信号机制：模型生命周期钩子

实战

基于pip安装：pip install Django

示例：

from django.db import models

class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
created_at = models.DateTimeField(auto_now_add=True)

class Meta:
db_table = 'users'

class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
created_at = models.DateTimeField(auto_now_add=True)

# 使用
users = User.objects.filter(name__startswith='A').order_by('-created_at')
posts = Post.objects.select_related('author').filter(author__email='alice@example.com')

Tortoise-ORM

项目主页，专为asyncio设计的开源（GitHub，5.6K Star，501 Fork）ORM，语法灵感来自Django ORM。

核心特点：

• 完全异步：从底层到API全部基于async/await设计
• Django风格：类Django的模型定义和查询API，Django用户容易上手
• 自动迁移：内置Aerich迁移工具
• 多数据库支持：SQLite、PostgreSQL、MySQL、SQL Server、Oracle

适用场景：FastAPI、Sanic、Quart等异步Web框架项目

缺点：生态不如SQLAlchemy成熟，社区规模较小

实战

基于pip安装：

# 默认SQLite
pip install tortoise-orm
# PostgreSQL + psycopg
pip install tortoise-orm[psycopg]
# PostgreSQL + asyncpg
pip install tortoise-orm[asyncpg]
# MySQL + aiomysql
pip install tortoise-orm[aiomysql]
# MySQL + asyncmy
pip install tortoise-orm[asyncmy]
# MS SQL Server，Oracle
pip install tortoise-orm[asyncodbc]

示例：

from tortoise import Tortoise, fields
from tortoise.models import Model

class User(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100)
email = fields.CharField(max_length=200, unique=True)
posts = fields.ReverseRelation['Post']

class Post(Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=200)
user = fields.ForeignKeyField('models.User', related_name='posts')

# 初始化
await Tortoise.init(
db_url='sqlite://blog.db',
modules={'models': ['__main__']}
)
await Tortoise.generate_schemas()

# 使用
alice = await User.create(name='Alice', email='alice@example.com')
posts = await Post.filter(user=alice).order_by('-id')

MongoEngine

使用Django风格语法、开源（GitHub，4.3K Star，1.2K Fork）成熟的MongoDB ODM，支持引用、嵌入文档、聚合管道，官方文档。

PonyORM

​以Python生成器表达式作为查询语法的独特开源（GitHub，3.8K Star，256 Fork）ORM。

PyPi最后一个版本0.7.19停留在24年8月27日。

核心特点：

• 生成器查询：使用Python的for循环和生成器表达式写查询，翻译成SQL
• 在线ER图：提供Web界面的实体关系图编辑器
• 自动缓存：内置身份映射缓存，减少数据库查询

实战

基于pip安装：pip install pony

示例：

from pony.orm import *

db = Database()
class User(db.Entity):
name = Required(str)
email = Required(str, unique=True)
posts = Set('Post')

class Post(db.Entity):
title = Required(str)
user = Required(User)

db.bind('sqlite', 'blog.db')
db.generate_mapping(create_tables=True)

# 使用——生成器语法
@db_session
def query_users():
users = select(u for u in User if 'alice' in u.name.lower())
for user in users:
print(user.name, user.email)

PynamoDB

Amazon DynamoDB是AWS提供的高性能NoSQL数据库服务，具有低延迟、高可扩展性等特点，被广泛应用于各种云应用中。DynamoDB的原生API相对底层，使用起来较为繁琐。

PynamoDB是一个专为DynamoDB设计的开源（GitHub，2.7K Star，429 Fork）Python库，提供类似Django ORM的Pythonic接口，让开发者能够以面向对象的方式操作DynamoDB，官方文档。

特性：

• Pythonic接口：提供类似ORM的面向对象API，简化DynamoDB操作
• 模型定义：通过Python类定义表结构和属性
• 自动类型转换：自动处理Python类型与DynamoDB类型的转换
• 查询构建器：提供链式查询API，支持复杂查询条件
• 索引支持：完整支持全局二级索引和本地二级索引
• 批量操作：支持批量读写操作，提高性能
• 事务支持：支持DynamoDB事务操作
• 连接池管理：自动管理连接，优化性能

实战

基于pip安装：pip install pynamodb

在PynamoDB中，通过继承Model类来定义数据模型，每个模型对应DynamoDB中的一张表。模型类中定义的属性对应表的字段，需要指定属性类型和是否为主键，定义好模型后，可使用create_table方法创建表。

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute

class UserModel(Model):
"""用户模型"""
class Meta:
table_name = 'users'
region = 'us-west-2'

# 定义主键
user_id = UnicodeAttribute(hash_key=True)
# 定义属性
username = UnicodeAttribute()
age = NumberAttribute()

# 创建表（如果不存在）
ifnot UserModel.exists():
UserModel.create_table(
read_capacity_units=5,
write_capacity_units=5,
wait=True
)
print('表创建成功')

提供简洁的API来插入和查询数据：

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute

# 插入数据
user = UserModel(
user_id='user001',
username='张三',
age=25
)
user.save()
print('用户创建成功')

# 查询单条数据
try:
user = UserModel.get('user001')
print(f'用户名: {user.username}')
print(f'年龄: {user.age}')
except UserModel.DoesNotExist:
print('用户不存在')

当需要查询多条记录或根据非主键字段查询时，可使用scan方法扫描表，提供丰富的过滤条件，可根据属性值进行筛选，虽然扫描操作会遍历整个表，性能相对较低，但对于小型表或不频繁的查询来说是可接受的。

# 扫描所有用户
for user in UserModel.scan():
print(f'{user.username}')

# 使用过滤条件
for user in UserModel.scan(UserModel.age > 20):
print(f'{user.username}的年龄是{user.age}')

# 多个过滤条件
for user in UserModel.scan(
(UserModel.age > 20) & (UserModel.username.contains('张'))
):
print(f'找到用户: {user.username}')

全局二级索引（GSI）可使用非主键字段进行高效查询。在模型中定义索引类，指定索引的键和投影属性，就可使用索引进行查询。索引查询比扫描操作快得多，特别适合需要频繁查询的场景。

from pynamodb.models import Model
from pynamodb.indexes import GlobalSecondaryIndex, AllProjection
from pynamodb.attributes import UnicodeAttribute, NumberAttribute

class EmailIndex(GlobalSecondaryIndex):
"""邮箱索引"""
class Meta:
index_name = 'email-index'
read_capacity_units = 5
write_capacity_units = 5
projection = AllProjection()

email = UnicodeAttribute(hash_key=True)

class UserModel(Model):
class Meta:
table_name = 'users'
region = 'us-west-2'

user_id = UnicodeAttribute(hash_key=True)
username = UnicodeAttribute()
age = NumberAttribute()
email = UnicodeAttribute()
email_index = EmailIndex()

# 使用索引查询
for user in UserModel.email_index.query('zhangsan@example.com'):
print(f'找到用户: {user.username}')

当需要处理大量数据时，批量操作可显著提高性能，提供batch_write和batch_get方法来批量插入、更新和查询数据。批量操作会自动处理DynamoDB的批量限制，将大批量操作分割成多个小批次执行。

from pynamodb.models import Model

# 批量写入
with UserModel.batch_write() as batch:
for i in range(100):
user = UserModel(
user_id=f'user{i:03d}',
username=f'用户{i}',
age=20 + i % 30,
email=f'user{i}@example.com'
)
batch.save(user)

print('批量写入完成')

# 批量读取
user_ids = ['user001', 'user002', 'user003']
for user in UserModel.batch_get(user_ids):
print(f'{user.username}: {user.email}')

提供多种方式来更新和删除数据。可先查询记录，修改属性后保存，也可使用update方法进行原子更新。删除操作同样简单，调用delete方法即可，原子更新操作特别适合需要保证数据一致性的场景，如计数器增减、条件更新等。

# 更新数据
user = UserModel.get('user001')
user.age = 26
user.save()
print('用户信息已更新')
# 原子更新（增加年龄）
user.update(actions=[
UserModel.age.set(UserModel.age + 1)
])
# 条件更新
user.update(
actions=[UserModel.email.set('newemail@example.com')],
condition=(UserModel.age > 20)
)
# 删除数据
user = UserModel.get('user001')
user.delete()
print('用户已删除')

DynamoDB支持事务操作，PynamoDB也提供相应的API。

from pynamodb.transactions import TransactWrite

# 事务写入
with TransactWrite(connection=UserModel._get_connection()) as transaction:
# 创建新用户
user1 = UserModel(
user_id='user100',
username='李四',
age=30,
email='lisi@example.com'
)
transaction.save(user1)
# 更新现有用户
user2 = UserModel.get('user001')
user2.age = 27
transaction.save(user2)
# 删除用户
user3 = UserModel.get('user002')
transaction.delete(user3)

print('事务执行成功')

Beanie

官网，开源（GitHub，2.7K Star，303 Fork）基于Pydantic的异步MongoDB ODM，与FastAPI原生集成。

实战

基于pip安装：pip install beanie

示例：

from beanie import Document, init_beanie
from pydantic import BaseModel

class User(Document):
name: str
email: str
class Settings:
name = 'users'

await init_beanie(database=mongo_db, document_models=[User])
alice = await User(name='Alice', email='alice@example.com').insert()

ormar

项目主页，集成Pydantic、开源（GitHub，1.8K Star，98 Fork）异步迷你ORM库，支持PG、MySQL和SQLite；直接在数据库模型上做Pydantic验证。

底层用SQLAlchemy core做查询构建，用databases库做异步数据库支持，可用Alembic做迁移管理。

查询不到数据时，会抛NoMatch异常；查到多条时抛MultipleMatches。支持不抛异常。

官方提供SQLAlchemy迁移工具sqlalchemy-to-ormar。

实战

基于pip安装：pip install ormar

示例：

import ormar
import sqlalchemy
from ormar import DatabaseConnection
from typing import Optional

DATABASE_URL = "sqlite+aiosqlite:///db.sqlite"
base_ormar_config = ormar.OrmarConfig(
metadata=sqlalchemy.MetaData(),
database=DatabaseConnection(DATABASE_URL),
)

# 定义模型
class Author(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="authors")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
address: Optional[Address] = ormar.ForeignKey(Address)
age: int = ormar.Integer(minimum=0) # 这既是数据库约束，也是Pydantic验证

# 创建表（开发环境用，生产用Alembic）
engine = sqlalchemy.create_engine(DATABASE_URL.replace('+aiosqlite', ''))
base_ormar_config.metadata.create_all(engine)

# 创建
tolkien = await Author.objects.create(name="J.R.R. Tolkien")
# 查询
book = await Book.objects.get(title="The Hobbit")
# 更新
await book.update()
authors = await Author.objects.all()
# 带条件的查询
young_authors = await Author.objects.filter(age__lt=30).all()
# 关联查询
book_with_author = await Book.objects.select_related("author").get(id=1)
await book.delete()
author = await Author.objects.get_or_none(name="不存在的人")
if author is None:
print("没找到")

关系加载要手动：外键字段默认只存ID，不会自动加载关联对象，得调load()或select_related。