heims 1.1.15

基础后端管理框架

heims 使用说明

基础后端管理框架 — 基于 Django 的后端快速开发框架
版本：1.1.14 | 作者：Huey | Python >= 3.7.4

---

目录

1. 项目概述
2. 安装与依赖
3. 配置管理
4. 核心模块详解
   • 4.1 配置模块 (config.py)
   • 4.2 日志工具 (log_util.py)
   • 4.3 通用工具函数 (functions.py)
   • 4.4 异常定义 (exceptions.py)
   • 4.5 缓存层 (redis_util.py)
   • 4.6 数据库层 (mysql_util.py)
   • 4.7 服务层
   • 4.8 视图层 (common_views.py)
   • 4.9 微信工具 (wechat_util.py)
5. 初始化机制
6. 视图开发指南
   • 6.1 BaseView — 基础视图
   • 6.2 ModuleView — 模块视图
   • 6.3 ListView — 列表查询
   • 6.4 DetailView — 详情查询
   • 6.5 EditView — 添加/编辑
   • 6.6 DeleteView — 删除
   • 6.7 H5 微信视图
7. 数据库操作指南
   • 7.1 连接与配置
   • 7.2 基础查询
   • 7.3 条件筛选 (search_dict)
   • 7.4 批量操作与事务
   • 7.5 DataService — 关联数据服务
8. 缓存操作指南
9. 公开 API

---

1. 项目概述

heims 是一个基于 Django 的后端快速开发框架，内置了以下能力：

能力	说明
配置管理	通过 config.yaml 外部化配置，覆盖默认值
数据库	MySQL 客户端，支持 SSH 隧道、连接池(pool_size:5)、批量操作、事务、表结构缓存
缓存	Redis 客户端，支持 SSH 隧道、连接池、哨兵/集群、序列化/压缩、@cached 装饰器
日志	按日期分割的日志文件，支持上下文管理器
认证	Token 认证（Header / Session / GET / Cookie 多渠道获取）
权限	页面权限 + 功能权限控制
视图模板	开箱即用的 CRUD 视图模板（列表/详情/编辑/删除）
关联处理	外键(N:1)、多对多(N:N)、一对多(1:N)三种关联模式
微信	公众号 OAuth2.0 授权 + 小程序登录，自动创建用户
工具函数	加密、树形结构、字典操作、命名转换、类型检查等

核心设计模式：

• 抽象基类 + 实现：CacheClient/RedisClient、DBClient/MySQLWithSSH — 方便切换后端
• 工厂模式：RedisUtil.new_redis_client()、MySQLUtil.new_db()
• Mixin 模式：H5Mixin 组合注入微信登录能力
• 模板方法模式：pre_get → do_get → ap_get 生命周期
• 多级缓存：表结构（内存 → Redis → DB → 文件）

---

2. 安装与依赖

安装

pip install heims

核心依赖

包	用途
Django >= 2.0.7	Web 框架
DBUtils >= 3.1.2	数据库连接池
mysqlclient >= 1.4.6	MySQL 驱动
redis >= 4.3.6	Redis 客户端
sshtunnel == 0.3.2	SSH 隧道
requests >= 2.31.0	HTTP 客户端
PyYAML >= 6.0	YAML 配置解析

---

3. 配置管理

框架通过 config.yaml 文件进行配置覆盖。在项目根目录下创建 config.yaml：

# config.yaml 示例
BASE_INFO:
  secret_key: "your-secret-key"
  auth_token_name: "my_auth_token"
  admin_role_id: 1
  user_table:
    user_info: "sys_user"
    user_role: "sys_user_role"
    user_permission: "sys_user_permission"

MYSQL_INFO:
  ssh_enable: true
  ssh_host: "10.0.0.1"
  ssh_port: 22
  ssh_user: "root"
  ssh_password: "your-ssh-password"
  host: "127.0.0.1"
  port: 3306
  user: "db_user"
  password: "db_password"
  database: "my_database"
  pool_size: 5

