PASTE 框架使用手册

```markdown
  ┌─────────────────────────────────────────────────┐
  │                                                 │
  │     ██████╗  █████╗ ███████╗████████╗███████╗   │
  │     ██╔══██╗██╔══██╗██╔════╝╚══██╔══╝██╔════╝   │
  │     ██████╔╝███████║███████╗   ██║   █████╗     │
  │     ██╔═══╝ ██╔══██║╚════██║   ██║   ██╔══╝     │
  │     ██║     ██║  ██║███████║   ██║   ███████╗   │
  │     ╚═╝     ╚═╝  ╚═╝╚══════╝   ╚═╝   ╚══════╝   │
  │                                                 │
  │      Python Api-first Scalable Task Engine      │
  │                                                 │
  └─────────────────────────────────────────────────┘
  ```

## 📘 PASTE 框架使用手册 v2.0.1

> **副本信息**
  > 文件：`docs/PASTE框架使用手册.html`
  > 最后更新：2025-04-08
  > 对应框架版本：2.0.1

---

### 一、框架概述

**PASTE** —— Python Api-first Scalable Task Engine

PASTE 是一个基于 **Tornado** 的生产级 Python 轻量框架，提供：

| 特性 | 说明 |
  |------|------|
  | 自动路由加载 | 定义 `route_pattern = "/user"` 即可自动注册 API，无需手动配置路由 |
  | RBAC 权限控制 | 动态规则引擎，规则以序列化类存储于数据库，支持时间/IP/自定义规则链 |
  | Swagger 自动生成 | `/docs` 自动输出 Swagger UI，无需手写 YAML |
  | 异步任务池（带背压） | `run_background_task(coro)` 安全管理并发任务，支持任务队列上限 |
  | JWT 无状态认证 | 集成 Token 签发/验证/刷新，`@auth_token` 装饰器一行开启 |
  | Redis Stream 消息队列 | `StreamActor` 支持消费者组、消息 ACK、僵尸任务自动恢复 |
  | 雪花 ID 生成器 | 内嵌线程安全实现，无需外部依赖，单机 1 万+ ID/秒 |
  | 配置系统 | 点号路径风格：`get_config("db.engine.engine")`，单文件配置 |
  | 工具库 | 字符串/字典/文件/分页/编码器/BaseX 编解码/图表/PDF/SVG/Excel |

---

### 二、目录结构说明

