Endpoint 层

Endpoint 层是 HTTP 接口层，负责定义路由、权限装饰器和请求参数绑定。每个业务模块对应一个 endpoint 文件，所有业务逻辑委托给 Service 层处理。

文件位置

src/api/v1/endpoints/position.py

完整代码

from fastapi import APIRouter, Request

from modules.system.position.schemas import PositionForm, PositionStatusForm
from core.access_decorators import check_demo
from core.access_decorators import permission_required
from modules.system.position.service import position_service

from core import R

# ============================================================
# 创建路由实例
# ============================================================
router = APIRouter()


# ============================================================
# 查询分页数据
# ============================================================
@router.get('/page', summary='查询分页数据')
@permission_required("sys:position:page")
def page(request: Request):
    """
    获取岗位分页列表
    :param request: FastAPI请求对象
    :return: 分页数据响应
    """
    return position_service.get_page(request)


# ============================================================
# 查询岗位详情
# ============================================================
@router.get('/detail/{id}', summary='查询岗位详情')
@permission_required("sys:position:detail")
def detail(request: Request, id):
    """
    根据岗位ID查询详细信息
    :param request: FastAPI请求对象
    :param id: 岗位ID（路径参数）
    :return: 岗位详情数据
    """
    return R.ok(data=position_service.get_detail(id))


# ============================================================
# 获取岗位列表
# ============================================================
@router.get('/list', summary='获取岗位列表')
@permission_required("sys:position:list")
def get_position_list(request: Request):
    """
    获取所有岗位列表（无分页，用于下拉选择）
    :param request: FastAPI请求对象
    :return: 岗位列表数据
    """
    return R.ok(data=position_service.get_position_list(request))


# ============================================================
# 添加岗位
# ============================================================
@router.post('/add', summary='添加岗位')
@permission_required("sys:position:add")
@check_demo
def add(request: Request, data: PositionForm):
    """
    新增岗位
    :param request: FastAPI请求对象
    :param data: 岗位表单数据
    :return: 添加结果响应
    """
    return position_service.add(request, data)


# ============================================================
# 更新岗位
# ============================================================
@router.put('/update', summary='更新岗位')
@permission_required("sys:position:update")
@check_demo
def update(request: Request, data: PositionForm):
    """
    更新岗位信息
    :param request: FastAPI请求对象
    :param data: 岗位表单数据（必须包含 id 字段）
    :return: 更新结果响应
    """
    return position_service.update(request, data)


# ============================================================
# 删除岗位
# ============================================================
@router.delete('/delete/{id}', summary='删除岗位')
@permission_required("sys:position:delete")
@check_demo
def delete(request: Request, id):
    """
    根据ID删除单个岗位
    :param request: FastAPI请求对象
    :param id: 岗位ID（路径参数）
    :return: 删除结果响应
    """
    return position_service.delete(id)


# ============================================================
# 设置岗位状态
# ============================================================
@router.put('/status', summary='设置岗位状态')
@permission_required("sys:position:status")
@check_demo
def status(request: Request, data: PositionStatusForm):
    """
    修改岗位状态（启用/禁用）
    :param request: FastAPI请求对象
    :param data: 状态表单数据
    :return: 状态更新结果响应
    """
    return position_service.update_status(request, data)


# ============================================================
# 批量删除岗位
# ============================================================
@router.delete('/batchDelete', summary='批量删除岗位')
@permission_required("sys:position:batchDelete")
@check_demo
async def batch_delete(request: Request):
    """
    批量删除岗位（请求体需传入 ID 数组）

    Request Body 示例:
        [1, 2, 3]  # 岗位ID数组

    :param request: FastAPI请求对象
    :return: 批量删除结果响应
    """
    return await position_service.batch_delete(request)

代码解析

路由实例

router = APIRouter()

每个 endpoint 文件创建一个独立的 APIRouter 实例，后续在 router.py 中统一注册。

装饰器说明

装饰器	作用	位置
`@router.get/post/put/delete`	定义 HTTP 方法和路径	最外层
`@permission_required("sys:position:page")`	RBAC 权限校验	路由装饰器之后
`@check_demo`	演示模式拦截（仅写操作）	权限装饰器之后

装饰器顺序必须严格遵守：路由 → 权限 → 演示模式。顺序错误会导致功能异常。

权限字符串命名规范

sys:{module}:{action}

动作	HTTP 方法	说明
`page`	GET	分页查询
`detail`	GET	详情查询
`list`	GET	列表查询
`add`	POST	新增
`update`	PUT	更新
`delete`	DELETE	单条删除
`batchDelete`	DELETE	批量删除
`status`	PUT	状态变更

request 参数

def page(request: Request):

所有 endpoint 都接收 request: Request 参数，即使当前不直接使用。原因：

Service 层需要通过 request.query_params 读取分页参数
中间件通过 request.state 传递用户信息
保持接口一致性

Pydantic 表单参数

def add(request: Request, data: PositionForm):

data: PositionForm 声明为函数参数后，FastAPI 自动：

从请求体读取 JSON
解析为 PositionForm 实例
执行 Pydantic 校验
校验失败返回标准错误响应

响应封装

Service 返回 R 对象的（如 add、update、delete）：直接 return
Service 返回原始数据的（如 get_detail、get_position_list）：用 R.ok(data=...) 包装

批量删除的异步处理

async def batch_delete(request: Request):
    return await position_service.batch_delete(request)

batch_delete 端点必须是 async def，因为需要 await parse_batch_ids(request) 异步读取请求体。

开发要点

endpoint 层不含业务逻辑：仅做路由定义、权限校验和参数绑定
装饰器顺序：路由 → 权限 → 演示模式
所有端点都需要 request: Request 参数
写操作必须加 @check_demo：防止演示环境被修改
批量删除端点必须是 async def：异步读取请求体

总结

Endpoint 层是 HTTP 请求的入口，负责路由定义、权限校验和参数绑定，不含业务逻辑。装饰器顺序固定为路由 → 权限 → 演示模式，写操作必须加 @check_demo，批量删除端点必须是 async def。

7.3 KiB Raw Blame History