REDIS_INFO:
  ssh_enable: true
  ssh_host: "10.0.0.1"
  ssh_port: 22
  ssh_user: "root"
  ssh_password: "your-ssh-password"
  host: "127.0.0.1"
  port: 6379
  password: "redis-password"
  db: 0
  max_connections: 50

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"

配置说明：

配置项	类型	用途
BASE_INFO.secret_key	str	加密盐值，用于 Token 加密
BASE_INFO.auth_token_name	str	Token 在请求中的键名
BASE_INFO.admin_role_id	int	管理员角色 ID
BASE_INFO.user_table	dict	用户相关表名映射
MYSQL_INFO	dict	MySQL 数据库连接配置
REDIS_INFO	dict	Redis 缓存连接配置
WECHAT_INFO	dict	微信小程序和公众号配置

SSH 隧道：MySQL 和 Redis 都支持通过 SSH 隧道连接，只需设置 ssh_enable: true 并提供 SSH 连接参数。

---

4. 核心模块详解

4.1 配置模块 (config.py)

定义了五大数据字典，可通过 config.yaml 覆盖：

from heims.common import config

secret = config.BASE_INFO['secret_key']
db_config = config.MYSQL_INFO
redis_config = config.REDIS_INFO
error_info = config.ERROR_INFO

错误码定义（ERROR_INFO）：

键	code	说明
no_login	60001	未登录
no_permission	60002	无权限
code_error	61001	验证码错误
user_disabled	62001	用户被禁用

---

4.2 日志工具 (log_util.py)

LogUtil 是基于 Python logging 的日志类，按日期分割日志文件。

from heims import LogUtil

# 基本使用
logger = LogUtil(name="my_module", base_dir="/path/to/base")
logger.info("处理完成")
logger.error("处理失败", exc_info=True)
logger.debug("调试信息")
logger.warning("警告信息")

特点：

• 日志文件位于 {base_dir}/log/{YYYYMMDD}.log
• 支持 with 上下文管理器，自动释放文件句柄
• 类级别缓存机制，同一 name 不会重复创建 handler
• 日志格式：时间 - logger名称 - 级别 - 消息内容

# 上下文管理器
with LogUtil(name="my_task") as logger:
    logger.info("任务开始")
    # ... 业务逻辑 ...
    logger.info("任务完成")
# 退出 with 时自动关闭

---

4.3 通用工具函数 (functions.py)

提供丰富的工具函数集合：

加密与 ID 生成

from heims.common import functions

# MD5
h = functions.md5("hello")

# 加盐加密（用于密码）
encrypted = functions.crypt("password123")

# 唯一 ID 生成（时间戳 + 随机数）
new_id = functions.make_id()

JSON 序列化

# 支持 datetime、date、decimal.Decimal 等特殊类型
data = {"time": datetime.datetime.now(), "price": decimal.Decimal("19.99")}
json_str = functions.json_dumps(data)

字典操作

# 多 key 优先级取值（类似 lodash 的 get）
value = functions.get_data_from_dict(data, ["field_a", "field_b", "field_c"], default="")

# 深度合并字典
functions.deep_copy_dict(source_dict, dest_dict)

# 选择性复制字段
new_dict = functions.copy_dict(source, keys=["name", "age", "email"])
new_dict = functions.copy_dict(source, filter_keys=["password", "token"])

# 删除字段
functions.drop_key_from_dict(data, filter_keys=["password"])

# 批量重命名键
functions.change_keys(data, {"old_key": "new_key"})

# 递归转换所有键的命名风格
functions.trans_key(data, "camel")   # → 驼峰
functions.trans_key(data, "snake")   # → 下划线

命名风格转换

functions.to_snake_case("UserName")    # → "user_name"
functions.to_camel_case("user_name")   # → "userName"

树形结构