```
  paste-project/
  │
  ├── paste/                       # 框架核心（禁止修改！）
  │   ├── core/                    #   基础设施层
  │   │   ├── config.py            #     点号路径配置加载器
  │   │   ├── logging.py           #     日志系统（RotatingFile + 控制台）
  │   │   └── aio_pool.py          #     异步任务池 + 背压
  │   ├── db/                      #   数据库层
  │   │   ├── engine.py            #     SQLAlchemy 引擎工厂
  │   │   ├── redis.py             #     Redis 连接 + StreamActor
  │   │   ├── basemodel.py         #     异步 ORM 基类
  │   │   ├── basetable.py         #     表反射工具
  │   │   ├── baseadapter.py       #     结果集适配器
  │   │   └── gen_models.py        #     自动生成模型类
  │   ├── web/                     #   Web 层
  │   │   ├── application.py       #     Application（自动装载 Handler）
  │   │   ├── handler.py           #     RequestHandler 基类
  │   │   ├── decorators.py        #     @route / @auth_token / @auth_permission
  │   │   ├── swagger.py           #     Swagger UI 自动生成
  │   │   ├── form.py              #     WTForms 集成
  │   │   ├── param_aware_loader.py #    异步 UIModule 数据预加载
  │   │   └── websocket.py         #     WebSocket 支持
  │   ├── rbac/                    #   访问控制层
  │   │   ├── rbac_user.py         #     用户类（权限继承）
  │   │   ├── rbac_role.py         #     角色管理
  │   │   ├── rbac_rule.py         #     规则引擎
  │   │   ├── rbac_item.py         #     权限项层次结构
  │   │   ├── rbac_assignment.py   #     用户-项分配
  │   │   └── rbac_permission.py   #     权限查询
  │   ├── security/                #   安全层
  │   │   ├── token.py             #     JWT 编解码
  │   │   └── shash.py             #     PBKDF2 密码哈希
  │   ├── service/                 #   服务层
  │   │   ├── server.py            #     进程管理
  │   │   ├── daemonize.py         #     Unix 守护进程 + PID 文件
  │   │   └── task_service.py      #     定时任务调度器
  │   ├── chart/                   #   图表生成（可选依赖）
  │   │   ├── bar.py / pie.py / line.py
  │   ├── util/                    #   工具库
  │   │   ├── ustr.py / udict.py / ufile.py
  │   │   ├── pagination.py / snow_id.py / encoder.py
  │   │   └── pdf.py / svg.py / xlsx.py
  │
  ├── examples/                    # 即开即用的示例代码
  │   ├── 01_hello_world/          #   最小化 Web 应用
  │   ├── 02_background_task/      #   后台异步任务
  │   ├── 03_redis_stream/         #   Redis Stream 发布/消费
  │   ├── 04_tasks_service/        #   定时任务服务
  │   └── 05_gen_models/           #   自动生成数据库模型
  │
  ├── tests/                       # 测试套件
  │   ├── unit/                    #   单元测试（Mock 模式）
  │   └── integration/             #   集成测试（需数据库/Redis）
  │
  ├── pyproject.toml               # 项目元数据和依赖
  ├── README.md                    # 英文文档
  ├── CHANGELOG.md                 # 变更日志
  ├── CONTRIBUTING.md              # 贡献指南
  ├── LICENSE                      # MIT 许可证
  └── .github/workflows/ci.yml     # CI 流水线
  ```

---

### 三、快速开始

#### 3.1 安装

```bash
  git clone https://github.com/wayne-zwf/paste.git
  cd paste
  pip install -e .
  ```

#### 3.2 运行 HelloWorld 示例

```bash
  cd examples/01_hello_world
  python main.py
  ```

