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 参数，即使当前不直接使用。原因：

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

Pydantic 表单参数

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

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

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

响应封装

• 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) 异步读取请求体。

开发要点

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

总结

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