# 将扁平父子关系列表转为树
flat_data = [
    {"id": 1, "parent": 0, "name": "总公司"},
    {"id": 2, "parent": 1, "name": "研发部"},
    {"id": 3, "parent": 1, "name": "市场部"},
]
tree = functions.build_tree(flat_data, sort_field="name")

类型检查

# 检查可迭代对象元素类型
functions.check_all_elements_type([1, 2, 3], int)                # True
functions.check_all_elements_type([1, "2", 3], int)              # True (宽松模式)
functions.check_all_elements_type([1, "2", 3], int, strict=True) # False (严格模式)

# 判断字符串是否为纯整数
functions.is_integer("123")  # True
functions.is_integer("12a")  # False

# 安全 int 转换
functions.set_default_int("123", 0)  # 123
functions.set_default_int("abc", 0)  # 0

YAML 文件操作

config_data = functions.read_yaml_file("/path/to/config.yaml")
functions.write_yaml_file({"key": "value"}, "/path/to/output.yaml")

---

4.4 异常定义 (exceptions.py)

RenderException — 中断视图执行并返回响应

from heims.common.exceptions import RenderException

# 直接创建响应并抛出，中断视图后续执行
raise RenderException.create_exception(code=400, msg="参数错误")

# 快捷抛出
RenderException.raise_exception(code=403, msg="无权访问")

# 重定向异常
RenderException.raise_redirect_exception("/login.html")

CommonException — 通用业务异常

from heims.common.exceptions import CommonException

if not is_valid:
    raise CommonException("数据不合法")

---

4.5 缓存层 (redis_util.py)

RedisUtil 是静态工厂，RedisClient 是完整实现。

from heims import RedisUtil

# 创建 Redis 客户端
cache = RedisUtil.new_redis_client(
    ssh_enable=True,
    ssh_host="10.0.0.1",
    ssh_port=22,
    ssh_user="root",
    ssh_password="xxx",
    host="127.0.0.1",
    port=6379,
    password="redis-pass",
    db=0,
    max_connections=50,
    serializer="json",         # 可选 json/pickle
    enable_compression=False   # 是否启用 zlib 压缩
)

基础操作：

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

# 自增自减
cache.incr("counter")                  # → 1
cache.incr("counter", 5)              # → 6
cache.decr("counter")                  # → 5

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 / Sorted Set：

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

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):
    # 这个函数的结果将自动缓存 300 秒
    return db.query("SELECT * FROM user")

# 自定义 key 生成函数
@cache.cached(key_func=lambda *args, **kw: f"user:{args[0]}:{args[1]}", ttl=600)
def get_data(category, id):
    return fetch_from_db(category, id)

统计与维护：

stats = cache.get_stats()
# {'operations': 1000, 'cache_hits': 850, 'cache_hit_rate': '85.00%'}

cache.ping()   # 检查连接
info = cache.info()  # Redis INFO
conn_info = cache.get_connection_info()
cache.clear_db()  # 清空当前数据库
cache.close()     # 关闭连接

---

4.6 数据库层 (mysql_util.py)

MySQLUtil 是静态工厂，MySQLWithSSH 是完整实现。

from heims import MySQLUtil

# 创建 MySQL 客户端
dao = MySQLUtil.new_db(
    ssh_enable=True,
    ssh_host="10.0.0.1",
    ssh_port=22,
    ssh_user="root",
    ssh_password="xxx",
    host="127.0.0.1",
    port=3306,
    user="db_user",
    password="db_pass",
    database="my_db",
    pool_size=5,
    cache=cache  # 可选的 RedisClient 实例，用于表结构缓存
)

基础 CRUD

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

# 查询多条（分页）
users = dao.get_list("sys_user", page=1, page_size=20, order="id DESC")

# 查询单个值
count = dao.query_value("SELECT COUNT(*) FROM sys_user WHERE status = %s", (1,))

# 添加
dao.upsert("sys_user", {"name": "张三", "status": 1})

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

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

条件筛选 (search_dict)

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
})

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

