vorm

English

Async ORM for Python 3.10+ — MySQL and PostgreSQL, 基于 Pydantic v2 开发, 以pydantic作为数据模型,同时增加额外字段属性作为数据库字段必要属性,从而可以操作数据库表结构.这个库以精简为主,不提供外键以及多对一,多对多等键功能

特性

Async-first — 所有 I/O 均为 async/await，基于 asyncmy（MySQL）和 asyncpg（PostgreSQL）
Pydantic 原生 — 模型即 BaseModel，享受类型校验、序列化等全部 Pydantic 功能
表达式 API — == != > < like() between() in_() json_path() 生成参数化 SQL
DDL 自动同步 — Model.sync_table() 建表、新增列、修改列、创建索引，只增改不删
JSON 原生支持 — dict/list 自动映射 JSON/JSONB，默认值使用数据库原生函数
时间戳行为 — timestamp_behavior="create"|"update"|"both" 控制自动时间戳
联合索引 — Meta.indexes 声明，Field(index=True) 简写，只增不删
断连自动重试 — 连接断开时自动重试 3 次（指数退避），非连接错误直接抛出
多表 JOIN — 显式 ON 条件，Row 对象支持属性/字典双模式访问
原生 SQL — execute_raw() / fetch_raw() / fetch_one_raw() 绕过 ORM 直接执行

安装

pip install vorm

数据库	驱动	说明
MySQL	`asyncmy`	连接池 + 事务
PostgreSQL	`asyncpg`	连接池 + 事务

字段示例

import datetime
from vorm import Field, Model
from vorm.ddl import Index
from enum import Enum

class FruitEnum(str, Enum):
    pear = 'pear'
    banana = 'banana'
    apple = 'apple'


class BaseField(Model):
    id_base:int|None=None

class AllFieldTypes(BaseField):
    """覆盖所有 Python 类型 → SQL 类型的映射。"""

    class Meta:
        table = "all_field_types"
        indexes = [Index(fields=("short_text",),unique=True),
                Index(fields=("age","score"))
                ]

    # -- 主键 ---------------------------------------------------------

    id: int = Field(primary_key=True, auto_increment=True)
    # MySQL    → `id` INT AUTO_INCREMENT PRIMARY KEY
    # PG       → "id" SERIAL PRIMARY KEY

    # -- 字符串 -------------------------------------------------------

    short_text: str = Field(max_length=200, nullable=True,default="",comment="短字符")
    # max_length 指定 → VARCHAR(100)

    free_text: str | None = None
    # 不指定 max_length → TEXT

    # -- 数值 ---------------------------------------------------------

    age: int | None = 0
    # MySQL → INT,  PG → INTEGER

    score: float = Field(default=0)
    # MySQL → DOUBLE,  PG → DOUBLE PRECISION

    price: decimal.Decimal | None = Field(default=0.0,db_type="decimal(15,5)")
    # MySQL → DECIMAL(18,6),  PG → DECIMAL(18,6)

    # -- 布尔 ---------------------------------------------------------

    is_active: bool = Field(default=True)
    # MySQL → BOOL,  PG → BOOLEAN

    # -- 日期时间 -----------------------------------------------------

    created_at: datetime.datetime = Field(
        default_factory=datetime.datetime.now,
        timestamp_behavior="create",  # INSERT 时写入，UPDATE 时跳过
    )
    # MySQL → DATETIME,  PG → TIMESTAMP

    updated_at: datetime.datetime = Field(
        default_factory=datetime.datetime.now,
        timestamp_behavior="both",  # INSERT / UPDATE 都会刷新
        index=True
    )

    event_date: datetime.date | None = None
    # MySQL → DATE,  PG → DATE

    # -- 二进制 -------------------------------------------------------

    payload: bytes | None = None
    # MySQL → BLOB,  PG → BYTEA

    # -- JSON ---------------------------------------------------------

    config: dict = Field(default_factory=dict)
    # MySQL → JSON,  PG → JSONB
    # DDL: MySQL DEFAULT (JSON_OBJECT()), PG DEFAULT jsonb_build_object()

    tags: list = Field(default_factory=list)
    # MySQL → JSON,  PG → JSONB
    # DDL: MySQL DEFAULT (JSON_ARRAY()), PG DEFAULT jsonb_build_array()
    new_field:list = Field(default_factory=list)
    
    enum_f:FruitEnum = Field(default=FruitEnum.banana)

