hframe 0.1.3

分层架构后端框架 — Handler → Service → Repository

hframe

Huey's Framework — 分层架构后端框架
版本：0.1.0 | 作者：Huey | Python >= 3.8 | MIT 协议

---

目录

1. 项目概述
2. 安装
3. 快速开始
4. 配置管理
5. 架构详解
   • 5.1 分层架构总览
   • 5.2 Registry 注册表
6. Repository 层
   • 6.1 BaseRepository
   • 6.2 ListFilters 结构化查询
   • 6.3 VersionRepository
7. Service 层
   • 7.1 GenericService
   • 7.2 ServiceConfig 配置
   • 7.3 ServiceHooks 钩子体系
   • 7.4 工厂方法 new_service()
8. Handler 层
   • 8.1 GenericHandler
   • 8.2 HandlerConfig 配置
   • 8.3 级联关系
   • 8.4 引用解析
   • 8.5 级联展开深度控制
   • 8.6 HandlerHooks
   • 8.7 工厂方法 new_handler()
   • 8.8 响应数据映射与裁剪
9. View 层（Django 可选）
   • 9.1 BaseView
   • 9.2 ModuleView 与自动装配
   • 9.3 CRUD 视图
   • 9.4 H5 视图
10. 基础设施层
    • 10.1 MySQL（PyMySQL）
    • 10.2 Redis
    • 10.3 微信
11. 工具层
12. 异常体系
13. 日志
14. 完整示例：用户管理模块
15. 公共 API

---

1. 项目概述

hframe 是一个通用后端分层框架，核心理念是 Handler → Service → Repository 三层架构：

层	职责	特点
Handler	HTTP 无关的业务编排	级联、事务、权限、引用解析
Service	单实体生命周期	before/do/after 钩子管线、审计字段、唯一性校验、版本管理
Repository	数据访问唯一出口	标准 CRUD、结构化查询、软/硬删除

核心能力：

能力	说明
工厂模式	GenericHandler.new_handler("user") 一行创建整条链路
Registry 注册表	全局注册实体配置，驱动工厂自动装配
View 自动装配	View 中 self.sn = "user" 即可自动创建 Handler
级联操作	创建/删除时自动处理子表联动
引用解析	Get/List 时自动解析外键为完整对象
钩子管线	每个操作 before → do → after，可任意覆写
版本管理	内置版本字段映射，支持发布/回滚
Django 可选	View 层依赖 Django，核心层无 Web 框架绑定

设计原则：

• HTTP 无关：Handler/Service/Repository 不依赖任何 Web 框架，可用于 Django、FastAPI、CLI 等
• 配置驱动：行为通过 Config 数据类控制，而非硬编码
• 钩子优先：用 ServiceHooks / HandlerHooks 覆写行为，不用继承

---

2. 安装

# 核心层（无第三方依赖）
pip install hframe

# 带 Django View 层
pip install hframe[django]

# 带 MySQL + Redis 基础设施
pip install hframe[mysql]
pip install hframe[redis]

# 全部安装
pip install hframe[all]

可选依赖分组：

分组	包	用途
django	Django >= 2.0	View 层
mysql	DBUtils, PyMySQL, sshtunnel	MySQL 数据库
redis	redis, sshtunnel	Redis 缓存
wechat	requests	微信 API
infra	以上全部	完整基础设施
all	django + infra	全量安装

---

3. 快速开始

3.1 极简模式：5 行代码写 CRUD

import hframe
from hframe import Registry, GenericHandler, ListView, EditView, DeleteView

# 1. 初始化基础设施
from hframe.infra import MySQLWithSSH, RedisClient
dao = MySQLWithSSH(mysql_host="localhost", mysql_user="root", 
                   mysql_password="pass", mysql_database="my_db")
cache = RedisClient(redis_host="localhost")

# 2. 注册全局 DAO/Cache
Registry.set_defaults(dao=dao, cache=cache)

# 3. 注册实体
Registry.register("user", table="sys_user")

# 4. 工厂创建 Handler
handler = GenericHandler.new_handler("user")

# 5. CRUD
handler.create({"username": "zhangsan", "email": "[e-mail]"})
count, users = handler.list({"page": 1, "page_size": 20})
handler.update(1, {"email": "[e-mail]"})
handler.delete([1])

3.2 Django View 模式

# views.py
from hframe import ListView, EditView, DeleteView

class UserListView(ListView):
    def set_config(self):
        self.sn = "user"  # 一行配置，自动创建 Handler

class UserEditView(EditView):
    def set_config(self):
        self.sn = "user"

class UserDeleteView(DeleteView):
    def set_config(self):
        self.sn = "user"

# urls.py
from django.urls import path
from .views import UserListView, UserEditView, UserDeleteView

urlpatterns = [
    path('api/user/list/', UserListView.as_view()),
    path('api/user/edit/', UserEditView.as_view()),
    path('api/user/delete/', UserDeleteView.as_view()),
]

---

4. 配置管理

框架通过 Config 类集中管理配置，支持从 YAML 文件加载并深度合并默认值。

4.1 Config 类

from hframe import Config

# 读取配置
secret = Config.BASE_INFO['secret_key']
db_host = Config.MYSQL_INFO['host']

# 快捷取值（支持点号路径）
secret = Config.get('BASE_INFO.secret_key')

# 从 YAML 加载（覆盖默认值）
Config.load('config.yaml')

# 重置为默认
Config.reset()

配置分区：

分区	用途
BASE_INFO	应用基础信息（密钥、Token过期时间、管理员角色等）
MYSQL_INFO	MySQL 连接配置
REDIS_INFO	Redis 连接配置
SSH_INFO	SSH 隧道配置
WECHAT_INFO	微信应用配置

4.2 config.yaml

在项目根目录创建 config.yaml：

BASE_INFO:
  secret_key: "your-secret-key"
  auth_token_expire_time: 7200
  admin_role_id: 1