# 嵌套 AND/OR
dao.get_list("sys_user", search_dict={
    ":and": {
        "status": 1,
        ":or": {
            "department_id": 1,
            "role_id": [1, 2]
        }
    }
})

条件操作符速查表

操作符	格式	示例	SQL
等于	value	{"id": 1}	id = 1
IN	[list]	{"id": [1,2,3]}	id IN (1,2,3)
>	'>:value'	{"id": ">:100"}	id > 100
>=	'>=:value'	{"id": ">=:100"}	id >= 100
<	'<:value'	{"id": "<:100"}	id < 100
<=	'<=:value'	{"id": "<=:100"}	id <= 100
!=	'!=:value'	{"id": "!=:100"}	id != 100
BETWEEN	'between:v1|v2'	{"price": "between:100|500"}	price BETWEEN 100 AND 500
LIKE	'like:xxx'	{"name": "like:%张"}	name LIKE '%张'
左LIKE	'llike:xxx'	{"name": "llike:张%"}	name LIKE '张%'
右LIKE	'rlike:xxx'	{"name": "rlike:%张"}	name LIKE '%张'
NOT LIKE	'not_like:xxx'	{"name": "not_like:%张"}	name NOT LIKE '%张'

批量操作

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

# 批量插入（忽略重复）
dao.batch_insert("sys_user", ["name", "age"], rows, ignore_duplicate=True)

# 批量插入（重复时更新）
dao.batch_insert("sys_user", ["name", "age"], rows, on_duplicate_update=["age"])

事务

# 上下文管理器，自动 commit / rollback
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")

# 重置表结构缓存
dao.reset_table_structure()

# 检查列是否存在
valid = dao.check_columns("sys_user", ["name", "age"])

# 过滤有效列
valid_cols = dao.filter_columns("sys_user", ["name", "age", "unknown_field"])

原生 SQL

# 查询多条
rows = dao.query("SELECT id, name FROM sys_user WHERE status = %s", (1,))

# 查询单条
row = dao.query_one("SELECT * FROM sys_user WHERE id = %s", (1,))

# 执行（INSERT/UPDATE/DELETE）
affected = dao.execute("UPDATE sys_user SET status = 0 WHERE id = %s", (1,))

# 存储过程
dao.call_procedure("sp_update_user", (1, "张三"))

# 获取连接（用于复杂操作）
with dao.get_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT ...")

---

4.7 服务层

UserService — 用户服务

全静态方法，无需实例化。

from heims import UserService

# 用户登录（4种方式）
# 方式1：根据 user_id 直接登录
user_info = UserService.login(dao, cache, user_id=1)

# 方式2：用户名密码登录
user_info = UserService.login(dao, cache, username="admin", password="123456")

# 方式3：手机号/邮箱 + 验证码登录
user_info = UserService.login(dao, cache, mobile="13800138000", code="1234")

# 方式4：微信 openid/unionid 登录
user_info = UserService.login(dao, cache, openid="oXXXXXXX")

# 获取用户完整信息（角色、权限、菜单树）
user = UserService.get_info_by_id(1, dao, cache)

# 获取到的用户信息结构
# {
#     "id": 1,
#     "username": "admin",
#     "role_list": [...],        # 角色列表
#     "permission_list": [...],  # 权限列表
#     "menu_list": [...]         # 菜单树
# }

DataService — 关联数据服务

封装了 CRUD 操作，并自动处理外键、多对多、一对多三种关联。

from heims import DataService

# 基础 CRUD
ds = DataService(
    dao=dao,
    cache=cache,
    table="sys_user",
    column_list=["id", "name", "email", "status", "department_id", "created_at"]
)

# 查询列表（分页、筛选、排序、分组）
result = ds.get_list({
    "page": 1,
    "page_size": 20,
    "search_dict": {"status": 1},
    "order": "id DESC",
    "group_by": "department_id"
})

# 查询详情
detail = ds.get_detail({"id": 1})

