Service层规范

概述

Service 层是业务逻辑层，每个模块的 services.py 包含该模块的所有业务函数。采用函数式风格，每个函数对应一个业务操作。

设计原则

• 所有业务逻辑集中在 Service 层
• View 层只负责接收请求和返回响应
• Model 层只负责数据定义
• Form 层只负责数据验证

函数命名规范

函数名	说明	参数	返回值
get_{m}_page	分页查询	request	R.ok(data=page_data)
get_{m}_list	列表查询（不分页）	request	list
get_{m}_detail	详情查询	id	dict/None
add_{m}	添加记录	request	R.ok/R.failed
update_{m}	更新记录	request	R.ok/R.failed
delete_{m}s	删除记录（支持批量）	instance_ids	R.ok/R.failed
update_{m}_status	更新状态	request	R.ok/R.failed
get_{m}_data	获取数据列表（下拉选择）	-	list

完整示例

以案例模块为例，application/example/services.py 的典型结构：

import logging
from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage, EmptyPage
from application.example import forms
from application.example.models import Example
from constant.constants import PAGE_SIZE
from utils import R, regular
from utils.common import format_datetime, parse_request_body, save_file
from utils.security import get_username


# =============================================================
# 分页查询服务
# =============================================================

def get_example_page(request):
    """查询案例分页数据"""
    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="查询失败，请稍后重试")


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


def _build_page_data(page_list, paginator, page_no, page_size):
    """构建分页返回数据"""
    records = []
    for item in page_list:
        record = {
            'id': item.id,
            'name': item.name,
            '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
    }


# =============================================================
# 详情查询服务
# =============================================================

def get_example_detail(id):
    """根据ID查询案例详情"""
    try:
        instance = Example.objects.filter(is_delete=False, id=id).first()
        if not instance:
            return None

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


# =============================================================
# 添加服务
# =============================================================

def add_example(request):
    """添加案例"""
    # 解析请求参数
    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
        Example.objects.create(
            name=cleaned_data.get('name'),
            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="添加失败，请稍后重试")


# =============================================================
# 更新服务
# =============================================================

def update_example(request):
    """更新案例"""
    data, error = parse_request_body(request)
    if error:
        return R.failed(msg=error)

    instance_id = data.get('id')
    if not instance_id:
        return R.failed("案例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
        instance.name = cleaned_data.get('name')
        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="更新失败，请稍后重试")


# =============================================================
# 删除服务（支持批量）
# =============================================================

def delete_examples(instance_ids):
    """删除案例（批量删除）"""
    if not instance_ids:
        return R.failed("记录ID不存在")

    try:
        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="删除失败，请稍后重试")


# =============================================================
# 状态更新服务
# =============================================================

def update_example_status(request):
    """更新案例状态"""
    data, error = parse_request_body(request)
    if error:
        return R.failed(msg=error)

    instance_id = data.get('id')
    if not instance_id:
        return R.failed("案例ID无效")

    status = data.get('status')
    if status is None:
        return R.failed("状态值不能为空")

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

        instance.status = int(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="状态更新失败，请稍后重试")

关键约定

请求体解析

所有 POST/PUT/DELETE 请求使用 parse_request_body(request) 解析 JSON 请求体：

data, error = parse_request_body(request)
if error:
    return R.failed(msg=error)

表单验证

在 ORM 操作之前进行表单验证：

form = forms.ExampleForm(data)
if not form.is_valid():
    return R.failed(msg=regular.get_err(form))

操作人记录

创建和更新操作需记录操作人：

from utils.security import get_username

instance.create_user = get_username(request)
instance.update_user = get_username(request)

异常处理

所有 Service 函数使用 try/except 包裹，记录日志并返回友好错误信息：

try:
    # 业务逻辑
except Exception as e:
    logging.error(f"操作异常: {str(e)}")
    return R.failed(msg="操作失败，请稍后重试")

总结

Service 层采用函数式风格，每个函数对应一个业务操作。关键约定：使用 parse_request_body 解析请求体、表单验证在 ORM 操作之前、记录操作人、try/except 包裹异常处理。所有查询过滤 is_delete=False，返回标准 R 响应。