MYSQL_INFO:
  host: "127.0.0.1"
  port: 3306
  user: "db_user"
  password: "db_password"
  database: "my_database"
  pool_size: 5

REDIS_INFO:
  host: "127.0.0.1"
  port: 6379
  password: "redis-password"
  db: 0
  max_connections: 50

SSH_INFO:
  host: "10.0.0.1"
  port: 22
  username: "root"
  password: "ssh-password"

WECHAT_INFO:
  applet:
    app_id: "wx_applet_appid"
    app_secret: "wx_applet_secret"
  public_account:
    app_id: "wx_pa_appid"
    app_secret: "wx_pa_secret"

加载时机：在应用入口调用 Config.load() 或由 View 层自动加载。

---

5. 架构详解

5.1 分层架构总览

请求 → View(可选) → Handler → Service → Repository → DB
         ↑              ↑          ↑          ↑
      HTTP I/O      业务编排    实体生命周期   数据访问
      认证/权限     级联/事务    钩子/校验    CRUD/查询

数据流向：

• 写操作：raw_data → Handler.before → Handler.do(构造CrudRequest + 调用Service + 级联) → Handler.after → result
• 读操作：query → Handler.before → Handler.do(调用Service + 引用解析) → Handler.after → result
• Service 内部：before → do(审计字段 + 校验 + 持久化) → after

5.2 Registry 注册表

Registry 是全局实体注册表，供工厂方法使用：

from hframe import Registry, ServiceConfig, HandlerConfig

# 注册实体
Registry.register("user", table="sys_user",
    service_config=ServiceConfig(
        enable_unique_validation=True,
        unique_fields=[["username"]],
        create_time_field="create_time",
        update_time_field="update_time",
    ),
    handler_config=HandlerConfig(
        entity_name="user",
        cascades=[...],
        references=[...],
    )
)

# 设置全局默认
Registry.set_defaults(dao=mysql_client, cache=redis_client)

# 查询
cfg = Registry.get("user")       # EntityConfig 或 None
Registry.has("user")             # bool
Registry.all_names()             # ["user", ...]

# 测试清理
Registry.clear()

EntityConfig 字段：

字段	类型	默认值	说明
name	str	必填	实体名（唯一标识）
table	str	等于 name	数据库表名
service_config	ServiceConfig	None	Service 配置
handler_config	HandlerConfig	None	Handler 配置
repo_class	type	BaseRepository	自定义 Repository 类
service_class	type	GenericService	自定义 Service 类
handler_class	type	GenericHandler	自定义 Handler 类

---

6. Repository 层

6.1 BaseRepository

所有数据库操作必须通过 Repository 层，Service 不直接操作 DAO。

from hframe import BaseRepository

# 创建实例
repo = BaseRepository(dao=mysql_client, table="sys_user")

# CRUD
record = repo.get(1)                                  # 按主键查
record = repo.get_by_field("username", "zhangsan")    # 按字段查
count, records = repo.list(search_dict={"status": 1}, offset=0, size=20, order_by="id DESC")

new_id = repo.create({"username": "zhangsan"})        # 插入，返回主键
repo.batch_create([{"username": "u1"}, {"username": "u2"}])  # 批量插入
repo.update(1, {"email": "[e-mail]"})             # 按主键更新
repo.update_by_fields({"status": 0}, {"role": "admin"})  # 按条件更新

repo.delete(1)          # 删除（智能判断软删除/硬删除）
repo.delete([1, 2, 3])  # 批量删除

# 辅助
repo.exists({"username": "zhangsan"})  # 是否存在
total = repo.count({"status": 1})      # 计数
pk = repo.pk_field                     # 主键字段名（延迟加载）

智能删除：如果表有 is_del 字段，自动执行软删除（设 is_del=1），否则硬删除。

6.2 ListFilters 结构化查询

推荐使用 ListFilters 代替原始 search_dict：

from hframe import ListFilters, Filter, FilterOp

filters = ListFilters(
    page=1,
    page_size=20,
    filters=[
        Filter(field="status", op=FilterOp.EQ, value=1),
        Filter(field="age", op=FilterOp.GTE, value=18),
        Filter(field="name", op=FilterOp.LIKE, value="%张%"),
        Filter(field="department_id", op=FilterOp.IN, value=[1, 2, 3]),
        Filter(field="price", op=FilterOp.BETWEEN, value=[100, 500]),
    ],
    logic="and",
    order_by="id",
    order_dir="desc",
)

count, results = repo.list(filters=filters)

FilterOp 操作符：

操作符	值	SQL 等价
EQ	"eq"	field = value
NEQ	"neq"	field != value
LIKE	"like"	field LIKE value
GT	"gt"	field > value
GTE	"gte"	field >= value
LT	"lt"	field < value
LTE	"lte"	field <= value
IN	"in"	field IN (values)
BETWEEN	"between"	field >= v1 AND field <= v2

6.3 VersionRepository

继承 BaseRepository，增加版本管理相关的查询方法。适用于需要版本控制的实体（如配置、规则）。

---

7. Service 层

7.1 GenericService

每个实体类型对应一个 Service 实例，负责单实体完整生命周期。

from hframe import GenericService, ServiceConfig

# 手动创建
repo = BaseRepository(dao, "sys_user")
svc = GenericService(repo, ServiceConfig(
    enable_unique_validation=True,
    unique_fields=[["username"]],
    create_time_field="create_time",
    update_time_field="update_time",
))
svc.set_user(user_id=1)

# CRUD
result = svc.create({"username": "zhangsan", "email": "[e-mail]"})
count, users = svc.list({"page": 1, "page_size": 20})
user = svc.get(1)
svc.update(1, {"email": "[e-mail]"})
svc.delete([1])

# 版本管理（需启用 version_mode）
svc.activate(version_id)
versions = svc.list_versions("CONFIG_001")
svc.edit_version(version_id, {"version_remark": "修改说明"})

