示例模块（Example）

说明

src/modules/example/ 是一个完整的示例模块，演示了标准 CRUD 模块的六文件结构和开发模式。开发者可以此为模板快速创建新业务模块。

文件结构

src/modules/example/
├── __init__.py       # 模块声明
├── models.py         # 数据模型
├── schemas.py        # 参数验证
├── repository.py     # 数据访问层
├── service.py        # 业务逻辑层
└── handlers.py       # 请求处理器

各文件说明

models.py — 数据模型

from sqlalchemy import Column, String, Integer, text
from core.base_model import BaseModel

class Example(BaseModel):
    """示例模型"""
    __tablename__ = "example"
    __table_args__ = {"comment": "示例表"}

    name = Column(String(255), nullable=False, comment="名称")
    status = Column(Integer, default=0, server_default=text("0"), comment="状态")
    sort = Column(Integer, default=0, server_default=text("0"), comment="排序")

schemas.py — 参数验证

from pydantic import Field
from core.base_schema import BaseSchema, BaseStatusSchema

class ExampleSchema(BaseSchema):
    """示例表单"""
    name: str = Field(..., min_length=1, max_length=150, description="名称")
    status: int = Field(default=0, ge=0, le=1, description="状态")
    sort: int = Field(default=0, ge=0, le=99999, description="排序")

class ExampleStatusSchema(BaseStatusSchema):
    """状态更新表单"""
    pass

repository.py — 数据访问层

from core.base_repository import BaseRepository
from modules.example.models import Example

class ExampleRepository(BaseRepository):
    """示例数据仓库"""
    model = Example

service.py — 业务逻辑层

from core.base_service import BaseService
from modules.example.models import Example
from modules.example.repository import ExampleRepository
from modules.example.schemas import ExampleSchema, ExampleStatusSchema

class ExampleService(BaseService):
    """示例服务"""
    model = Example
    repo_cls = ExampleRepository
    create_schema = ExampleSchema
    update_schema = ExampleSchema
    status_schema = ExampleStatusSchema
    page_like_fields = ("name",)
    page_eq_fields = ("status",)
    page_order_by = (("sort", "asc"),)

example_service = ExampleService()

handlers.py — 请求处理器

from common.middleware.access_decorators import check_demo, permission_required
from common.middleware.operation_log import operation_log
from core import response as R
from core.base_handler import BaseHandler
from modules.example.service import example_service

class ExamplePageHandler(BaseHandler):
    @permission_required("sys:example:list")
    async def get(self):
        return await example_service.get_page(self)

class ExampleDetailHandler(BaseHandler):
    @permission_required("sys:example:detail")
    async def get(self, id):
        return R.ok(self, data=example_service.get_detail(id))

class ExampleAddHandler(BaseHandler):
    @permission_required("sys:example:add")
    @check_demo
    @operation_log("示例管理", "添加")
    async def post(self):
        return await example_service.add(self)

class ExampleUpdateHandler(BaseHandler):
    @permission_required("sys:example:update")
    @check_demo
    @operation_log("示例管理", "更新")
    async def put(self):
        return await example_service.update(self)

class ExampleDeleteHandler(BaseHandler):
    @permission_required("sys:example:delete")
    @check_demo
    @operation_log("示例管理", "删除")
    async def delete(self, id):
        return await example_service.delete(self, id)

class ExampleStatusHandler(BaseHandler):
    @permission_required("sys:example:status")
    @check_demo
    @operation_log("示例管理", "设置状态")
    async def put(self):
        return await example_service.update_status(self)

class ExampleBatchDeleteHandler(BaseHandler):
    @permission_required("sys:example:delete")
    @check_demo
    @operation_log("示例管理", "批量删除")
    async def post(self):
        return await example_service.batch_delete(self)

路由注册

# src/api/v1/example.py
from modules.example.handlers import (
    ExamplePageHandler, ExampleDetailHandler, ExampleAddHandler,
    ExampleUpdateHandler, ExampleDeleteHandler, ExampleStatusHandler,
    ExampleBatchDeleteHandler,
)

routes = [
    (r"/example/page", ExamplePageHandler),
    (r"/example/detail/(\d+)", ExampleDetailHandler),
    (r"/example/add", ExampleAddHandler),
    (r"/example/update", ExampleUpdateHandler),
    (r"/example/delete/(\d+)", ExampleDeleteHandler),
    (r"/example/status", ExampleStatusHandler),
    (r"/example/batchDelete", ExampleBatchDeleteHandler),
]

# src/api/v1/router.py — 在 _RESOURCE_MODULES 中追加
from api.v1 import example
_RESOURCE_MODULES = [..., example]

开发新模块的步骤

基于 Example 模块创建新模块的流程：

1. 复制 src/modules/example/ 为新目录
2. 修改 models.py 中的表名和字段
3. 修改 schemas.py 中的表单验证
4. 修改 repository.py 中的 model 引用
5. 修改 service.py 中的配置项
6. 修改 handlers.py 中的权限字符串和日志描述
7. 复制 src/api/v1/example.py 为新路由文件
8. 在 router.py 中注册新路由
9. 在菜单管理中配置权限节点

总结

Example 模块是标准 CRUD 模块的参考实现，包含 models → schemas → repository → service → handlers → 路由六层结构。开发新模块时以此为模板，修改差异点即可快速完成。