快速开始

1. 定义模型

import datetime
from vorm import Field, Model, col
from vorm.ddl import Index

class Article(Model):
    class Meta:
        table = "articles"
        indexes = [
            Index(fields=("title",)),           # 单列索引
            Index(fields=("title", "created_at")),  # 联合索引
        ]

    id: int = Field(primary_key=True, auto_increment=True)
    title: str = Field(max_length=200)
    body: str                                 # → TEXT
    tags: list = Field(default_factory=list)  # → JSON/JSONB
    view_count: int = Field(default=0)
    created_at: datetime.datetime = Field(
        default_factory=datetime.datetime.now,
        timestamp_behavior="create",           # INSERT 写入，UPDATE 保留原值
    )
    updated_at: datetime.datetime = Field(
        default_factory=datetime.datetime.now,
        timestamp_behavior="both",             # INSERT + UPDATE 均刷新
    )

2. 连接数据库

from vorm.engine import create_mysql_engine, create_postgresql_engine

# Mysql 连接池（推荐）
mysql_engine = await engine.create_mysql_engine(host="127.0.0.1",port=3306,user="username",password="password",database="test",minsize=10,maxsize=20)
session = mysql_engine.session()

# PostgreSQL 连接池（推荐）
postgresql_engine = await engine.create_postgresql_engine(host="127.0.0.1",port=3306,user="username",password="password",database="test",min_size=50,max_size=100)
session = postgresql_engine.session()

3. 建表 & 索引

await Article.sync_table(session.dialect)

首次 → CREATE TABLE IF NOT EXISTS + CREATE INDEX。后续新增列/修改类型/新增索引分别合并为单条 ALTER TABLE / CREATE INDEX。永不删除列、表或索引。

4. CRUD

# INSERT
article = Article(title="Hello", tags=["tech"])
await session.save(article)          # 单条插入，id 自动回填
await session.bulk_insert([...])     # 批量插入

# SELECT
await session.query(Article).all()                             # 全表
await session.query(Article).filter(Article.view_count >= 10).all()  # 条件
await session.query(Article).filter(Article.id == 1).one()     # 精确一条
await session.query(Article).filter(Article.title.like("Post%")).first()  # 有则返无则 None
await session.query(Article).count()                            # 计数
await session.query(Article).filter(Article.id == 1).exists()  # 存在性
await session.query(Article).order_by(col(Article.id).desc()).limit(10).offset(20).all()  # 分页

# UPDATE
article.view_count += 1
await session.save(article)          # 主键不为 None → UPDATE
await session.query(Article).filter(Article.view_count == 0).update(view_count=1)  # 批量

# DELETE
await session.delete(article)
await session.query(Article).filter(Article.view_count == 0).delete()  # 批量

排序（ORDER BY）

用 col() 包裹类级字段再调用方法，可消除类型检查器对类级字段的误报：

from vorm import col

# 单字段
await session.query(Article).order_by(col(Article.id).desc()).all()
await session.query(Article).order_by(col(Article.id).asc()).all()

# 多字段：按传入顺序依次排序
await session.query(Article).order_by(
    col(Article.view_count).desc(),  # 先按浏览量降序
    col(Article.id).asc(),           # 再按 id 升序
).all()

# 裸字段默认升序，可与 desc()/asc() 混用
await session.query(Article).order_by(Article.status, col(Article.created_at).desc()).all()