操作管线：每个 CRUD 操作都经过 before → do → after 三阶段：

create:  before_create → do_create → after_create
update:  before_update → do_update → after_update
delete:  before_delete → do_delete → after_delete
get:     before_get    → do_get    → after_get
list:    before_list   → do_list   → after_list

内置 before 行为：

操作	自动处理
before_create	设置默认值、审计字段（create_time/create_user）、版本字段、唯一性校验
before_update	获取旧记录、合并数据、审计字段（update_time/update_user）、唯一性校验
before_delete	校验 ID 列表
do_create	调用 Repository.create
do_update	调用 Repository.update（版本模式：旧行退位 + 新行插入）
do_delete	调用 Repository.delete

7.2 ServiceConfig 配置

from hframe import ServiceConfig, VersionFieldMapping

cfg = ServiceConfig(
    # 唯一性校验
    enable_unique_validation=True,
    unique_fields=[["username"], ["email"]],  # 每组内字段组合唯一

    # 审计字段（自动填充）
    create_time_field="create_time",
    update_time_field="update_time",
    create_user_field="create_user_id",
    update_user_field="update_user_id",

    # 软删除
    delete_field="is_del",

    # 版本管理
    version_mode=True,
    version_fields=VersionFieldMapping(
        ulid_field="ulid",
        code_field="code",
        version_field="version_code",
        current_field="is_current",
        status_field="version_status",
        parent_field="parent_ulid",
    ),

    # 操作日志
    enable_op_log=True,
    entity_name="用户",
)

7.3 ServiceHooks 钩子体系

通过钩子覆写任意阶段的行为，无需继承：

from hframe import GenericService, ServiceHooks

hooks = ServiceHooks()

# 覆写 before_create：添加自定义校验
def my_before_create(service, input_list):
    for item in input_list:
        if len(item.get("username", "")) < 3:
            raise ValidationException("用户名至少3个字符")
    return input_list

hooks.before_create = my_before_create

# 覆写 after_create：发送通知
def my_after_create(service, result):
    for record in result:
        send_notification(record)
    return result

hooks.after_create = my_after_create

# 注入
svc = GenericService(repo, config)
svc.set_hooks(hooks)

钩子签名约定：

before_xxx(service, *args) → 处理后的参数
do_xxx(service, *args) → 执行结果
after_xxx(service, result, *args) → 最终返回

7.4 工厂方法 new_service()

从 Registry 自动创建 Repository + Service：

from hframe import GenericService

# 前提：Registry 已注册
Registry.register("user", table="sys_user",
    service_config=ServiceConfig(enable_unique_validation=True, unique_fields=[["username"]])
)

# 一行创建
svc = GenericService.new_service("user")
# 等价于：
# repo = BaseRepository(dao, "sys_user")
# svc = GenericService(repo, service_config)

---

8. Handler 层

8.1 GenericHandler

Handler 是 HTTP 无关的业务编排层，在 Service 之上处理跨实体逻辑。

from hframe import GenericHandler, HandlerConfig

# 手动创建
svc = GenericService(repo, service_config)
handler = GenericHandler(svc, HandlerConfig(entity_name="user"))

# CRUD（自动处理认证/权限 + 钩子管线）
handler.create({"username": "zhangsan"}, context=request_context)
user = handler.get(GetRequest(id=1), context=request_context)
count, users = handler.list({"page": 1}, context=request_context)
handler.update(1, {"email": "[e-mail]"}, context=request_context)
handler.delete([1], context=request_context)

# 访问底层 Service
service = handler.service()

context 参数：可选的请求上下文 dict，供认证/权限钩子使用：

context = {
    "user_id": 1,
    "user_info": {...},
    "auth_token": "xxx",
    "ip": "127.0.0.1",
    "trace": "trace-id",
}

8.2 HandlerConfig 配置

from hframe import HandlerConfig, CascadeRelation, ReferenceRelation, ChildRefRelation

config = HandlerConfig(
    entity_name="order",

    # 级联关系（向下：父→子）
    cascades=[
        CascadeRelation(
            child_table="order_item",
            parent_fk_field="items",
            child_fk_field="order_id",
            show_name="items",
            on_delete="cascade",  # cascade | soft_delete | restrict
        ),
    ],

    # 向上引用（子→父）
    references=[
        ReferenceRelation(
            parent_table="product",
            fk_field="product_id",
            parent_pk_field="id",
            show_name="product",
        ),
    ],

    # 向下子引用（父→子 FK 列表）
    child_refs=[
        ChildRefRelation(
            child_table="tag",
            fk_list_field="tag_ids",
            child_pk_field="id",
            show_name="tags",
        ),
    ],

    # 级联展开深度控制
    max_expand_depth=3,                          # 全局最大递归展开深度（默认 1 = 展开一层不递归）
    field_depth_limits={"product_id": 1},       # 单字段深度上限：product 只展开 1 层
    field_stop_rules={                          # 字段级截止规则
        "product_id": [
            StopRule(on_handler="product", field="category", stop=True),   # 不展开 category
        ],
    },

    # 响应数据映射与裁剪
    response_exclude=["is_deleted", "password_hash"],  # 排除敏感字段
    # response_mapper=lambda e: {"id": e["id"], "name": e["name"]},  # 可选：自定义映射

    # 认证/权限钩子
    auth=lambda ctx: get_user_from_token(ctx.get("auth_token")),
    perm=lambda user, resource, action: check_permission(user, resource, action),
)

8.3 级联关系

级联关系控制父子表联动，在 Create 和 Delete 时自动触发。

CascadeRelation 参数：

参数	说明
child_table	子表名
parent_fk_field	父表中指向子表的字段名
child_fk_field	子表中指向父表的外键字段名
show_name	返回数据中的键名
on_delete	删除策略：cascade（级联删除）/ soft_delete（软删除）/ restrict（禁止删除）

