easyorm-py 1.0.27

easy_orm

easy_orm

easy_orm 是一个基于 SQLAlchemy 的轻量级 ORM 辅助库，提供通用 Mapper、多数据源切换、SQL 装饰器和类 MyBatis XML Mapper 能力。

功能特性

• 基于 BaseMapper[T] 的通用增删改查
• 支持多个命名数据源
• 支持通过装饰器或上下文管理器切换数据源
• 支持 @select_one、@select_list、@select_page、@insert、@update、@delete 等 SQL 装饰器
• 支持类 MyBatis 的 XML Mapper
• 支持 SQLAlchemy ORM 2.0 查询语句
• 支持 MySQL 和 SQLite

环境要求

• Python >= 3.10
• SQLAlchemy
• Pydantic
• 使用 MySQL 时需要 PyMySQL

安装

从源码安装：

pip install .

本地开发模式安装：

pip install -e .

快速开始

1. 初始化数据库配置

from easy_orm import DBConfig, EasyOrmConfig

config = DBConfig(
    primary="default",
    raw_sql=True,
    mapper_xml_path="mapper_xml",
    default={
        "db_type": "mysql",
        "host": "127.0.0.1",
        "port": 3306,
        "user": "root",
        "password": "123456",
        "database": "template",
    },
    test={
        "db_type": "mysql",
        "host": "127.0.0.1",
        "port": 3306,
        "user": "root",
        "password": "123456",
        "database": "template_test",
    },
)

EasyOrmConfig(config)

参数说明：

参数	说明
primary	默认数据源名称
raw_sql	是否打印 SQLAlchemy 执行日志
mapper_xml_path	XML Mapper 文件目录
default / test	自定义数据源名称

如果只配置了一个数据源，可以不传 primary，框架会自动使用唯一的数据源作为默认数据源。

2. SQLite 配置示例

from easy_orm import DBConfig, EasyOrmConfig

config = DBConfig(
    primary="default",
    default={
        "db_type": "sqlite",
        "sqlite_path": "./example.db",
    },
)

EasyOrmConfig(config)

3. 动态添加和移除数据源

多数据源是有意累加的。后续添加新数据源时，不会清空已有数据源。

from easy_orm import DataSource, EasyOrmConfig

EasyOrmConfig.add_datasource_config(
    "reporting",
    DataSource(
        db_type="mysql",
        host="127.0.0.1",
        port=3306,
        user="root",
        password="123456",
        database="reporting",
    ),
)

EasyOrmConfig.remove_datasource("reporting")

例如，先初始化了 default，再添加 reporting，最终两个数据源都会保留。只有显式调用 remove_datasource(...) 时，数据源才会被移除。

定义 SQLAlchemy Model

from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(64))
    nickname: Mapped[str] = mapped_column(String(64))

BaseMapper[T] 中的泛型 T 要传入 SQLAlchemy ORM Model，例如 BaseMapper[User]。

使用 BaseMapper

from sqlalchemy import select
from easy_orm import BaseMapper, Page
from model.user import User


class UserMapper(BaseMapper[User]):
    pass


mapper = UserMapper()

user_id = mapper.insert({"username": "tom", "nickname": "Tom"})
user = mapper.select_by_id(user_id)

mapper.update_by_id({"id": user_id, "nickname": "Tommy"})
mapper.delete_by_id(user_id)

users = mapper.select_all(select(User).where(User.id > 1))
one_user = mapper.select_one(select(User).where(User.username == "tom"))

page_result = mapper.select_page_by_orm(
    select(User).where(User.id > 1),
    Page(1, 10),
)

常用方法：

方法	说明
insert(data)	插入一条数据，返回自增主键
save_batch(data)	批量插入数据
add(data)	添加一个 SQLAlchemy ORM 对象
add_all(data)	添加多个 SQLAlchemy ORM 对象
select_by_id(id)	根据主键查询
select_one(statement=None)	查询一条数据
select_all(statement=None)	查询多条数据
select_page_by_orm(statement, page)	使用 SQLAlchemy Select 分页查询
select_page_by_sql(sql_str, sql_param, page)	使用原生 SQL 分页查询
select_all_by_sql(sql_str, sql_param)	使用原生 SQL 查询列表
update_by_id(data, selective=False)	根据 data["id"] 更新数据
update(update_statement, data=None)	执行 SQLAlchemy update 语句
delete_by_id(id)	根据主键删除
delete(delete_statement)	执行 SQLAlchemy delete 语句
bulk_update_mappings(table_model, data_list)	批量更新映射数据