# 添加
ds.edit({"name": "新用户", "email": "[e-mail]", "status": 1})

# 更新
ds.edit({"id": 1, "name": "新名字"})

# 软删除（is_del=1）或硬删除
ds.delete({"id": 1})

三种关联模式配置：

ds = DataService(
    dao=dao,
    cache=cache,
    table="sys_user",
    column_list=["id", "name", "department_id"],

    # 1. 外键关联 (N:1) — 查询时自动 JOIN 获取关联表数据
    fk_info=[
        {
            "table": "sys_department",
            "fk_ids": "department_id",       # 本表外键字段
            "columns": ["id", "dept_name"],  # 关联表要查询的列
            "show_name": "department"        # 返回数据中的键名
        }
    ],

    # 2. 多对多关联 (N:N) — 通过中间表关联
    # 场景举例：订单表(order)通过中间表(order_item)关联商品表(product)
    m2m_info=[
        {
            "show_name": "products",              # 返回数据中的键名，默认取中间表名加"_list"
            "search_info": [
                # search_info[0] — 中间表配置（如 order_item）
                {
                    "table": "order_item",        # 中间表表名
                    "fk_ids": {"id": "order_id"}, # 主表主键 → 中间表外键 （此处 order_item.order_id 关联 order.id）
                    "columns": ["id", "order_id", "product_id"],  # 中间表要查询的字段
                    "expire": 0                    # 缓存过期时间（秒），0 表示不缓存
                },
                # search_info[1] — 目标表配置（如 product）
                {
                    "table": "product",           # 目标表表名
                    "fk_ids": {"product_id": "id"},# 中间表外键 → 目标表主键（此处 order_item.product_id 关联 product.id）
                    "columns": ["id", "name", "price", "img"],  # 目标表要查询的字段
                    "expire": 0                    # 缓存过期时间
                }
            ]
        },
        # 第二个多对多关联示例：如文章(article)通过中间表关联标签(tag)
        {
            "show_name": "tags",
            "search_info": [
                {"table": "article_tag", "fk_ids": {"article_id": "id"}, "columns": ["id"]},
                {"table": "tag", "fk_ids": {"tag_id": "id"}, "columns": ["id", "name", "color"]}
            ]
        }
    ],

    # 3. 一对多关联 (1:N) — 子表数据
    content_info=[
        {
            "table": "user_log",
            "fk_ids": "user_id",
            "columns": ["id", "action", "created_at"],
            "show_name": "logs",
            "model": "1vn",             # 1vn 模式（一对多）
            "search_dict": {},          # 子表筛选条件
            "order": "id DESC",
            "page": 1,
            "page_size": 100
        }
    ],
)

---

4.8 视图层 (common_views.py)

视图继承层次：

View (Django)
  └── BaseView               # 基础视图
        └── ModuleView        # 模块视图（带关联配置）
              ├── ListView    # 列表查询
              ├── DetailView  # 详情查询
              ├── EditView    # 添加/编辑
              └── DeleteView  # 删除

BaseView — 完整请求生命周期

每次请求都会经历以下流程：