级联创建流程：

1. handler.create(data) → Service.create() → 主表插入
2. 检查 data 中是否有 cascade.show_name 对应的子数据
3. 自动设置 child_fk_field = parent_id
4. child_handler.create(child_data)

级联删除策略：

策略	行为
cascade	子表记录一同删除
soft_delete	子表记录设 is_del=1
restrict	如果有子记录，抛出异常阻止删除

8.4 引用解析

引用解析在 Get/List 时自动将外键字段解析为完整对象。

向上引用（ReferenceRelation）：

# 订单表中 product_id → 解析为完整的 product 对象
references=[
    ReferenceRelation(
        parent_table="product",
        fk_field="product_id",     # 本表外键
        parent_pk_field="id",      # 父表主键
        show_name="product",       # 挂载到结果的键名
    )
]

# 查询结果自动包含：
# {"id": 1, "product_id": 5, "product": {"id": 5, "name": "商品A", ...}}

向下子引用（ChildRefRelation）：

# 文章表 tag_ids: [1,3,5] → 解析为完整 tag 对象列表
child_refs=[
    ChildRefRelation(
        child_table="tag",
        fk_list_field="tag_ids",   # 本表中存放子ID列表的字段
        child_pk_field="id",
        show_name="tags",
    )
]

# 查询结果自动包含：
# {"id": 1, "tag_ids": [1,3,5], "tags": [{"id":1,"name":"Python"}, ...]}

8.5 级联展开深度控制

Handler 的引用解析（References / ChildRefs）支持递归展开——被解析出的父/子对象，如果自身也有引用关系，会继续递归解析。深度控制机制提供精细化的递归管理。

8.5.1 三层控制机制

控制层	配置项	作用
全局深度	max_expand_depth	所有字段统一的最大递归层数
单字段深度	field_depth_limits	对特定字段设定不同的深度上限
字段截止规则	field_stop_rules	对子 Handler 中的指定字段完全跳过或展开一层后截止

默认行为：未设置任何深度控制时，展开一层不递归（max_expand_depth=1）。

8.5.2 全局深度 — max_expand_depth

from hframe import HandlerConfig

config = HandlerConfig(
    entity_name="user",
    max_expand_depth=3,  # 递归展开最多 3 层
    references=[
        ReferenceRelation(parent_table="department", fk_field="dept_ulid", show_name="dept"),
    ],
)

• max_expand_depth=1：展开一层，不递归（默认）
• max_expand_depth=2：展开一层，被展开的对象再递归一层
• max_expand_depth=N：递归 N 层

8.5.3 单字段深度 — field_depth_limits

对某个字段单独限制递归层数，覆盖全局 max_expand_depth。

场景：User 通过 dept_ulid 引用 Department，Department 自引用 parent_ulid。默认会递归到顶层部门，但只需直属部门信息。

from hframe import HandlerConfig

config = HandlerConfig(
    entity_name="user",
    max_expand_depth=5,                         # 全局允许 5 层
    field_depth_limits={"dept_ulid": 1},        # 但 dept 只展开 1 层
    references=[
        ReferenceRelation(parent_table="department", fk_field="dept_ulid", show_name="dept"),
    ],
)

HTTP 降级覆盖：

GET /api/v1/users/get?id=xxx&fdepth=dept_ulid:1

• 格式：字段:深度，多字段逗号分隔（fdepth=a:1,b:2）
• HTTP 的 fdepth 只能降级（≤ 服务端配置值），不能放大

8.5.4 字段截止规则 — field_stop_rules

对子 Handler 中的指定字段进行截止控制。

StopRule 结构：

from hframe import StopRule

# 完全跳过（- 前缀 = stop=True）
StopRule(on_handler="department", field="manager", stop=True)

# 展开一层后截止（无 - 前缀 = stop=False）
StopRule(on_handler="department", field="parent_id", stop=False)

场景：User 引用 Department，Department 有 manager 和 parent_id 两个引用。展开 dept_ulid 时：

• 跳过 manager（不需要经理信息）
• parent_id 只展开一层（拿到父部门名称即可）

from hframe import HandlerConfig, StopRule, ReferenceRelation

config = HandlerConfig(
    entity_name="user",
    max_expand_depth=3,
    field_stop_rules={
        "dept_ulid": [
            StopRule(on_handler="department", field="manager", stop=True),    # 不查 manager
            StopRule(on_handler="department", field="parent_id", stop=False), # parent 只展开 1 层
        ],
    },
    references=[
        ReferenceRelation(parent_table="department", fk_field="dept_ulid", show_name="dept"),
    ],
)

HTTP 覆盖：

GET /api/v1/users/get?fstop=dept_ulid=-department:manager,department:parent_id

• 格式：字段=规则列表，多条规则逗号分隔
• -handler:field = 完全跳过
• handler:field = 展开一层后截止
• 可传多条 fstop 参数

8.5.5 优先级

解析顺序（优先级从高到低）：

1. fieldLimitMap（父 Handler 通过 build_field_ctx 注入）→ 最先检查
2. 字段级 FieldDepthLimits（当前 Handler 配置）
3. 全局 MaxExpandDepth（当前 Handler 配置）
4. 默认值 → 展开一层不递归

8.5.6 上下文传递流程

HTTP 请求
  │
  ├─ ?depth=N ──────────────→ inject_expand_params()  → context._expand_depth
  ├─ ?fdepth=field:depth ──→ inject_expand_params()  → context._fd_override
  └─ ?fstop=field=rules ───→ inject_expand_params()  → context._fs_override
         │
         ▼
  Handler.get() / Handler.list()
         │
         ├─ _prepare_expand_context() 解析参数注入 context
         │
         ├─ _resolve_references / _resolve_child_refs
         │      │
         │      ├─ effective_expand_depth(ctx, field, config)
         │      │      ├─ fieldLimitMap 命中 → 按 Stop 规则返回
         │      │      └─ 否则 → 使用 depth 值
         │      │
         │      └─ depth > 1 时 → build_field_ctx() 构建子 context
         │                      → 递归调用子 Handler 的 resolve 方法
         │
         └─ 返回结果