分页返回结构：

{
    "list": [],
    "total": 0,
}

XML 分页和原生 SQL 分页会额外包含：

{
    "pageNum": 1,
    "pageSize": 10,
    "total": 0,
    "list": [],
}

SQL 装饰器

#{name} 会被转换成 SQL 绑定参数，推荐优先使用这种写法。

from easy_orm import BaseMapper, Page, select_one, select_list, select_page, insert, update, delete
from model.user import User


class UserMapper(BaseMapper[User]):

    @select_one("select id, username, nickname from user where id = #{id}")
    def find_by_id(self, id: int) -> dict: ...

    @select_list("select id, username, nickname from user where username like #{username}")
    def find_by_username(self, username: str) -> list[dict]: ...

    @select_page("select id, username, nickname from user where id > #{min_id}")
    def page_by_min_id(self, min_id: int, page: Page) -> list[dict]: ...

    @insert("insert into user(username, nickname) values (#{username}, #{nickname})")
    def add_user(self, username: str, nickname: str): ...

    @update("update user set nickname = #{nickname} where id = #{id}")
    def update_nickname(self, id: int, nickname: str): ...

    @delete("delete from user where id = #{id}")
    def delete_user(self, id: int): ...

返回类型映射

如果函数声明了返回类型，查询结果会尝试转换成对应类型。

class UserMapper(BaseMapper[User]):

    @select_one("select id, username, nickname from user where id = #{id}")
    def find_model_by_id(self, id: int) -> User: ...

    @select_list("select id, username, nickname from user")
    def find_models(self) -> list[User]: ...

常见返回类型：

返回类型	说明
dict	返回单条字典结果
list[dict]	返回字典列表
User	返回 ORM Model / Pydantic Model 对象
list[User]	返回对象列表

XML Mapper

初始化配置中传入 mapper_xml_path 后，框架会扫描该目录下的 XML 文件。

config = DBConfig(
    primary="default",
    mapper_xml_path="mapper_xml",
    default={
        "db_type": "sqlite",
        "sqlite_path": "./example.db",
    },
)

EasyOrmConfig(config)

