Metadata-Version: 2.4
Name: easyorm-py
Version: 1.0.27
Summary: easy_orm
Requires-Python: >=3.6
Description-Content-Type: text/markdown
Dynamic: description
Dynamic: description-content-type
Dynamic: requires-python
Dynamic: summary

﻿# 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

## 安装

从源码安装：

```bash
pip install .
```

本地开发模式安装：

```bash
pip install -e .
```

## 快速开始

### 1. 初始化数据库配置

```python
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 配置示例

```python
from easy_orm import DBConfig, EasyOrmConfig

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

EasyOrmConfig(config)
```

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

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

```python
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

```python
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

```python
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)` | 批量更新映射数据 |

分页返回结构：

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

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

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

## SQL 装饰器

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

```python
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): ...
```

### 返回类型映射

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

```python
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 文件。

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

EasyOrmConfig(config)
```

XML 示例：

```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 类：

```python
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 片段

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

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

### 动态更新

```xml
<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>
```

## 切换数据源

### 使用装饰器切换

```python
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` 表示该方法会在指定数据源上开启事务。方法执行成功时提交，抛出异常时回滚。

### 使用上下文管理器切换

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

### 使用原生 Session

```python
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：

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

## 参数占位符

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

推荐使用：

```sql
select * from user where id = #{id}
```

谨慎使用：

```sql
select * from user order by ${order_by}
```

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

## EasyOrmConfig 单例说明

`EasyOrmConfig` 是单例配置对象。

```python
EasyOrmConfig(config)
```

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

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

```python
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？

初始化配置时设置：

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

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

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

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

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

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

## 注意事项

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