8.5.7 完整示例

from hframe import (
    Registry, HandlerConfig, ServiceConfig,
    ReferenceRelation, StopRule,
)

# 注册 User（带深度控制）
Registry.register("user", table="sys_user",
    handler_config=HandlerConfig(
        entity_name="user",
        max_expand_depth=3,
        field_depth_limits={"dept_ulid": 2},
        field_stop_rules={
            "dept_ulid": [
                StopRule(on_handler="department", field="manager", stop=True),
                StopRule(on_handler="department", field="parent_id", stop=False),
            ],
        },
        references=[
            ReferenceRelation(
                parent_table="department",
                fk_field="dept_ulid",
                parent_pk_field="ulid",
                show_name="dept",
            ),
        ],
    ),
)

# 注册 Department（带自引用）
Registry.register("department", table="sys_department",
    handler_config=HandlerConfig(
        entity_name="department",
        max_expand_depth=3,
        references=[
            ReferenceRelation(
                parent_table="user",
                fk_field="manager_ulid",
                parent_service_name="user",
                show_name="manager",
            ),
            ReferenceRelation(
                parent_table="department",
                fk_field="parent_ulid",
                parent_service_name="department",
                show_name="parent",
            ),
        ],
    ),
)

# 查询结果示例：
# GET /api/v1/users/get?id=xxx
# {
#   "id": 1, "name": "张三", "dept_ulid": "D001",
#   "dept": {                              # ← 展开 dept（2层深度）
#     "ulid": "D001", "name": "技术部",
#     "parent_ulid": "D000",
#     "parent": {                          # ← stop=False，展开 1 层
#       "ulid": "D000", "name": "总公司",
#       "parent_ulid": null
#       # ← 不再递归 parent
#     },
#     # ← manager 被跳过（stop=True）
#   }
# }

8.6 HandlerHooks

与 ServiceHooks 类似，Handler 层也支持钩子覆写：

from hframe import HandlerHooks

hooks = HandlerHooks()

# 覆写 do_create：添加自定义逻辑
hooks.do_create = lambda handler, input_list: handler.do_create(input_list)

# 覆写 before_delete：删除前检查
def check_before_delete(handler, ids):
    # 检查是否允许删除
    if has_active_orders(ids):
        raise ValidationException("存在关联订单，无法删除")
    return ids

hooks.before_delete = check_before_delete

handler.set_hooks(hooks)

8.7 工厂方法 new_handler()

从 Registry 一行创建整条链路（Repository → Service → Handler）：

from hframe import GenericHandler

# 前提：Registry 已注册
Registry.register("order", table="sys_order",
    handler_config=HandlerConfig(
        entity_name="order",
        cascades=[CascadeRelation(...)],
        references=[ReferenceRelation(...)],
    )
)

# 一行创建
handler = GenericHandler.new_handler("order")

内部流程：

Registry.get("order") → EntityConfig
  ↓
GenericService.new_service("order")
  → BaseRepository(dao, "sys_order")
  → GenericService(repo, service_config)
  ↓
GenericHandler(service, handler_config)

8.8 响应数据映射与裁剪

Handler 默认直接将 DB 实体 dict 返回给 API 消费者，但这会暴露不应泄露的字段（如 is_deleted、password_hash、version_status）。

hframe 提供双轨机制，在 get() / list() 末尾自动执行响应映射，100% 向后兼容：

机制	配置项	适用场景
声明式字段排除	response_exclude	90% 场景：移除若干敏感字段
可编程响应映射器	response_mapper	10% 场景：自定义结构、只返回概要等

8.8.1 声明式字段排除 — response_exclude

最简单的用法，声明要从响应中移除的字段名列表：

from hframe import HandlerConfig

config = HandlerConfig(
    entity_name="user",
    response_exclude=["is_deleted", "password_hash", "version_status"],
)

效果：get() / list() 返回的每条记录中，这些字段会被自动移除。

注意：response_exclude 只排除顶层字段，不影响嵌套的引用展开数据。如需对嵌套数据也裁剪，请使用 response_mapper。

8.8.2 自定义响应映射器 — response_mapper

对每条记录执行自定义映射，完全控制返回结构：

from hframe import HandlerConfig

# 场景 1：只返回概要字段
config = HandlerConfig(
    entity_name="site",
    response_mapper=lambda entity: {
        "code": entity.get("site_code"),
        "name": entity.get("site_name"),
    },
)

# 场景 2：映射为自定义对象
class BriefUser:
    def __init__(self, id, name):
        self.id = id
        self.name = name

config = HandlerConfig(
    entity_name="user",
    response_mapper=lambda e: BriefUser(e["id"], e["name"]),
)

mapper 接收的是展开后的完整数据——此时 References/Cascades 已经注入到 dict 中，mapper 可以访问嵌套的关联数据：

# 利用展开后的 dept 数据
config = HandlerConfig(
    entity_name="user",
    response_mapper=lambda e: {
        "id": e["id"],
        "name": e["name"],
        "dept_name": e.get("dept", {}).get("name"),  # 只取部门名称
    },
)

8.8.3 两者共存

response_exclude 和 response_mapper 可以同时配置。执行顺序：先排除，再映射——mapper 接收到的是已排除后的数据。

config = HandlerConfig(
    entity_name="user",
    response_exclude=["is_deleted", "password_hash"],  # 先排除
    response_mapper=lambda e: {"id": e["id"], "name": e["name"]},  # 再映射
)

8.8.4 执行位置与级联隔离

映射仅在顶层 HTTP 入口（handler.get() / handler.list()）执行，在 after_get / after_list 钩子之后：