打开浏览器访问 [http://localhost:9000/hello](http://localhost:9000/hello)

#### 3.3 创建新项目

建议的目录模版：

```
  myapp/
  ├── main.py                 # 入口
  ├── config.json             # 配置
  ├── apps/                   # Handler 层
  │   ├── __init__.py
  │   ├── handler_user.py
  │   └── handler_product.py
  ├── models/                 # 数据模型
  │   └── db_models.py
  └── service/                # 业务服务
      ├── __init__.py
      └── task_service.py
  ```

---

### 四、核心模块详解

#### 4.1 配置系统 `paste.core.config`

所有配置集中在 `config.json`，通过点号路径读取：

```python
  from paste.core import config

db_url = config.get_config("db.engine.engine")
  port = config.get_config("tornado.demo.port", 9000)  # 带默认值
  ```

配置结构：

```json
  {
    "tornado": { ... },
    "db": { "engine": { ... } },
    "redis": { "connection": "...", "streams": { ... } },
    "rbac": { "user_class": "...", "table": { ... } },
    "logger": { "default": { "basic": { ... } } }
  }
  ```

#### 4.2 自动路由装载 `paste.web.application`

Handler 只需定义 `route_pattern`，框架自动扫描注册：

```python
  # handler.py —— 无需手动添加路由
  from paste.web.decorators import route
  from paste.web.handler import RequestHandler

@route("/users")
  class UserHandler(RequestHandler):
      async def get(self):
          self.response_ok(users=await get_all_users())
  ```

```python
  # main.py —— 自动装载处理器
  from paste.web.application import Application

app = Application(
      handlers_pkg="apps",        # ← 自动扫描 apps 包下的所有 Handler
      **config.get_config("tornado.demo", {})
  )
  ```

#### 4.3 RBAC 权限系统 `paste.rbac`

三层架构：**用户 → 角色 → 权限（规则）**

```python
  # 1. 创建用户
  await RbacUser.create(username="alice", password="secure123")

# 2. 分配权限
  user = await RbacUser.find_by_username("alice")
  await user.assign({"view_reports", "edit_profile"})

# 3. 自定义规则（以序列化类存储于数据库）
  from paste.rbac.rbac_rule import RbacRule

class BusinessHoursRule(RbacRule):
      async def run(self, **kwargs) -> bool:
          hour = datetime.now().hour
          return 9 <= hour < 18
  ```

在 Handler 中一行开启：

```python
  @route("/admin/reports")
  class ReportHandler(RequestHandler):
      @auth_token
      @auth_permission
      async def get(self):
          ...
  ```

#### 4.4 Swagger 自动生成

使用 `ApplicationSwagger` 替代 `Application`：

```python
  from paste.web.application import ApplicationSwagger

app = ApplicationSwagger(
      handlers_pkg="apps",
      swagger_title="My API",
      swagger_description="My API Description",
      swagger_api_version="1.0.0",
      **settings
  )
  ```

访问 `http://localhost:9000/docs` 即可看到交互式 API 文档。

#### 4.5 Redis StreamActor `paste.db.redis`

```python
  from paste.db.redis import StreamActor

# 创建执行器（从配置读取）
  actor = StreamActor.new_actor("redis.streams.user_event")

# 发布消息
  msg_id = await actor.publish({"user_id": "123", "event": "login"})

# 消费消息（阻塞式）
  async def handler(data: dict):
      print(f"处理消息: {data}")
      return True  # ACK

await actor.run_forever(func=handler, is_delete=True)
  ```

#### 4.6 任务池 + 背压 `paste.core.aio_pool`

```python
  from paste.core.logging import echo_log
  from paste.core.aio_pool import run_background_task

async def heavy_task(data):
      result = await process(data)
      echo_log(f"任务完成: {result}")

# 提交任务（队列满时自动阻塞）
  await run_background_task(heavy_task(some_data))
  ```

#### 4.7 定时任务服务 `paste.service.task_service`

```python
  from paste.service.task_service import TaskService

ts = TaskService(service_name="数据同步服务", pid_file="/var/run/sync.pid")
  ts.add_task(
      creator=ts.create_delay_task(),
      fn=sync_data,
      delay=300  # 每 5 分钟执行一次
  )
  ts.start_service()  # 控制台模式
  # 或 ts.start()    # 守护进程模式
  ```

---

### 五、范例速查表

| 示例 | 路径 | 学习重点 |
  |------|------|----------|
  | HelloWorld | `examples/01_hello_world` | 最小化应用、自动路由、`response_ok()` |
  | 后台任务 | `examples/02_background_task` | `aio_pool`、后台协程、日志输出 |
  | Redis Stream | `examples/03_redis_stream` | `StreamActor`、发布/订阅、消息 ACK |
  | 定时任务 | `examples/04_tasks_service` | `TaskService`、守护进程、PID 管理 |
  | 模型生成 | `examples/05_gen_models` | `gen_models`、表反射、自动生成 ORM |

---

### 六、测试指南

```bash
  # 安装测试依赖
  pip install -e ".[test]"

# 运行单元测试（纯 Mock，无需外部服务）
  pytest tests/unit -v

# 运行集成测试（需 MySQL + Redis）
  pytest tests/integration -v

# 全量测试 + 覆盖率报告
  pytest --cov=paste --cov-report=term-missing
  ```

集成测试需要配置环境变量或 `config.json`：

```bash
  export PASTE_DB_URL="mysql+pymysql://root:pass@localhost:3306/paste_test"
  export PASTE_REDIS_URL="redis://localhost:6379/15"
  ```

---

### 七、语义版本号规则

本项目遵循 [SemVer 2.0](https://semver.org/)：

- **主版本号** — 不兼容的 API 修改
  - **次版本号** — 向下兼容的功能新增（如新的 Handler 装饰器、新的 Stream 功能）
  - **修订号** — 向下兼容的问题修复（如 bug fix、性能优化）

---

### 八、发布检查清单

在推送至 GitHub 公开发布前，请确认以下事项：

- [ ] `README.md` 中的徽章 URL、仓库地址已替换为真实值
  - [ ] `pyproject.toml` 中 `authors`、`Homepage`、`Repository` 已填写真实信息
  - [ ] `CHANGELOG.md` 已根据实际变更更新
  - [ ] `.gitignore` 已添加 `!examples/*/config.json` 放行示例配置
  - [ ] git 仓库已清理：`bash clean_pycache.sh`
  - [ ] `.github/workflows/ci.yml` 已完善（含 checkout、test、lint 步骤）
  - [ ] 单元测试全部通过：`pytest tests/unit -v`
  - [ ] 集成测试已通过（如有外部服务）
  - [ ] 所有示例至少手动运行一次验证
  - [ ] LICENSE 文件存在且内容正确

---

### 九、附录：常见问题

**Q: 为什么我的 Handler 没有被自动装载？**
  A: 检查三点：① 类上有 `@route(...)` 装饰器；② 类继承自 `RequestHandler`；③ `main.py` 中 `Application(handlers_pkg="你的包名")` 的包名正确。

**Q: 如何关闭 Swagger？**
  A: 使用 `Application` 而非 `ApplicationSwagger`，或不传 `swagger_*` 参数。

**Q: RBAC 规则如何持久化？**
  A: 规则类以 pickle 序列化后存入数据库 `rbac_rule` 表。建议仅内部使用，对外 API 应使用 JSON + 白名单方式。

**Q: 如何扩展用户模型？**
  A: 继承 `RbacUser` 并添加自定义字段；然后在 `config.json` 中通过 `rbac.user_class` 指向你的自定义类。

**Q: 日志文件在哪里？**
  A: 默认在 `logs/` 目录下，文件名和格式由 `config.json` 中 `logger.default.basic` 配置控制。

**Q: 为什么我的配置读取失败？**
  A: 确保项目根目录存在 `config.json` 文件，且该文件未被 `.gitignore` 排除。使用 `config.get_config("路径.字段", 默认值)` 时可以设置安全默认值。

---

### 十、附录：示例配置文件参考（`config.json` 模版）

```json
  {
    "tornado": {
      "demo": {
        "autoreload": false,
        "handlers_pkg": "apps.demo",
        "port": 9000,
        "static_path": "static",
        "template_path": "templates",
        "swagger_title": "My API",
        "swagger_description": "My API Description",
        "swagger_api_version": "1.0.0",
        "swagger_contact": "admin@example.com"
      }
    },
    "db": {
      "engine": {
        "engine": "mysql+pymysql://root:password@localhost:3306/mydb",
        "async_engine": "mysql+aiomysql://root:password@localhost:3306/mydb",
        "engine_option": {
          "echo": false,
          "pool_size": 10,
          "pool_recycle": 3600
        }
      }
    },
    "redis": {
      "connection": "redis://localhost:6379/0",
      "streams": {
        "user_event": {
          "group": "user_event_group",
          "consumer": "processor_01"
        }
      }
    },
    "rbac": {
      "user_class": "myapp.models.MyRbacUser",
      "table": {
        "rule": "rbac_rule",
        "user": "rbac_user",
        "item": "rbac_item",
        "assignment": "rbac_assignment",
        "item_child": "rbac_item_child"
      }
    },
    "logger": {
      "default": {
        "basic": {
          "filename": "logs/app.log",
          "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
          "level": 20
        },
        "name": "MyApp",
        "max_bytes": 10485760,
        "backup_count": 5
      },
      "task": {
        "basic": {
          "filename": "logs/task.log",
          "format": "%(asctime)s - %(levelname)s - %(message)s",
          "level": 20
        },
        "name": "TaskService",
        "filename": "logs/task_service.log",
        "max_bytes": 10485760,
        "backup_count": 3
      }
    }
  }