小蚂蚁云

Appearance

Become a sponsor
企业级中台解决方案
企业级中台解决方案

业务逻辑层（Service） ​

业务逻辑层是模块开发的第三步，也是最核心的部分。所有 CRUD 操作、业务规则、数据处理都封装在 Service 层的独立函数中。

文件位置 ​

application/example/services.py

命名规范 ​

函数类型命名格式示例
分页查询get_<m>_pageget_example_page
列表查询get_<m>_listget_example_list
详情查询get_<m>_detailget_example_detail
添加add_<m>add_example
更新update_<m>update_example
删除delete_<m>sdelete_examples
状态更新update_<m>_statusupdate_example_status
数据列表get_<m>_dataget_example_data

分页查询 ​

python
def get_example_page(request):
    """
    查询案例分页数据

    Args:
        request: HttpRequest对象
            - pageNo: 当前页码（可选，默认为1）
            - pageSize: 每页记录数（可选，默认为PAGE_SIZE）
            - name: 案例名称（可选，模糊查询）
            - type: 案例类型（可选，精确匹配）
            - status: 案例状态（可选，精确匹配）

    Returns:
        R: 包含分页数据的JSON响应
    """
    try:
        # 获取分页参数
        page_no = int(request.GET.get('pageNo', 1))
        page_size = int(request.GET.get('pageSize', PAGE_SIZE))

        # 构建基础查询（只查询未删除的记录）
        query = Example.objects.filter(is_delete=False)

        # 应用筛选条件
        query = _apply_filters(query, request)

        # 按排序字段升序排列
        query = query.order_by("sort")

        # 执行分页查询
        paginator = Paginator(query, page_size)
        try:
            page_list = paginator.page(page_no)
        except (PageNotAnInteger, InvalidPage, EmptyPage):
            page_list = paginator.page(1)
            page_no = 1

        # 构建并返回分页数据
        page_data = _build_page_data(page_list, paginator, page_no, page_size)
        return R.ok(data=page_data)

    except Exception as e:
        logging.error(f"查询案例分页数据异常: {str(e)}")
        return R.failed(msg="查询失败，请稍后重试")

筛选条件应用 ​

python
def _apply_filters(queryset, request):
    """应用查询筛选条件（私有辅助函数）"""
    # 案例名称模糊筛选
    name = request.GET.get('name')
    if name:
        queryset = queryset.filter(name__contains=name)

    # 案例类型筛选（精确匹配）
    type = request.GET.get('type')
    if type:
        queryset = queryset.filter(type=type)

    # 案例状态筛选（精确匹配）
    status = request.GET.get('status')
    if status:
        queryset = queryset.filter(status=status)

    return queryset

分页数据构建 ​

python
def _build_page_data(page_list, paginator, page_no, page_size):
    """构建分页返回数据（私有辅助函数）"""
    records = []
    for item in page_list:
        record = {
            'id': item.id,
            'name': item.name,
            'avatar': get_file_url(item.avatar) if item.avatar else "",
            'type': item.type,
            'status': item.status,
            'sort': item.sort,
            'createUser': item.create_user,
            'createTime': format_datetime(item.create_time),
            'updateUser': item.update_user,
            'updateTime': format_datetime(item.update_time),
        }
        records.append(record)

    return {
        'records': records,
        'total': paginator.count,
        'size': page_size,
        'current': page_no,
        'pages': paginator.num_pages
    }

分页响应格式 ​

json
{
  "code": 0,
  "data": {
    "records": [...],
    "total": 100,
    "size": 10,
    "current": 1,
    "pages": 10
  },
  "msg": "操作成功",
  "ok": true
}

详情查询 ​

python
def get_example_detail(id):
    """
    根据ID查询案例详情

    Args:
        id: 案例ID

    Returns:
        dict/None: 案例详情字典，不存在时返回None
    """
    try:
        instance = Example.objects.filter(is_delete=False, id=id).first()
        if not instance:
            return None

        return {
            'id': instance.id,
            'name': instance.name,
            'avatar': get_file_url(instance.avatar) if instance.avatar else "",
            'type': instance.type,
            'status': instance.status,
            'sort': instance.sort,
        }
    except Exception as e:
        logging.error(f"查询案例详情异常，ID: {id}, 错误: {str(e)}")
        return None