handler.get()
  → before_get → do_get（含引用展开）→ after_get
  → _apply_response_mapper()   ← 在此执行
  → 返回

级联解析（_resolve_references / _resolve_child_refs）直接操作 result dict，不走映射路径，避免中间数据被误裁剪。

8.8.5 默认行为

配置	默认值	行为
response_exclude	[]	不排除任何字段
response_mapper	None	不执行映射，原样返回

两者均为默认值时，行为与未添加此功能前完全一致——零破坏性变更。

---

9. View 层（Django 可选）

View 层依赖 Django，仅做 HTTP I/O 适配，业务逻辑全部委托给 Handler。

9.1 BaseView

Django 视图基类，提供完整的请求生命周期：

set_config → init_utils → get_info_from_request → log_in → check_permission → 业务方法 → close_utils

GET 请求： pre_get → do_get → ap_get
POST 请求： pre_post → do_post → ap_post

核心属性：

属性	类型	说明
self.trace	str	请求唯一标识
self.user_info	dict	当前用户信息
self.user_id	int	当前用户 ID
self.get_data	dict	GET 参数
self.post_data	dict	POST 参数
self.resp_data	dict	响应数据 {code, msg, data}
self.dao	MySQLWithSSH	数据库客户端
self.cache	RedisClient	Redis 客户端
self.logger	LogUtil	日志实例

统一响应：

# 成功
self.resp_data["data"] = result
return self.return_success()    # → {"code": 200, "msg": "", "data": {...}}

# 失败
return self.return_error(code=400, msg="参数错误")

9.2 ModuleView 与自动装配

ModuleView 持有 Handler 实例，支持两种配置方式：

方式一：极简模式（self.sn）

class UserListView(ListView):
    def set_config(self):
        self.sn = "user"  # 自动从 Registry 创建 Handler

方式二：手动模式

class UserListView(ListView):
    def set_config(self):
        handler = GenericHandler.new_handler("user")
        self.set_handler(handler)

自动装配流程：

ModuleView.initialize()
  → set_config()
  → if self.sn and not self.handler:
      GenericHandler.new_handler(self.sn)
        → Registry → Service → Repository → Handler
  → init_utils() → get_info_from_request() → log_in() → ...

9.3 CRUD 视图

ListView — 列表查询

from hframe import ListView

class UserListView(ListView):
    def set_config(self):
        self.sn = "user"

    def pre_get(self, request):
        # 添加自定义筛选
        if request.GET.get("keyword"):
            self.get_data["name"] = f"like:%{request.GET['keyword']}%"

# GET /api/user/list/?page=1&page_size=20
# → {"code": 200, "data": {"count": 100, "list": [...]}}

DetailView — 详情查询

from hframe import DetailView

class UserDetailView(DetailView):
    def set_config(self):
        self.sn = "user"

# GET /api/user/detail/?id=1
# → {"code": 200, "data": {"id": 1, "username": "zhangsan", ...}}

EditView — 添加/编辑

自动根据是否包含主键判断是添加还是编辑。

from hframe import EditView

class UserEditView(EditView):
    def set_config(self):
        self.sn = "user"

# POST /api/user/edit/  {"username": "new_user"}       → 创建
# POST /api/user/edit/  {"id": 1, "username": "张三"}   → 更新

DeleteView — 删除

from hframe import DeleteView

class UserDeleteView(DeleteView):
    def set_config(self):
        self.sn = "user"

# POST /api/user/delete/  {"ids": [1, 2, 3]}

9.4 H5 视图

H5 视图预置 AJAX 模式（is_ajax=True），适用于微信小程序/H5 页面：

from hframe import H5ListView, H5EditView, H5DetailView, H5DeleteView, H5Mixin

class MyH5ListView(H5ListView):
    def set_config(self):
        self.sn = "product"

---

10. 基础设施层

10.1 MySQL（PyMySQL）

MySQLWithSSH 是 DBClient 的完整实现，基于 PyMySQL + DBUtils 连接池。

from hframe.infra import MySQLWithSSH

# 直连
dao = MySQLWithSSH(
    mysql_host="127.0.0.1",
    mysql_port=3306,
    mysql_user="root",
    mysql_password="password",
    mysql_database="my_db",
    pool_size=5,
)

# SSH 隧道
dao = MySQLWithSSH(
    ssh_host="10.0.0.1",
    ssh_port=22,
    ssh_username="root",
    ssh_password="ssh_pass",
    mysql_host="127.0.0.1",
    mysql_user="db_user",
    mysql_password="db_pass",
    mysql_database="my_db",
)

# 带 Redis 缓存表结构
dao = MySQLWithSSH(..., redis_client=redis_client)

基础 CRUD：

# 查询单条
user = dao.get_info("sys_user", search_dict={"id": 1})

# 查询列表（分页）
count, users = dao.get_list("sys_user", search_dict={"status": 1}, 
                             offset=0, size=20, order_by="id DESC")

# 插入
dao.upsert("sys_user", {"username": "zhangsan", "status": 1})

# 更新
dao.update("sys_user", data={"status": 0}, search_dict={"id": 1})

# 删除
dao.delete("sys_user", search_dict={"id": 1})

# 原生 SQL
rows = dao.query("SELECT * FROM sys_user WHERE status = %s", (1,))
row = dao.query_one("SELECT * FROM sys_user WHERE id = %s", (1,))
count = dao.query_value("SELECT COUNT(*) FROM sys_user")
affected = dao.execute("UPDATE sys_user SET status = 0 WHERE id = %s", (1,))

search_dict 条件筛选：