运行时 Article.id 经元类拦截本就是 Column，Article.id.desc() 也能跑通；但静态类型检查器把类级字段注解为值类型 int，会报「int 无 desc 属性」。用 col(Article.id) 把它重新标注为 Column 即可消除误报。col() 对 .like()、.in_()、.between()、.is_null()、.json_path() 等所有 Column 方法同样适用。

5. 高级查询

# LIKE / BETWEEN
await session.query(User).filter(col(User.name).like("张%")).all()
await session.query(Book).filter(col(Book.price).between(10, 50)).all()

# IN / IS NULL
await session.query(User).filter(col(User.role).in_(["admin", "editor"])).all()
await session.query(User).filter(col(User.email).is_null()).all()

# AND / OR
cond = (User.age >= 18) & (User.role == "member")       # AND
cond = (User.role == "admin") | (User.role == "staff")   # OR

# JSON 路径查询（MySQL: column->>'$.path' / PG: column->'key'->>'sub'）
await session.query(Article).filter(col(Article.tags).json_path("$[0]") == "tech").all()
await session.query(Product).filter(col(Product.data).json_path("$.price") > 100).all()

6. 多表 JOIN

# 两表 INNER JOIN
rows = await session.query(Author).join(
    Book, on=Author.id == Book.author_id
).filter(Book.price > 20).order_by(col(Author.name).asc()).all()

# LEFT JOIN
rows = await session.query(Author).join(
    Book, on=Author.id == Book.author_id, type="LEFT"
).all()

# Row 访问
print(rows[0].authors.name)    # 属性
print(rows[0]["authors.id"])   # 字典 key
print(dict(rows[0]))           # 扁平字典

# select() 可用于查询指定字段,alias()可用于别名
d = await mysql_session.query(AllFieldTypes)\
    .join(JoinTable,on=AllFieldTypes.id==JoinTable.all_id,type="left")\
    .join(JoinTableSecond,on=JoinTable.id==JoinTableSecond.join_all_id,type="left")\
    .select(AllFieldTypes.id,AllFieldTypes.short_text,JoinTable.short_text.alias("j_s_t"),JoinTableSecond.short_text.alias("j_ss_t")).all()

class ReturnJoinTableSecond(Model):
    id: int|None

    short_text: str|None

    j_s_t: str|None
    
    j_ss_t:str|None


# 后续可以使用serialization提供的功能反序列化到pydantic模型中
from vorm.query import serialization
data = serialization(ReturnJoinTableSecond,d)

7. 事务 & 健康检查

# Session 事务
async with session.transaction():
    await session.save(alice)
    await session.save(bob)
    # 异常自动回滚，正常自动提交

# 健康检查
ok = await session.dialect.ping()  # True → 连接正常

8. 原生 SQL

await session.execute_raw("DELETE FROM users WHERE status = %s", ["inactive"])
rows = await session.fetch_raw("SELECT id, name FROM users WHERE age > %s", [18])
row = await session.fetch_one_raw("SELECT * FROM users WHERE id = %s", [1])

查看日志

import logging

# 看所有 SQL 语句
logging.getLogger("vorm").setLevel(logging.DEBUG)

# 只看 DDL 和重连信息
logging.getLogger("vorm").setLevel(logging.INFO)

# 配置输出格式
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)

模型定义参考

Field 参数

参数	类型	默认值	说明
`primary_key`	`bool`	`False`	主键
`auto_increment`	`bool`	`False`	自增（仅 int 主键）
`column_name`	`str \| None`	`None`	数据库列名，默认取 Python 属性名
`nullable`	`bool`	`True`	是否允许 NULL
`unique`	`bool`	`False`	唯一约束
`index`	`bool`	`False`	单列索引（自动转为 Index）
`max_length`	`int \| None`	`None`	字符串长度 → VARCHAR(n)；无 → TEXT
`db_type`	`str \| None`	`None`	直接指定数据库类型
`timestamp_behavior`	`"create" \| "update" \| "both" \| None`	`None`	自动时间戳行为
`default` / `default_factory`	—	—	Python 默认值（同时生成 DDL DEFAULT）

