pyorm-yongheng0912 0.1.0

A GORM-like Python ORM for MySQL.

PyORM

PyORM 是一个面向 MySQL 的 Python ORM，目标是提供接近 GORM 的使用体验：模型定义、链式查询、自动迁移、事务、Hooks、软删除、批量写入和 UPSERT。

当前安装包名、导入包名和 GitHub 仓库名都统一为 pyorm。

获取项目

从 GitHub 克隆仓库：

git clone https://github.com/yongheng0912/pyorm.git
cd pyorm

如果已经克隆过，更新到最新代码：

git pull

安装

建议先创建虚拟环境：

python -m venv .venv
.\.venv\Scripts\Activate.ps1

开发模式安装：

pip install -e .

普通安装：

pip install .

如果需要读取 YAML 配置：

pip install -e ".[yaml]"

也可以直接从 GitHub 安装：

pip install git+https://github.com/yongheng0912/pyorm.git

运行测试：

python -m unittest discover -s tests

提交代码前建议至少跑一次：

python -m unittest discover -s tests
python -m compileall pyorm tests examples

仓库说明

当前仓库地址：

https://github.com/yongheng0912/pyorm

项目里有一个本地设计稿 设计文档.txt，它只用于个人开发记录，已经加入 .gitignore，不会再提交到仓库。公开文档以本 README 为准。

快速开始

from pyorm import Model, fields, connect

db = connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="123456",
    database="test",
    charset="utf8mb4",
    echo=True,
)

class User(Model):
    __tablename__ = "users"

    id = fields.BigInt(primary_key=True, auto_increment=True)
    name = fields.String(64, nullable=False, index=True)
    age = fields.Int(default=0)
    email = fields.String(128, unique=True)
    status = fields.String(20, default="normal")
    created_at = fields.DateTime(auto_now_add=True)
    updated_at = fields.DateTime(auto_now=True)
    deleted_at = fields.DateTime(nullable=True, soft_delete=True)

db.auto_migrate(User)

user = User(name="Tom", age=18, email="[e-mail]")
db.create(user)

users = (
    db.query(User)
    .where(User.age > 18)
    .where(User.status == "normal")
    .order_by(User.id.desc())
    .limit(10)
    .all()
)

配置方式

代码配置

db = connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="123456",
    database="test",
    echo=True,
    pool_size=10,
    max_overflow=20,
    allow_add_column=True,
    allow_modify_column=False,
    allow_drop_column=False,
)

JSON/YAML 配置

from pyorm import connect_from_config

db = connect_from_config("pyorm.yaml")

YAML 示例：

database:
  host: 127.0.0.1
  port: 3306
  user: root
  password: "123456"
  database: test
  charset: utf8mb4

pool:
  pool_size: 10
  max_overflow: 20
  pool_recycle: 3600

migration:
  allow_create_table: true
  allow_add_column: true
  allow_modify_column: false
  allow_drop_column: false
  dry_run: false

logging:
  echo: true
  slow_query_ms: 500

完整示例可以参考仓库里的 pyorm.yaml.example。

配置项说明

上面的配置按用途分成四组：database、pool、migration、logging。它们最后都会被展平成 DBConfig，也就是说 YAML 里的：

database:
  host: 127.0.0.1
logging:
  echo: true

等价于代码里的：

connect(host="127.0.0.1", echo=True)

database：数据库连接

配置	类型	默认值	作用
host	str	127.0.0.1	MySQL 服务地址。
port	int	3306	MySQL 服务端口。
user	str	None	登录 MySQL 的用户名。真实连接时通常必填。
password	str	None	登录 MySQL 的密码。
database	str	None	默认数据库名。auto_migrate() 读取 information_schema 时必须提供。
charset	str	utf8mb4	连接字符集，推荐 utf8mb4，可以完整支持 emoji 和常见多语言字符。
connect_timeout	int	10	建立连接的超时时间，单位秒。
read_timeout	int	30	读取 MySQL 响应的超时时间，单位秒。
write_timeout	int	30	向 MySQL 写入请求的超时时间，单位秒。

pool：连接池

配置	类型	默认值	作用
pool_size	int	10	空闲连接池最多保留多少个连接，方便复用，减少频繁握手成本。
max_overflow	int	20	当空闲池没有连接时，最多允许额外创建多少个临时连接。
pool_timeout	int	30	获取连接时最多等待多少秒。超过后抛出超时异常。
pool_recycle	int	3600	连接最长复用时间，单位秒。超过后会丢弃旧连接并新建，避免 MySQL 断开长连接后继续使用坏连接。
ping_before_use	bool	True	每次复用连接前先 ping 一下，尽量提前发现断线。

migration：自动迁移

配置	类型	默认值	作用
auto_migrate_on_start	bool	False	预留配置：是否在启动时自动迁移。当前建议显式调用 db.auto_migrate()。
migration_mode	str	safe	迁移模式标记，当前主要作为配置语义保留，具体安全边界由下面的 allow 开关控制。
allow_create_table	bool	True	允许自动创建不存在的表。
allow_add_column	bool	True	允许给已有表新增字段。
allow_modify_column	bool	False	是否允许自动修改字段类型或 NULL 约束。默认关闭，因为可能锁表或导致数据截断。
allow_drop_column	bool	False	是否允许自动删除数据库里有、模型里没有的字段。默认关闭，避免误删数据。
allow_add_index	bool	True	允许自动新增索引。
allow_drop_index	bool	False	是否允许自动删除索引。默认关闭，避免影响查询性能或唯一约束。
dry_run	bool	False	只生成 SQL，不执行。适合上线前检查迁移语句。
record_migration	bool	True	执行迁移后写入 pyorm_schema_migrations，方便排查变更历史。