dao.get_list("sys_user", search_dict={
    "status": 1,                        # 等于
    "department_id": [1, 2, 3],         # IN 查询
    "name": "like:张三",                # LIKE
    "created_at": ">=:2024-01-01",      # >=
    "id": "!=:100",                     # !=
    "price": "between:100|500",         # BETWEEN
    "email": "llike:[e-mail]",          # 左 LIKE
    "phone": "rlike:138%",              # 右 LIKE
    "title": "not_like:%test%",         # NOT LIKE
})

# OR 条件
dao.get_list("sys_user", search_dict={
    "status": 1,
    ":or": {"name": "张三", "email": "[e-mail]"}
})

批量操作与事务：

# 批量插入
dao.batch_insert("sys_user", ["name", "age"], [
    {"name": "张三", "age": 25},
    {"name": "李四", "age": 30},
], batch_size=500)

# 事务
with dao.transaction() as conn:
    conn.execute("UPDATE account SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE account SET balance = balance + 100 WHERE id = 2")

表结构缓存：

columns = dao.get_table_structure("sys_user")  # 自动缓存
valid_cols = dao.filter_columns("sys_user", ["name", "age", "unknown"])  # ["name", "age"]
dao.reset_table_structure()  # 重置缓存

10.2 Redis

RedisClient 是 CacheClient 的完整实现。

from hframe.infra import RedisClient

# 创建
cache = RedisClient(
    redis_host="127.0.0.1",
    redis_port=6379,
    redis_password="password",
    redis_db=0,
    max_connections=50,
)

# SSH 隧道
cache = RedisClient(
    ssh_host="10.0.0.1", ssh_username="root", ssh_password="pass",
    redis_host="127.0.0.1", redis_port=6379,
)

基础操作：

cache.set("key", "value", ex=3600)     # 设置，带过期时间
cache.set("key", {"a": 1})             # 自动序列化
result = cache.get("key")              # 自动反序列化
cache.delete("key")
cache.exists("key")
cache.expire("key", 7200)
cache.ttl("key")
cache.incr("counter")
cache.decr("counter", 5)

Hash / List / Set / Sorted Set：

# Hash
cache.hset("user:1001", "name", "张三")
name = cache.hget("user:1001", "name")
all_fields = cache.h_get_all("user:1001")
cache.hm_set("user:1001", {"name": "张三", "age": "25"})

# List
cache.l_push("queue", "task1")
cache.r_push("queue", "task2")
items = cache.lrange("queue", 0, -1)

# Set
cache.s_add("tags", "python", "redis")
members = cache.s_members("tags")

# Sorted Set
cache.z_add("leaderboard", {"user_a": 100, "user_b": 85})
top10 = cache.z_range("leaderboard", 0, 9, desc=True)

管道与批量：

with cache.pipeline(transaction=True) as pipe:
    pipe.set("key1", "val1")
    pipe.set("key2", "val2")

values = cache.mget(["key1", "key2", "key3"])
cache.mset({"key1": "val1", "key2": "val2"})
keys = cache.get_keys("user:*")

@cached 装饰器：

@cache.cached(ttl=300, prefix="data")
def get_user_list(page):
    return dao.get_list("sys_user", offset=(page-1)*20, size=20)

# 自定义 key
@cache.cached(key_func=lambda *args: f"user:{args[0]}", ttl=600)
def get_user(user_id):
    return dao.get_info("sys_user", search_dict={"id": user_id})

统计与维护：

stats = cache.get_stats()     # 命令数、命中率等
cache.ping()                  # 连接检查
info = cache.info()           # Redis INFO
cache.clear_db()              # 清空
cache.close()                 # 关闭

10.3 微信

from hframe.infra import WechatUtil

wechat = WechatUtil(app_id="wx_app_id", app_secret="app_secret", redis=cache)

# 公众号 OAuth2.0
result = wechat.log_by_code(code, client_type="public_account")
# → {"openid": "xxx", "access_token": "xxx", ...}

# 小程序 jscode2session
result = wechat.log_by_code(code, client_type="applet")
# → {"openid": "xxx", "session_key": "xxx", "unionid": "xxx"}

# 获取手机号
phone_info = wechat.get_phone_info(code)

# 发送客服消息
wechat.send_text({"touser": "OPENID", "msgtype": "text", "text": {"content": "你好"}})

---

11. 工具层

from hframe import crypto, data, format, transform, yaml_util
from hframe import UserService, CodeService

模块	函数	说明
crypto	md5(s)	MD5 哈希
	crypt(password)	加盐加密（用于密码）
	make_id()	唯一 ID 生成
data	build_tree(flat_data, sort_field)	扁平列表转树形结构
	copy_dict(source, keys, filter_keys)	选择性复制字段
	check_data_type(data, type_map)	批量类型检查
format	to_snake_case("UserName")	→ "user_name"
	to_camel_case("user_name")	→ "userName"
	json_dumps(data)	支持 datetime/Decimal 的 JSON 序列化
	trans_key(data, "camel")	递归转换所有键的命名风格
transform	is_integer("123")	→ True
	set_default_int("abc", 0)	→ 0（安全 int 转换）
	trans_str_to_arr("1,2,3")	→ ["1", "2", "3"]
yaml_util	read_yaml_file(path)	读取 YAML 文件
	write_yaml_file(data, path)	写入 YAML 文件
auth	UserService	用户登录/注册/身份服务（静态方法）
verify	CodeService	验证码服务

---

12. 异常体系

hframe 提供类型化异常体系，每个异常携带业务码和消息：

from hframe import (
    HFrameException,           # 基础异常（code=500）
    ValidationException,       # 数据校验失败（400）
    UniqueValidationException, # 唯一性校验失败（409）
    NotFoundException,        # 资源不存在（404）
    UnauthorizedException,    # 未认证（401）
    PermissionException,      # 权限不足（403）
    ServiceException,         # 业务异常（500）
    ConflictException,        # 状态冲突（409）
    ForbiddenOperationException, # 操作被禁止（403）
    RepositoryException,      # 数据访问异常（500）
    InfrastructureException,  # 基础设施异常（500）
    is_exception,             # 异常类型匹配
    get_exception_code,       # 获取异常码
)

