async-pybatis-orm

一个面向 MySQL 异步场景的 Python ORM，API 风格参考 MyBatis-Plus。 适合希望在 Python 中使用类似 Wrapper + CRUD 体验的项目。

特性

• 异步优先：基于 asyncio + databases + aiomysql
• MyBatis-Plus 风格 API：select_* / save / update_by_id / remove_*
• 条件构造器：QueryWrapper、UpdateWrapper
• 内置分页：Page + PageResult
• Mixin 组合式设计，便于扩展

安装

pip install async-pybatis-orm

开发依赖：

pip install "async-pybatis-orm[dev]"

快速开始

1. 初始化数据库连接

from async_pybatis_orm.base.connection import DatabaseManager

await DatabaseManager.initialize(
    database_url="mysql+aiomysql://root:123456@127.0.0.1:3306/test_db"
)

2. 定义模型

from datetime import datetime
from typing import Optional

from async_pybatis_orm.base.common_model import CommonModel
from async_pybatis_orm.fields import Field, PrimaryKey


class User(CommonModel):
    __table_meta__ = {"table_name": "user", "primary_key": "id"}

    id: Optional[int] = PrimaryKey(column_name="id", auto_increment=True)
    username: str = Field(column_name="username", nullable=False, max_length=64)
    created_at: datetime = Field(default_factory=datetime.now, nullable=False)
    updated_at: datetime = Field(
        default_factory=datetime.now,
        auto_update=True,
        nullable=False,
    )

3. CRUD 示例

from async_pybatis_orm.wrapper.query_wrapper import QueryWrapper
from async_pybatis_orm.pagination.page import Page


# 插入
u = User(username="alice")
ok = await User.save(u)

# 按主键查询
u2 = await User.select_by_id(u.id)

# 条件查询
wrapper = QueryWrapper().like("username", "ali").order_by("id", desc=True)
users = await User.select_list(wrapper)

# 分页
page = Page(current=1, size=10)
result = await User.select_page(page, wrapper)
print(result.total, len(result.records))

# 更新
u.username = "alice_new"
updated = await User.update_by_id(u)

# 删除
removed = await User.remove_by_id(u.id)

4. 关闭连接

await DatabaseManager.close()

FastAPI 集成

完整可运行示例见 examples/fastapi_app.py，包含：

1. lifespan 生命周期初始化/关闭连接
2. CommonModel ORM 模型定义
3. Pydantic 请求/响应模型分离
4. CRUD、分页、事务提交与回滚示例

运行示例：

pip install async-pybatis-orm fastapi uvicorn
cd examples
python fastapi_app.py

访问文档：

• Swagger UI: http://127.0.0.1:8000/docs
• ReDoc: http://127.0.0.1:8000/redoc

关键 API 速查

模型 CRUD

• save(entity)
• insert(entity)
• insert_batch(entities) / batch_save(entities)
• select_by_id(id) / get_by_id(id)
• select_one(wrapper)
• select_list(wrapper) / list_by_condition(wrapper)
• select_count(wrapper) / count_by_condition(wrapper)
• select_page(page, wrapper) / page_query(page, wrapper)
• update_by_id(entity)
• update_by_wrapper(update_wrapper)
• remove_by_id(id)
• remove_by_ids(ids)
• remove_by_wrapper(wrapper)

条件构造器

QueryWrapper 常用方法：

• 比较：eq/ne/gt/ge/lt/le
• 模糊：like/not_like/like_left/like_right
• 范围：in_list/not_in/between/not_between
• 空值：is_null/is_not_null
• 其它：order_by/group_by/select/last

UpdateWrapper 常用方法：

• set(field, value)
• set_sql(field, expression)
• 搭配 eq/like/... 组成更新条件

导入说明

包根目录 async_pybatis_orm 当前默认导出：

• BaseModel
• CRUDModel
• CommonModel
• QueryWrapper
• DatabaseManager

其余对象请从子模块导入，例如：

• Field / PrimaryKey：async_pybatis_orm.fields
• Page：async_pybatis_orm.pagination.page
• PageResult：async_pybatis_orm.pagination.page_result
• UpdateWrapper：async_pybatis_orm.wrapper.query_wrapper

数据库 URL 格式

mysql+aiomysql://用户名:密码@主机:端口/数据库名

例如：

• mysql+aiomysql://root:123456@localhost:3306/sakila
• mysql+aiomysql://user:pass@127.0.0.1:3306/test_db

基于表自动生成 Model

支持通过 SDK 或脚本按表结构批量生成 ORM Model。

SDK 用法

from async_pybatis_orm import DBConfig, GenerateOptions, generate_models

db = DBConfig(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="123456",
    database="sakila",
)

options = GenerateOptions(
    output_dir="generated_models",
    tables=["actor", "film"],
    overwrite=False,
    file_layout="per_table",  # 默认一表一文件
    base_class="CommonModel",
)

result = generate_models(db, options)
print("generated:", len(result.generated))
print("skipped:", len(result.skipped))

CLI 用法

python scripts/generate_models.py \
  --host 127.0.0.1 \
  --port 3306 \
  --user root \
  --password 123456 \
  --database sakila \
  --output-dir generated_models \
  --tables actor,film \
  --overwrite

可选参数包括：

• --file-layout per_table|single_file
• --single-file-name models.py
• --base-class CommonModel|CRUDModel
• --filename-case snake|kebab|pascal
• --class-name-case pascal|camel
• --no-init（不自动更新 __init__.py 导出）

项目结构

async_pybatis_orm/
├── base/
│   ├── abstracts.py
│   ├── base_model.py
│   ├── common_model.py
│   ├── connection.py
│   ├── database_manager.py
│   ├── global_config.py
│   ├── integrated_model.py
│   └── sql_builder.py
├── crud/
│   ├── crud_model.py
│   ├── delete_mixin.py
│   ├── insert_mixin.py
│   ├── select_mixin.py
│   └── update_mixin.py
├── pagination/
│   ├── page.py
│   └── page_result.py
├── wrapper/
│   ├── base_wrapper.py
│   └── query_wrapper.py
└── fields.py

发布

发布流程与脚本请参考：

• PUBLISH.md
• PUBLISH_QUICKSTART.md
• PUBLISH_CMD_GUIDE.md

贡献

欢迎提 PR：

1. Fork 本仓库
2. 新建分支：git checkout -b feature/xxx
3. 提交改动
4. 推送并发起 Pull Request

许可证

MIT License，详见 LICENSE。

致谢

• MyBatis-Plus
• SQLAlchemy
• Tortoise ORM