timestamp_behavior

值	INSERT	UPDATE
`"create"`	写入值	跳过，保留原值
`"update"`	写入值	重新计算当前时间
`"both"`	写入值	重新计算当前时间

Meta 配置

class Meta:
    table = "table_name"                          # 表名（必填）
    indexes = [                                    # 联合索引（可选）
        Index(fields=("col1", "col2")),            # 普通索引
        Index(fields=("col1",), unique=True),      # 唯一索引
        Index(fields=("col1", "col2"), name="idx_custom"),  # 自定义名称
    ]

索引名自动生成规则：ix_{table}_{fields}（普通），unq_{table}_{fields}（唯一）。

Python → SQL 类型映射

Python 类型	MySQL	PostgreSQL
`int`	`INT`	`INTEGER`
`int` + auto_increment PK	`INT AUTO_INCREMENT`	`SERIAL`
`str` + max_length	`VARCHAR(n)`	`VARCHAR(n)`
`str` 无 max_length	`TEXT`	`TEXT`
`bool`	`BOOL`	`BOOLEAN`
`float`	`DOUBLE`	`DOUBLE PRECISION`
`bytes`	`BLOB`	`BYTEA`
`datetime.datetime`	`DATETIME`	`TIMESTAMP`
`datetime.date`	`DATE`	`DATE`
`decimal.Decimal`	`DECIMAL(18,6)`	`DECIMAL(18,6)`
`dict`	`JSON`	`JSONB`
`list`	`JSON`	`JSONB`

开发

uv sync                     # 安装依赖
uv run pytest               # 运行测试（112 个）
uv run mypy vorm/           # 类型检查
uv run ruff check vorm/     # 代码风格

设计原则

参数化查询 — 所有 SQL 参数绑定，永不拼接用户输入
方言隔离 — MySQL / PG 差异封装在 Dialect 子类，业务层无感
安全 DDL — 只增不改不删（ADD COLUMN + ALTER COLUMN + CREATE INDEX，不 DROP）
Pydantic 优先 — 模型定义、校验、序列化完全复用 Pydantic 生态
数据库原生优先 — JSON 默认值使用 JSON_OBJECT() / jsonb_build_object() 等原生函数，不依赖代码序列化

License

Apache License

Metadata

Release files for vorm 0.1.5

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for vorm 0.1.5
File	Size	Uploaded
vorm-0.1.5.tar.gz	180.5 kB	Sep 11, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for vorm 0.1.5
File	Interpreter	ABI	Platform	Reset
vorm-0.1.5-py3-none-any.whl 44.0 kB Sep 11, 2026	Python 3	none	any	Details

Total release size: 224.5 kB

Release files / vorm-0.1.5.tar.gz

Download URL	vorm-0.1.5.tar.gz
Size	180.5 kB
Tags	Source
SHA-256 checksum How to use checksums	`2038f26e854f5bd97eae79343accc10f99a90b68095dcc6d61d9580259ac1956`
BLAKE2b-256 checksum How to use checksums	`6d8aee96dc87c975ed36c861442edea4bd99f6fa34f00b3e448628af2dad52b5`
Upload date	Sep 11, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.8.3`

Release files / vorm-0.1.5-py3-none-any.whl

Download URL	vorm-0.1.5-py3-none-any.whl
Size	44.0 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`512cc00051dc8f75499738144c40b6536d73f59364c7c876280e159087117fc2`
BLAKE2b-256 checksum How to use checksums	`92f3fc2e4a5c7a2c2b3870cf75930bed3d060beef2ea92f0538f99d236ffbae9`
Upload date	Sep 11, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.8.3`