使用示例：

# 抛出
raise ValidationException("用户名不能为空")
raise NotFoundException(f"用户不存在: {user_id}")

# 匹配（类比 Go 的 errors.Is）
try:
    handler.delete([1])
except Exception as e:
    if is_exception(e, NotFoundException):
        return "未找到"
    elif is_exception(e, PermissionException):
        return "无权限"
    raise

# 获取码
code = get_exception_code(e)  # 404 / 403 / 500

---

13. 日志

from hframe import LogUtil

# 创建
logger = LogUtil(name="my_module", log_dir="logs")

# 使用
logger.info("处理完成")
logger.error("处理失败", exc_info=True)
logger.warning("警告")
logger.debug("调试信息")

# 关闭
logger.close()

特点：

• 日志文件：logs/business_2024-01-01.log
• 同时输出到文件和控制台
• 类级别缓存，同名不重复创建
• 格式：时间 - 名称 - 级别 - 消息

---

14. 完整示例：用户管理模块

14.1 注册实体

# registry.py
from hframe import (
    Registry, ServiceConfig, HandlerConfig,
    CascadeRelation, ReferenceRelation,
)
from hframe.infra import MySQLWithSSH, RedisClient

# 初始化基础设施
dao = MySQLWithSSH(
    mysql_host="localhost", mysql_user="root",
    mysql_password="password", mysql_database="my_app",
)
cache = RedisClient(redis_host="localhost")
Registry.set_defaults(dao=dao, cache=cache)

# 注册用户实体
Registry.register("user", table="sys_user",
    service_config=ServiceConfig(
        enable_unique_validation=True,
        unique_fields=[["username"], ["email"]],
        create_time_field="create_time",
        update_time_field="update_time",
        create_user_field="create_user_id",
        update_user_field="update_user_id",
    ),
    handler_config=HandlerConfig(
        entity_name="用户",
    ),
)

# 注册订单实体（带级联和引用）
Registry.register("order", table="sys_order",
    service_config=ServiceConfig(
        create_time_field="create_time",
        update_time_field="update_time",
    ),
    handler_config=HandlerConfig(
        entity_name="订单",
        cascades=[
            CascadeRelation(
                child_table="order_item",
                parent_fk_field="items",
                child_fk_field="order_id",
                show_name="items",
                on_delete="cascade",
            ),
        ],
        references=[
            ReferenceRelation(
                parent_table="sys_user",
                fk_field="user_id",
                parent_pk_field="id",
                show_name="user",
            ),
        ],
    ),
)

14.2 Django Views

# views.py
from hframe import ListView, DetailView, EditView, DeleteView

class UserListView(ListView):
    def set_config(self):
        self.sn = "user"

class UserDetailView(DetailView):
    def set_config(self):
        self.sn = "user"

class UserEditView(EditView):
    def set_config(self):
        self.sn = "user"

class UserDeleteView(DeleteView):
    def set_config(self):
        self.sn = "user"

14.3 Django URLs

# urls.py
from django.urls import path
from .views import UserListView, UserDetailView, UserEditView, UserDeleteView

urlpatterns = [
    path('api/user/list/', UserListView.as_view(), name='user_list'),
    path('api/user/detail/', UserDetailView.as_view(), name='user_detail'),
    path('api/user/edit/', UserEditView.as_view(), name='user_edit'),
    path('api/user/delete/', UserDeleteView.as_view(), name='user_delete'),
]

14.4 非 Django 使用（FastAPI / CLI / 任意场景）

from hframe import GenericHandler, Registry

# handler 完全不依赖 Django
handler = GenericHandler.new_handler("user")

# 创建
result = handler.create({"username": "zhangsan", "email": "[e-mail]"})

# 查询列表
count, users = handler.list({"page": 1, "page_size": 20})

# 查询详情
from hframe import GetRequest
user = handler.get(GetRequest(id=1))

# 更新
handler.update(1, {"email": "[e-mail]"})

# 删除
handler.delete([1, 2])

---

15. 公共 API

hframe 包导出的所有公共 API：

from hframe import (
    # 版本
    '__version__',

    # 配置
    'Config',

    # 异常
    'HFrameException', 'ValidationException', 'UniqueValidationException',
    'NotFoundException', 'UnauthorizedException', 'PermissionException',
    'ServiceException', 'ConflictException', 'ForbiddenOperationException',
    'RepositoryException', 'InfrastructureException',
    'is_exception', 'get_exception_code',

    # 日志
    'LogUtil',

    # Repository
    'BaseRepository', 'VersionRepository', 'ListFilters', 'Filter', 'FilterOp',

    # Service
    'GenericService', 'ServiceConfig', 'ServiceHooks', 'CrudRequest', 'VersionFieldMapping',

    # Handler（核心）
    'GenericHandler', 'HandlerConfig', 'HandlerHooks',
    'CascadeRelation', 'ReferenceRelation', 'ChildRefRelation',
    'GetRequest', 'MapRequest', 'RequestFactory',
    'StopRule', 'FieldLimit',
    'inject_expand_params', 'build_field_ctx', 'effective_expand_depth', 'get_stop_cfg',

    # 注册表
    'Registry', 'EntityConfig',

    # 工具
    'UserService', 'CodeService',
    'crypto', 'data', 'format', 'transform', 'yaml_util',
)

Django 可选（需安装 hframe[django]）：

from hframe import (
    'BaseView', 'ModuleView', 'ListView', 'DetailView', 'EditView', 'DeleteView',
    'H5BaseView', 'H5ModuleView', 'H5ListView', 'H5DetailView', 'H5EditView', 'H5DeleteView', 'H5Mixin',
)

基础设施（需安装对应可选依赖）：

from hframe.infra import MySQLWithSSH, RedisClient, RedisUtil, WechatUtil