添加记录 ​

python
def add_example(request):
    """
    添加案例

    处理流程：
    1. 解析请求体中的JSON数据
    2. 表单验证
    3. 处理头像图片（保存到 example 目录）
    4. 创建数据库记录
    """
    # 解析请求参数
    data, error = parse_request_body(request)
    if error:
        return R.failed(msg=error)

    # 表单验证
    form = forms.ExampleForm(data)
    if not form.is_valid():
        return R.failed(msg=regular.get_err(form))

    try:
        cleaned_data = form.cleaned_data

        # 处理头像图片
        avatar = cleaned_data.get('avatar')
        if avatar:
            avatar = save_file(avatar, "example")

        # 创建案例记录
        Example.objects.create(
            name=cleaned_data.get('name'),
            avatar=avatar,
            type=cleaned_data.get('type'),
            status=cleaned_data.get('status'),
            sort=cleaned_data.get('sort'),
            create_user=get_username(request),
            update_user=get_username(request)
        )
        return R.ok(msg="创建成功")
    except Exception as e:
        logging.error(f"添加案例异常: {str(e)}")
        return R.failed(msg="添加失败，请稍后重试")

添加流程说明 ​

请求 → parse_request_body → 表单验证 → 处理文件 → ORM.create → 返回结果
步骤函数说明
解析请求体parse_request_body(request)解析 JSON 请求体
表单验证forms.ExampleForm(data)验证数据合法性
处理文件save_file(url, dir)临时文件迁移到正式目录
创建记录Example.objects.create(...)ORM 创建数据库记录
获取用户名get_username(request)从 JWT token 解析当前用户

更新记录 ​

python
def update_example(request):
    """
    更新案例

    处理流程：
    1. 解析请求体中的JSON数据
    2. 验证案例ID
    3. 表单验证
    4. 查询现有记录
    5. 处理头像图片（如有变化则保存新图片）
    6. 更新各字段
    7. 保存更新
    """
    # 解析请求参数
    data, error = parse_request_body(request)
    if error:
        return R.failed(msg=error)

    # 验证案例ID
    instance_id = data.get('id')
    if not instance_id or not str(instance_id).isdigit() or int(instance_id) <= 0:
        return R.failed("案例ID无效")
    instance_id = int(instance_id)

    # 表单验证
    form = forms.ExampleForm(data)
    if not form.is_valid():
        return R.failed(msg=regular.get_err(form))

    try:
        # 查询案例记录
        instance = Example.objects.filter(id=instance_id, is_delete=False).first()
        if not instance:
            return R.failed("案例不存在")

        cleaned_data = form.cleaned_data

        # 处理头像图片（仅当有新图片且与当前不同时保存）
        avatar = cleaned_data.get('avatar')
        if avatar and avatar != instance.avatar:
            avatar = save_file(avatar, "example")
        else:
            avatar = instance.avatar

        # 更新各字段
        instance.name = cleaned_data.get('name')
        instance.avatar = avatar
        instance.type = cleaned_data.get('type')
        instance.status = cleaned_data.get('status')
        instance.sort = cleaned_data.get('sort')
        instance.update_user = get_username(request)
        instance.save()

        return R.ok(msg="更新成功")
    except Exception as e:
        logging.error(f"更新案例异常，ID: {instance_id}, 错误: {str(e)}")
        return R.failed(msg="更新失败，请稍后重试")

删除记录（批量） ​