XML 示例：

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "">
<mapper namespace="UserMapper">

    <select id="find_by_ids">
        select id, username, nickname
        from user
        where id in
        <foreach collection="id_list" item="id" open="(" close=")" separator=",">
            #{id}
        </foreach>
    </select>

    <select id="find_by_condition" resultType="model.user.User">
        select id, username, nickname
        from user
        <where>
            <if test="params.get('id') is not None">
                id = #{id}
            </if>
            <if test="params.get('username') is not None">
                and username = #{username}
            </if>
        </where>
    </select>

    <select id="page_users">
        select id, username, nickname from user
    </select>

    <insert id="insert_user">
        insert into user(username, nickname) values (#{username}, #{nickname})
    </insert>

    <update id="update_user">
        update user set nickname = #{nickname} where id = #{id}
    </update>

    <delete id="delete_user">
        delete from user where id = #{id}
    </delete>

</mapper>

对应的 Mapper 类：

from easy_orm import BaseMapper, Page
from model.user import User


class UserMapper(BaseMapper[User]):

    def find_by_ids(self, id_list: list[int]) -> list[dict]: ...

    def find_by_condition(self, id: int | None = None, username: str | None = None) -> list[User]: ...

    def page_users(self, page: Page) -> list[dict]: ...

    def insert_user(self, username: str, nickname: str): ...

    def update_user(self, id: int, nickname: str): ...

    def delete_user(self, id: int): ...

XML 约定：

• namespace 必须和 Mapper 类名一致，例如 UserMapper
• XML 中的 id 必须和 Mapper 方法名一致，例如 find_by_ids
• resultType 可以指定返回对象类型，例如 model.user.User
• 方法参数会作为 XML 动态 SQL 的参数来源
• Page 类型参数会被识别为分页参数，不会作为 SQL 参数传入

当前支持的 XML 标签：

标签	说明
select	查询语句
insert	插入语句
update	更新语句
delete	删除语句
sql	可复用 SQL 片段
include	引入 SQL 片段
if	条件判断
where	自动拼接 WHERE，并处理开头的 AND / OR
trim	自定义前缀、后缀和覆盖规则
set	更新语句中的动态 SET
choose / when / otherwise	分支判断
foreach	遍历集合，常用于 IN 查询

复用 SQL 片段

<sql id="user_columns">
    id, username, nickname
</sql>

<select id="find_all">
    select <include refid="user_columns" /> from user
</select>

动态更新

<update id="update_selective">
    update user
    <set>
        <if test="params.get('username') is not None">
            username = #{username},
        </if>
        <if test="params.get('nickname') is not None">
            nickname = #{nickname},
        </if>
    </set>
    where id = #{id}
</update>

切换数据源

使用装饰器切换

from easy_orm import datasource


class UserService:

    @datasource("test")
    def query_from_test(self):
        return UserMapper().select_by_id(1)

    @datasource("test", transactional=True)
    def update_from_test(self):
        UserMapper().update_by_id({"id": 1, "nickname": "new name"})

transactional=True 表示该方法会在指定数据源上开启事务。方法执行成功时提交，抛出异常时回滚。

使用上下文管理器切换

with UserMapper.switch_datasource("test"):
    UserMapper().select_by_id(1)

使用原生 Session

from sqlalchemy import text
from easy_orm import db_session


with db_session("default") as session:
    row = session.execute(text("select * from user where id = :id"), {"id": 1}).first()
    print(row._asdict() if row else None)

也可以通过 Mapper 获取 Session：

with UserMapper.db_session("default") as session:
    row = session.execute(text("select * from user where id = :id"), {"id": 1}).first()

参数占位符

写法	说明	推荐程度
#{param}	转换成 SQL 绑定参数	推荐
${param}	直接字符串替换	谨慎使用

推荐使用：

select * from user where id = #{id}

谨慎使用：

select * from user order by ${order_by}

${param} 适合表名、字段名、排序字段等无法使用绑定参数的位置，但必须确保参数来自可信白名单。

EasyOrmConfig 单例说明

EasyOrmConfig 是单例配置对象。

EasyOrmConfig(config)

第一次调用会初始化 ORM。后续再次调用 EasyOrmConfig(...) 时，会返回已有实例，不应该重复执行初始化逻辑。

多数据源注册是累加行为：

EasyOrmConfig(config_main)
EasyOrmConfig.add_datasource_config("test", test_config)

执行后，main 和 test 应该同时存在。后添加的 test 不应该冲掉已有的 main。

如果测试或 demo 需要干净状态，应该单独提供 reset/testing API，不应该改变正常的多数据源累加行为。

常见问题

1. 为什么 XML 方法没有执行？

请检查：

• DBConfig.mapper_xml_path 是否正确
• XML 文件是否在该目录下
• XML 的 namespace 是否等于 Mapper 类名
• XML 中的 id 是否等于 Mapper 方法名

2. 为什么默认数据源不对？

请检查 DBConfig(primary="...") 是否指定了正确的数据源名称。如果只配置一个数据源，可以不传 primary。

3. 如何查看执行 SQL？

初始化配置时设置：

config = DBConfig(
    primary="default",
    raw_sql=True,
    default={...},
)

4. #{} 和 ${} 有什么区别？

• #{} 会使用 SQL 绑定参数，更安全
• ${} 会直接字符串替换，需要调用方保证安全

5. XML 中的 <if test="..."> 怎么写？

可以通过 params 访问方法参数：

<if test="params.get('username') is not None">
    and username = #{username}
</if>

注意事项

• 推荐优先使用 #{param} 传参。
• ${param} 是直接字符串替换，不要直接传入用户输入。
• XML 动态表达式会被解析执行，因此 XML 应视为可信项目代码。
• EasyOrmConfig 是单例，不要依赖重复实例化来重新初始化配置。
• 多数据源是累加注册，不是状态污染。