set_config() → init_utils() → get_info_from_request() → log_in() → save_log() → check_permission() → 业务方法(do_get / do_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.logger	LogUtil	日志实例
self.cache	RedisClient	Redis 客户端
self.dao	MySQLWithSSH	数据库客户端
self.template_name	str	模板路径（页面渲染时使用）

统一响应方法：

# 成功
self.set_data({"total": 100, "list": [...]})  # 设置响应数据
return self.return_success()                   # 返回 {"code": 200, "msg": "", "data": {...}}

# 失败
return self.return_error(code=400, msg="参数错误")
return self.return_error(code=403, msg="无权访问")

# 页面渲染
self.template_name = "index.html"
return self.return_render({"title": "首页"})

# 设置 Cookie
self.set_cookie_data.append({
    "key": "auth_token",
    "value": "xxx",
    "options": {"max_age": 86400, "httponly": True}
})

日志记录：

def do_get(self, request):
    self.logger.info(f"查询用户列表, page={self.get_data.get('page')}")
    # ... 业务逻辑 ...
    self.save_response_log(self.resp_data)  # 保存响应日志到文件
    return self.return_success()

---

4.9 微信工具 (wechat_util.py)

from heims import WechatUtil

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

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

# 生成小程序码
buffer = wechat.get_qr_code(page="pages/index/index", scene="id=123", _type="unlimited")

---

5. 初始化机制

导入 heims 包时自动执行 initialize()：

1. 读取当前工作目录下的 config.yaml
2. 将 YAML 配置深度合并到 config.py 中各配置字典
3. 规则：YAML 的 section 名按规则映射到 config 中的字典名

例如，YAML 中 MYSQL_INFO 会覆盖 config.MYSQL_INFO 的值。

使用方式：

import heims  # 自动执行 initialize()，读取 config.yaml

或在配置初始化后：

from heims import initialize
initialize()  # 手动调用初始化

---

6. 视图开发指南

6.1 BaseView — 基础视图

用于需要自定义 GET/POST 业务逻辑的场景。

from heims import BaseView

class MyCustomView(BaseView):
    """自定义视图"""

    def set_config(self):
        """配置页面参数"""
        self.page_permission = {"my_permission": "/404.html"}  # 权限检查
        self.template_name = ""  # 不为空时表示页面渲染，为空表示 AJAX/API

    def pre_get(self, request):
        """GET 请求前置处理"""
        self.logger.info("pre_get: 准备查询")

    def do_get(self, request):
        """GET 请求核心逻辑"""
        keyword = self.get_data.get("keyword", "")
        result = some_query_logic(keyword)
        self.set_data(result)

    def ap_get(self, request):
        """GET 请求后置处理"""
        self.save_response_log(self.resp_data)

    def pre_post(self, request):
        """POST 请求前置处理"""
        # 参数校验
        if not self.post_data.get("name"):
            self.return_error(code=400, msg="名称不能为空", raise_error=True)

    def do_post(self, request):
        """POST 请求核心逻辑"""
        self.dao.upsert("my_table", self.post_data)
        self.set_data({"id": self.post_data.get("id")})

    def ap_post(self, request):
        """POST 请求后置处理"""
        self.save_response_log(self.resp_data)

请求生命周期（模板方法模式）：

GET 请求	POST 请求	说明
pre_get()	pre_post()	前置处理（参数校验）
do_get()	do_post()	核心业务逻辑
ap_get()	ap_post()	后置处理（日志等）

6.2 ModuleView — 模块视图

继承 BaseView，增加 DataService 自动配置。适合需要关联处理的场景。

from heims import ModuleView

class MyModuleView(ModuleView):
    """带关联配置的模块视图"""

    def set_config(self):
        super().set_config()

        # 关联配置
        self.fk_info = [
            {
                "table": "sys_department",
                "fk_ids": "department_id",
                "columns": ["id", "dept_name"],
                "show_name": "department"
            }
        ]
        self.m2m_info = []
        self.content_info = []

6.3 ListView — 列表查询

内置分页、排序、筛选、分组功能。

from heims import ListView

class MyListView(ListView):
    """列表查询视图"""

    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.column_list = ["id", "name", "email", "status", "created_at"]
        self.page_permission = {"view_user_list": "/404.html"}
        self.fk_info = [...]   # 外键关联配置

    def pre_get(self, request):
        """自定义筛选条件"""
        # 默认只查自己的数据
        self.search_dict["department_id"] = self.user_info["department_id"]
        # 前端传来的筛选
        if self.get_data.get("keyword"):
            self.search_dict["name"] = f"like:%{self.get_data['keyword']}%"

# URL 配置示例
# path('api/user/list/', MyListView.as_view(), name='user_list')

6.4 DetailView — 详情查询

from heims import DetailView

class MyDetailView(DetailView):
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.column_list = ["id", "name", "email", "status", "created_at"]

# GET /api/user/detail/?id=1
# 响应: {"code": 200, "msg": "", "data": {"id": 1, "name": "张三", ...}}

6.5 EditView — 添加/编辑

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

from heims import EditView

class MyEditView(EditView):
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.column_list = ["name", "email", "status", "department_id"]
        self.page_permission = {"edit_user": "/404.html"}

# POST /api/user/edit/  {"name": "新用户", "email": "[e-mail]"}           → 添加
# POST /api/user/edit/  {"id": 1, "name": "新名字"}                          → 编辑

6.6 DeleteView — 删除

支持软删除（默认，is_del=1）和硬删除。

from heims import DeleteView

class MyDeleteView(DeleteView):
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.hard_delete = False  # 默认软删除，设为 True 即硬删除
        self.page_permission = {"delete_user": "/404.html"}

# POST /api/user/delete/  {"id": 1}

6.7 H5 微信视图

提供带微信 OAuth2.0 授权的视图模板，适用于公众号 H5 页面。

from heims import H5ListView, H5EditView, H5DetailView, H5DeleteView, H5BaseView

class MyH5ListView(H5ListView):
    """H5 微信列表视图"""
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.column_list = ["id", "name", "email", "status", "created_at"]

H5 视图特点：

• 自动进行微信 OAuth2.0 授权
• 获取用户 OpenID，自动创建/关联系统用户
• Token 认证，保持登录态
• 其余行为与普通视图完全一致

---

7. 数据库操作指南

7.1 连接与配置

from heims import MySQLUtil, config

# 方式1：使用 config.yaml 配置（推荐）
dao = MySQLUtil.new_db(**config.MYSQL_INFO, cache=cache)

# 方式2：直接传参
dao = MySQLUtil.new_db(
    ssh_enable=False,
    host="127.0.0.1",
    port=3306,
    user="root",
    password="password",
    database="my_db",
    pool_size=5
)

连接池说明：使用 DBUtils 的 PooledDB，连接自动管理。pool_size 控制最大连接数。

SSH 隧道：当 ssh_enable=True 时，框架会自动建立 SSH 隧道，将数据库连接转发给远程服务器。

7.2 基础查询

# 查询全部
rows = dao.get_list("sys_user", page=1, page_size=100)
# 返回: {"total": 1000, "list": [...], "pages": ...}

# 条件查询
rows = dao.get_list("sys_user",
    search_dict={"status": 1, "department_id": [1, 2]},
    order="id DESC",
    page=1, page_size=20
)

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

# 查询计数
total = dao.query_value("SELECT count(*) FROM sys_user WHERE status=1")

7.3 条件筛选 (search_dict)

丰富灵活的条件构建器：

# IN 查询
{"id": [1, 2, 3, 4]}                         # → id IN (1,2,3,4)

# 模糊查询
{"name": "like:%张"}                           # → name LIKE '%张'
{"name": "llike:张%"}                          # → name LIKE '张%'
{"name": "not_like:%test%"}                   # → name NOT LIKE '%test%'

# 范围查询
{"age": ">:18"}                               # → age > 18
{"age": "<=:60"}                              # → age <= 60
{"created_at": ">=:2024-01-01"}               # → created_at >= '2024-01-01'

# BETWEEN
{"price": "between:100|500"}                  # → price BETWEEN 100 AND 500

# 不等于
{"status": "!=:0"}                            # → status != 0

# OR 条件
{":or": {"status": 0, "is_del": 1}}           # → (status = 0 OR is_del = 1)

# 嵌套组合
{":and": {
    "department_id": 1,
    ":or": {"status": 1, "status": 2}
}}

7.4 批量操作与事务

# 批量插入（默认）
rows = [{"name": f"user_{i}", "age": i} for i in range(1000)]
dao.batch_insert("sys_user", ["name", "age"], rows, batch_size=500)

# 忽略重复
dao.batch_insert("sys_user", ["name", "age"], rows, ignore_duplicate=True)

# 重复则更新
dao.batch_insert("sys_user", ["name", "age"], rows, on_duplicate_update=["age"])

# 事务
try:
    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")
except Exception as e:
    # 自动回滚
    logger.error(f"事务失败: {e}")

7.5 DataService — 关联数据服务

DataService 在 CRUD 基础上自动处理关联数据的查询和写入：

from heims import DataService

ds = DataService(
    dao=dao,
    cache=cache,
    table="sys_article",
    column_list=["id", "title", "content", "category_id", "author_id"],
    fk_info=[{
        "table": "sys_category",
        "fk_ids": "category_id",
        "columns": ["id", "name"],
        "show_name": "category"
    }],
    m2m_info=[{
        "search_info": [
            {"search_dict": {}, "expire": 0},
            {"search_dict": {}, "expire": 0}
        ],
        "show_name": "tags"
    }]
)

# 查询列表 → 自动关联 category 和 tags
result = ds.get_list({"page": 1, "page_size": 20})
# 响应中自动包含 category_name, tags 等关联数据

# 添加/编辑 → 自动处理关联表写入
ds.edit({
    "id": 1,
    "title": "新标题",
    "category_id": 2,
    "tag_list": [1, 3, 5]  # 多对多关联自动写入中间表
})

---

8. 缓存操作指南

from heims import RedisUtil, config

cache = RedisUtil.new_redis_client(**config.REDIS_INFO)

# 字符串
cache.set("key", "value", ex=3600)
value = cache.get("key")

# Hash（适合存储对象）
cache.hm_set("user:1001", {"name": "张三", "email": "[e-mail]"})
name = cache.hget("user:1001", "name")
all_info = cache.h_get_all("user:1001")

# 列表（消息队列场景）
cache.l_push("task_queue", json.dumps({"task_id": 1, "action": "send_email"}))
task = cache.r_push("task_queue", json.dumps({"task_id": 2}))

# 装饰器缓存
@cache.cached(ttl=300)
def get_config():
    return db.query("SELECT * FROM sys_config")
# 首次调用查询数据库，300秒内后续调用直接返回缓存

# 管道（减少网络往返）
with cache.pipeline() as pipe:
    pipe.set("a", 1)
    pipe.set("b", 2)
    pipe.set("c", 3)

---

9. 公开 API

heims 包导出的所有公共 API 可直接使用：

from heims import (
    # 初始化
    'initialize',
    # 配置模块
    'config',
    # 日志
    'LogUtil',
    # Redis
    'RedisUtil', 'RedisClient',
    # MySQL
    'MySQLUtil', 'MySQLWithSSH',
    # 微信
    'WechatUtil',
    # 视图
    'BaseView', 'ModuleView', 'ListView', 'DetailView', 'EditView', 'DeleteView',
    # H5 视图
    'H5DeleteView', 'H5EditView', 'H5DetailView', 'H5ListView', 'H5BaseView', 'H5ModuleView', 'H5Mixin',
    # 服务
    'UserService', 'DataService', 'CodeService',
)

---

快速开始示例

一个完整的最小化视图示例：

import heims
from heims import ListView, EditView, DeleteView

class UserListView(ListView):
    """用户列表"""
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.column_list = ["id", "name", "email", "status", "created_at"]
        self.page_permission = {"view_user": "/404.html"}

class UserEditView(EditView):
    """用户编辑"""
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.column_list = ["name", "email", "status"]
        self.page_permission = {"edit_user": "/404.html"}

class UserDeleteView(DeleteView):
    """用户删除"""
    def set_config(self):
        super().set_config()
        self.table = "sys_user"
        self.page_permission = {"delete_user": "/404.html"}

然后创建 config.yaml 配置数据库/Redis/微信信息，放在项目根目录。导入 heims 包时自动加载。