python
def delete_examples(instance_ids):
    """
    删除案例（批量删除）

    采用逻辑删除方式（is_delete=True），而非物理删除。
    支持逗号分隔的多个ID。

    Args:
        instance_ids: 案例ID字符串，多个ID用逗号分隔
                      例如: "1" 或 "1,2,3"
    """
    if not instance_ids:
        return R.failed("记录ID不存在")

    try:
        # 解析ID列表
        id_list = [int(id_str.strip()) for id_str in instance_ids.split(',') if id_str.strip()]
        if not id_list:
            return R.failed("无效的记录ID")

        # 批量更新删除标识（逻辑删除）
        updated_count = Example.objects.filter(
            id__in=id_list,
            is_delete=False
        ).update(is_delete=True)

        return R.ok(msg=f"本次共删除{updated_count}条数据")
    except ValueError:
        return R.failed("记录ID格式错误")
    except Exception as e:
        logging.error(f"删除案例异常，IDs: {instance_ids}, 错误: {str(e)}")
        return R.failed(msg="删除失败，请稍后重试")

逻辑删除说明 ​

项目采用逻辑删除（软删除），而非物理删除：

python
# 逻辑删除：设置 is_delete=True
Example.objects.filter(id__in=id_list).update(is_delete=True)

# 物理删除（不推荐）：真正删除数据库记录
Example.objects.filter(id__in=id_list).delete()

所有查询都必须加 is_delete=False 条件过滤已删除的数据。

状态更新 ​

python
def update_example_status(request):
    """
    更新案例状态

    单独更新案例的启用/禁用状态，不涉及其他字段。
    """
    # 解析请求参数
    data, error = parse_request_body(request)
    if error:
        return R.failed(msg=error)

    # 验证案例ID
    instance_id = data.get('id')
    if not instance_id or not str(instance_id).isdigit() or int(instance_id) <= 0:
        return R.failed("案例ID无效")
    instance_id = int(instance_id)

    # 验证状态值（1=启用, 2=禁用）
    status = data.get('status')
    if status is None:
        return R.failed("状态值不能为空")
    try:
        status = int(status)
        if status not in [1, 2]:
            return R.failed("状态值无效")
    except ValueError:
        return R.failed("状态值格式错误")

    try:
        instance = Example.objects.filter(id=instance_id, is_delete=False).first()
        if not instance:
            return R.failed("案例不存在")

        instance.status = status
        instance.update_user = get_username(request)
        instance.save()

        return R.ok(msg="状态更新成功")
    except Exception as e:
        logging.error(f"更新案例状态异常，ID: {instance_id}, 错误: {str(e)}")
        return R.failed(msg="状态更新失败，请稍后重试")

数据列表（下拉选择） ​

python
def get_example_data():
    """
    获取所有启用的案例列表

    通常用于前端下拉选择框等场景。
    只返回 status=1（启用）且未删除的记录。
    """
    try:
        queryset = Example.objects.filter(
            is_delete=False,
            status=1
        ).order_by("sort").values()

        return list(queryset)
    except Exception as e:
        logging.error(f"获取案例列表异常: {str(e)}")
        return []

常用工具函数 ​

函数来源用途
parse_request_body(request)utils/common.py解析 JSON 请求体
regular.get_err(form)utils/regular.py提取表单错误信息
get_username(request)utils/security.py从 JWT 获取当前用户名
get_file_url(path)utils/common.py拼接文件完整 URL
save_file(url, dir)utils/common.py保存文件到正式目录
format_datetime(dt)utils/common.py格式化日期时间
R.ok() / R.failed()utils/R.py统一响应格式

开发要点 ​

  1. 所有查询都要加 is_delete=False：过滤已删除的数据
  2. 使用 parse_request_body() 解析请求体：统一的 JSON 解析方式
  3. 表单验证失败时使用 regular.get_err() 提取错误：统一的错误信息格式
  4. 使用 get_username(request) 获取当前用户：从 JWT token 解析
  5. 文件处理使用 save_file()：临时文件迁移到正式目录
  6. 批量删除支持逗号分隔的多个 ID：如 "1,2,3"
  7. 所有异常都要捕获并记录日志：使用 logging.error()

总结 ​

Service 层封装所有业务逻辑，每个 CRUD 操作都是独立的函数。使用 parse_request_body 解析请求体，forms.ModelForm 验证数据，R.ok() / R.failed() 返回统一响应。所有查询都必须加 is_delete=False 过滤已删除数据，批量删除采用逻辑删除方式。

小蚂蚁云团队 · 提供技术支持

Copyright © 2021~2026 www.djangoadmin.cn. All rights reserved.