logging：SQL 日志

配置	类型	默认值	作用
echo	bool	False	是否打印执行的 SQL。开发环境建议打开，生产环境按日志策略决定。
echo_args	bool	True	打印 SQL 时是否同时打印参数。生产环境如果参数可能包含敏感信息，可以关闭。
pretty_sql	bool	False	预留配置：是否格式化 SQL 输出。
slow_query_ms	int	1000	慢查询阈值配置，当前作为日志配置保留，后续可用于慢查询告警。

behavior：行为安全和返回格式

这些配置也可以放在 YAML 顶层，或放在任意分组里，加载时会统一展平：

配置	类型	默认值	作用
strict_model	bool	True	模型初始化时遇到未定义字段是否报错。默认报错，避免字段拼写错误悄悄混入对象。
default_limit	int | None	None	预留配置：默认查询条数限制。
empty_update_protection	bool	True	禁止无条件批量更新，除非调用 allow_global_update()。
empty_delete_protection	bool	True	禁止无条件批量删除，除非调用 allow_global_update()。
return_dict	bool	False	查询结果是否直接返回字典列表。默认返回模型对象。
batch_size	int	1000	预留配置：批量写入时的批大小。

模型和字段

from pyorm import Model, fields, Index, UniqueIndex

class Order(Model):
    __tablename__ = "orders"
    __indexes__ = [
        Index("idx_user_status", "user_id", "status"),
        UniqueIndex("uk_order_no", "order_no"),
    ]

    id = fields.BigInt(primary_key=True, auto_increment=True)
    user_id = fields.BigInt(index=True)
    order_no = fields.String(64)
    status = fields.Enum("new", "paid", "closed", default="new")
    amount = fields.Decimal(precision=10, scale=2, default=0)
    payload = fields.JSON(nullable=True)

已支持字段：

• Int
• BigInt
• Bool
• String
• Text
• Enum
• DateTime
• Decimal
• JSON

建表和迁移

只创建不存在的表：

db.create_all(User, Order)

自动迁移：

db.auto_migrate(User, Order)

只查看 SQL，不执行：

sql_list = db.auto_migrate(User, dry_run=True)

默认迁移策略偏安全：

• 允许创建表。
• 允许新增字段。
• 允许新增索引。
• 默认不允许修改字段。
• 默认不允许删除字段。
• 默认不允许删除索引。

迁移会读取 MySQL information_schema，并在执行变更时写入 pyorm_schema_migrations。

CRUD

新增：

user = User(name="Tom", age=18, email="[e-mail]")
db.create(user)

保存：

db.save(user)

save() 会根据主键是否为空自动选择 insert 或 update。

按主键更新：

user.name = "Jerry"
db.update(user)

删除：

db.delete(user)

如果模型有 soft_delete=True 字段，默认执行软删除：

UPDATE users SET deleted_at = NOW() WHERE id = ?

强制物理删除：

db.delete(user, force=True)

查询

users = (
    db.query(User)
    .where(User.age >= 18)
    .where(User.name.like("T%"))
    .order_by(User.id.desc())
    .limit(10)
    .offset(0)
    .all()
)

支持表达式：

• ==
• !=
• >
• >=
• <
• <=
• like()
• in_()
• not_in()
• between()
• is_null()
• is_not_null()
• &
• |

查询第一条：

user = db.query(User).where(User.id == 1).first()

计数：

total = db.query(User).where(User.age > 18).count()

分页：

page = db.query(User).order_by(User.id.desc()).paginate(page=1, per_page=20)

分页返回：

{
    "items": [...],
    "page": 1,
    "per_page": 20,
    "total": 100,
    "pages": 5,
}

软删除查询：

db.query(User).with_deleted().all()
db.query(User).only_deleted().all()

批量更新：

db.query(User).where(User.status == "new").update(status="active")

批量删除：

db.query(User).where(User.status == "inactive").delete()

默认禁止无条件更新和删除。如果确实要全表操作：

db.query(User).allow_global_update().update(status="archived")
db.query(User).allow_global_update().delete(force=True)

批量插入和 UPSERT

批量插入：

db.bulk_create([
    User(name="A", age=10, email="[e-mail]"),
    User(name="B", age=12, email="[e-mail]"),
])

UPSERT：

db.upsert(
    User(id=1, name="Tom", age=18, email="[e-mail]"),
    update_fields=["name", "age"],
)

生成 MySQL：

INSERT ... ON DUPLICATE KEY UPDATE `name`=VALUES(`name`), `age`=VALUES(`age`)

事务

with db.transaction() as tx:
    tx.create(User(name="A", age=1, email="[e-mail]"))
    tx.create(User(name="B", age=2, email="[e-mail]"))

事务块内复用同一个连接。正常退出提交，抛异常回滚。

Hooks

class User(Model):
    id = fields.BigInt(primary_key=True, auto_increment=True)
    name = fields.String(64)

    def before_create(self, db):
        self.name = self.name.strip()

    def after_find(self):
        self.name = self.name.upper()

支持：

• before_create
• after_create
• before_update
• after_update
• before_save
• after_save
• before_delete
• after_delete
• after_find

当前边界

已实现同步 MySQL ORM 的主要功能，包括模型、字段、CRUD、链式查询、软删除、批量写入、UPSERT、事务、Hooks、自动迁移和迁移记录。

以下能力已有设计边界，但建议单独迭代：

• 异步版 aiomysql
• 读写分离
• 分库分表
• 复杂关联关系和预加载
• CLI 代码生成