老旧测试用例生成平台异步任务与用户通知优化

在现代 Web 开发中，异步任务处理和用户通知是两个重要的功能。由于老旧测试平台【测试用例生成平台，源码分享】进行智能化升级后，未采用异步任务处理，大模型推理时间较长，导致任务阻塞，无法处理其他任务，体验较差。本文将以 Django 框架为基础，结合 Celery 和 Redis，完成一个完整的异步任务系统，并实现用户通知功能。我们还会对接阿里云百炼 DeepSeek-R1 模型，通过大模型生成测试用例，并提供数据的增删改查功能。以下是详细的实现步骤和完整代码示例。

一、需求分析与功能拆解

1.1 功能需求

本项目的主要需求如下：

  1. 用户通过 API 输入任务名称和字段信息，系统为任务生成唯一的 UUID，并提示用户任务已创建。
  2. 异步任务系统调用阿里云百炼 DeepSeek-R1 模型，传递字段信息生成 JSON 格式的测试用例。
  3. 将模型生成的测试用例存储到数据库的测试用例预览表中。
  4. 通知用户任务完成，支持通过 WebSocket 或邮件通知。
  5. 提供一个页面对生成的测试用例数据进行增删改查操作。

1.2 技术选型

  • Django：作为后端框架，负责处理 API 请求、任务状态管理和数据库操作。
  • Celery：用于实现任务的异步处理。
  • Redis：作为 Celery 的消息队列。
  • 阿里云百炼 DeepSeek-R1：调用大模型接口生成测试用例。
  • Django Channels：实现 WebSocket 通知功能。
  • 前端页面：用于展示和管理测试用例数据。

二、项目创建与环境配置

2.1 安装依赖

在开始开发之前，需要安装以下依赖工具和库：

pip install django celery redis requests channels

2.2 项目目录结构

项目的目录结构如下：

your_project_name/
├── your_project_name/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   ├── asgi.py
│   ├── celery.py
├── your_app_name/
│   ├── __init__.py
│   ├── models.py
│   ├── views.py
│   ├── tasks.py
│   ├── signals.py
│   ├── urls.py
├── manage.py

三、系统实现

3.1 配置 Celery 和 Redis

3.1.1 在 settings.py 中配置 Celery
# Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'# Django Channels 配置
ASGI_APPLICATION = 'your_project_name.asgi.application'
CHANNEL_LAYERS = {'default': {'BACKEND': 'channels_redis.core.RedisChannelLayer','CONFIG': {'hosts': [('127.0.0.1', 6379)],},},
}
3.1.2 创建 celery.py

在项目根目录下创建 celery.py 文件，用于初始化 Celery：

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery# 设置 Django 默认的 settings 模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')app = Celery('your_project_name')# 从 Django 的 settings.py 加载配置
app.config_from_object('django.conf:settings', namespace='CELERY')# 自动发现任务模块
app.autodiscover_tasks()
3.1.3 配置 __init__.py

在项目目录下的 __init__.py 文件中添加以下内容：

from __future__ import absolute_import, unicode_literals# Celery 应用
from .celery import app as celery_app__all__ = ('celery_app',)
3.1.4 配置 asgi.py

在项目根目录下的 asgi.py 文件中配置 WebSocket 支持：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStackos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')application = ProtocolTypeRouter({'http': get_asgi_application(),'websocket': AuthMiddlewareStack(URLRouter([# WebSocket 路由])),
})

3.2 数据模型设计

在 models.py 中定义两张表：任务表 和 测试用例预览表。

from django.db import models
import uuid# 测试用例预览表
class TestCasePreview(models.Model):id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)field_name = models.CharField(max_length=255)field_type = models.CharField(max_length=255)field_value = models.TextField()generated_test_case = models.JSONField()  # 存储生成的 JSON 测试用例created_at = models.DateTimeField(auto_now_add=True)# 异步任务表
class AsyncTask(models.Model):TASK_STATUS = (('PENDING', 'Pending'),('PROCESSING', 'Processing'),('COMPLETED', 'Completed'),('FAILED', 'Failed'),)id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)name = models.CharField(max_length=255, unique=True)status = models.CharField(max_length=20, choices=TASK_STATUS, default='PENDING')created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)result_message = models.TextField(null=True, blank=True)

3.3 异步任务实现

在 tasks.py 中实现异步任务逻辑：

import requests
from celery import shared_task
from .models import AsyncTask, TestCasePreview
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync@shared_task
def generate_test_cases(task_id, field_data):try:# 获取任务task = AsyncTask.objects.get(id=task_id)task.status = 'PROCESSING'task.save()# 调用阿里云百炼 DeepSeek-R1 APIclient = OpenAI(# 如果没有配置环境变量，请用百炼API Key替换：api_key="sk-xxx"# api_key='sk-xxx',api_key='sk-712a634dbaa7444d838d20b25eb938xx',  # todo 此处需更换base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")reasoning_content = ""  # 定义完整思考过程answer_content = ""  # 定义完整回复is_answering = False  # 判断是否结束思考过程并开始回复# 创建聊天完成请求completion = client.chat.completions.create(model="deepseek-r1",  # 此处以 deepseek-r1 为例，可按需更换模型名称messages=[{'role': 'user','content': prompt_param}],stream=True,# 解除以下注释会在最后一个chunk返回Token使用量# stream_options={#     "include_usage": True# })print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")for chunk in completion:# 如果chunk.choices为空，则打印usageif not chunk.choices:print("\nUsage:")print(chunk.usage)else:delta = chunk.choices[0].delta# 打印思考过程if hasattr(delta, 'reasoning_content') and delta.reasoning_content != None:print(delta.reasoning_content, end='', flush=True)reasoning_content += delta.reasoning_contentelse:# 开始回复if delta.content != "" and not is_answering:print("\n" + "=" * 20 + "完整回复" + "=" * 20 + "\n")is_answering = True# 打印回复过程print(delta.content, end='', flush=True)answer_content += delta.content# 存储测试用例到数据库for field in field_data:TestCasePreview.objects.create(field_name=field["field_name"],field_type=field["field_type"],field_value=field["field_value"],generated_test_case=extract_json_objects(answer_content))# 更新任务状态task.status = 'COMPLETED'task.result_message = "测试用例生成成功！"task.save()# 通过 WebSocket 通知用户channel_layer = get_channel_layer()async_to_sync(channel_layer.group_send)(f"task_{task_id}",{"type": "task_status", "message": "任务已完成"})except Exception as e:task.status = 'FAILED'task.result_message = str(e)task.save()

3.4 用户接口

在 views.py 中提供创建任务的接口：

from django.http import JsonResponse
from .models import AsyncTask
from .tasks import generate_test_casesdef create_task(request):if request.method == 'POST':task_name = request.POST.get('task_name')field_data = request.POST.get('field_data')  # JSON 格式的字段列表# 检查任务名称是否唯一if AsyncTask.objects.filter(name=task_name).exists():return JsonResponse({"error": "任务名称已存在"}, status=400)# 创建任务task = AsyncTask.objects.create(name=task_name)# 调用异步任务generate_test_cases.delay(task.id, field_data)return JsonResponse({"message": "任务已创建", "task_id": task.id})

3.5 WebSocket 通知

WebSocket 消费者 consumers.py：

from channels.generic.websocket import AsyncWebsocketConsumer
import jsonclass TaskConsumer(AsyncWebsocketConsumer):async def connect(self):self.task_id = self.scope['url_route']['kwargs']['task_id']self.task_group_name = f"task_{self.task_id}"await self.channel_layer.group_add(self.task_group_name,self.channel_name)await self.accept()async def disconnect(self, close_code):await self.channel_layer.group_discard(self.task_group_name,self.channel_name)async def task_status(self, event):message = event['message']await self.send(text_data=json.dumps({"message": message}))

3.6 前端页面展示

创建一个简单的页面用于展示测试用例：

<!DOCTYPE html>
<html lang="en">
<head><title>测试用例预览</title>
</head>
<body><h1>测试用例预览</h1><table id="test-cases"><thead><tr><th>字段名称</th><th>字段类型</th><th>字段值</th><th>测试用例</th></tr></thead><tbody><!-- 数据通过 Ajax 加载 --></tbody></table>
</body>
</html>

四、总结

本项目从需求分析出发，采用 Django 框架结合 Celery、Redis 和 Django Channels，完整实现了一个基于异步任务的测试用例生成系统。系统支持任务状态管理、用户通知，以及前端数据的增删改查功能，逻辑清晰，功能完善，非常适合在实际项目中推广应用。在老旧测试平台智能化中，异步是不可或缺的整改步骤，调试好异步任务后，后续优化将进入快速迭代阶段。加油吧~

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/901389.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Java使用ANTLR4对Lua脚本语法校验

Java使用ANTLR4对Lua脚本语法校验

文章目录 什么是ANTLR&#xff1f;第一个例子ANTLR4 的工作流程Lua脚本语法校验准备一个Lua Grammar文件maven配置新建实体类Lua语法遍历器语法错误监听器单元测试 参考 什么是ANTLR&#xff1f; https://www.antlr.org/ ANTLR (ANother Tool for Language Recognition) is a…
阅读更多...
观察者模式（行为模式）

观察者模式（行为模式）

观察者模式 观察者模式属于行为模式&#xff0c;个人理解&#xff1a;和发布订阅者魔模式是有区别的 细分有两种&#xff1a;推模式和拉模式两种&#xff0c;具体区别在于推模式会自带推送参数&#xff0c;拉模式是在接收通知后要自己获取更新参数 观察者模式&#xff08;Obs…
阅读更多...
内网渗透 --- 之杀软工具探测

内网渗透 --- 之杀软工具探测

目录 内网杀软探测与应对实战方案 一、总体思路 二、探测阶段——杀软工具与手法 2.1 进程与服务检测 2.2 注册表与文件系统检测 2.3 Nmap 与 NSE 脚本扫描 三、处理阶段——探测到杀软后的应对措施 3.1 分析评估 3.2 应对策略 四、判断与验证——注入 webshell 后如…
阅读更多...
（2025亲测可用）Chatbox多端一键配置Claude/GPT/DeepSeek-网页端配置

（2025亲测可用）Chatbox多端一键配置Claude/GPT/DeepSeek-网页端配置

1. 资源准备 API Key&#xff1a;此项配置填写在一步API官网创建API令牌&#xff0c;一键直达API令牌创建页面创建API令牌步骤请参考API Key的获取和使用API Host&#xff1a;此项配置填写https://yibuapi.com/v1查看支持的模型请参考这篇教程模型在线查询 2. ChatBox网页版配…
阅读更多...
【Pandas】pandas DataFrame keys

【Pandas】pandas DataFrame keys

Pandas2.2 DataFrame Indexing, iteration 方法描述DataFrame.head([n])用于返回 DataFrame 的前几行DataFrame.at快速访问和修改 DataFrame 中单个值的方法DataFrame.iat快速访问和修改 DataFrame 中单个值的方法DataFrame.loc用于基于标签&#xff08;行标签和列标签&#…
阅读更多...
Redis存储“大数据对象”的常用策略及StackOverflowError错误解决方案

Redis存储“大数据对象”的常用策略及StackOverflowError错误解决方案

Hi&#xff0c;大家好&#xff0c;我是灰小猿&#xff01; 在一些功能的开发中&#xff0c;我们一般会有一些场景需要将得到的数据先暂时的存储起来&#xff0c;以便后面的接口或业务使用&#xff0c;这种场景我们一般常用的场景就是将数据暂时存储在缓存中&#xff0c;之后再…
阅读更多...
【Python】读取xyz坐标文件输出csv文件

【Python】读取xyz坐标文件输出csv文件

Python读取xyz坐标文件输出csv文件 import sys import numpy as np import pandas as pd from tqdm import tqdm import cv2 import argparsedef read_xyz(file_path):with open(file_path, "r") as f: # 打开文件data f.readlines() # 读取文件datas []for …
阅读更多...
leetcode 139. Word Break

leetcode 139. Word Break

这道题用动态规划解决。 class Solution { public:bool wordBreak(string s, vector<string>& wordDict) {unordered_set<string> wordSet;for(string& word:wordDict){wordSet.insert(word);}int s_len s.size();//s的下标从1开始起算&#xff0c;dp[j]…
阅读更多...
驱动开发硬核特训 · Day 11（下篇）：从 virtio_blk 看虚拟总线驱动模型的真实落地

驱动开发硬核特训 · Day 11（下篇）：从 virtio_blk 看虚拟总线驱动模型的真实落地

&#x1f50d; B站相应的视屏教程&#xff1a; &#x1f4cc; 内核&#xff1a;博文视频 - 总线驱动模型实战全解析 敬请关注&#xff0c;记得标为原始粉丝。 &#x1f527; 在上篇中&#xff0c;我们已经从理论视角分析了“虚拟总线驱动模型”在 Linux 驱动体系中的独特定位。…
阅读更多...
音视频转换器 AV 接口静电保护方案

音视频转换器 AV 接口静电保护方案

方案简介 音视频转换器是将音视频&#xff08;AV&#xff09;信号转换成其他格式或信号类型的设备或软件。 它能够实现大多数视频、音频以及图像格式之间的转换&#xff0c;包括但不限于 RMVB、AVI、 MP4、MOV 等常见格式&#xff0c;同时也支持将不同采样率、位深度、声道数…
阅读更多...
AI agents系列之全从零开始构建

AI agents系列之全从零开始构建

在我们上一篇博客文章中&#xff0c;我们全面介绍了智能代理&#xff0c;讨论了它们的特性、组成部分、演变过程、面临的挑战以及未来的可能性。 这篇文章&#xff0c;咱们就来聊聊怎么用 Python 从零开始构建一个智能代理。这个智能代理能够根据用户输入做出决策&#xff0c;…
阅读更多...
【Python爬虫】详细工作流程以及组成部分

【Python爬虫】详细工作流程以及组成部分

目录 一、Python爬虫的详细工作流程 确定起始网页 发送 HTTP 请求 解析 HTML 处理数据 跟踪链接 递归抓取 存储数据 二、Python爬虫的组成部分 请求模块 解析模块 数据处理模块 存储模块 调度模块 反爬虫处理模块 一、Python爬虫的详细工作流程 在进行网络爬虫工…
阅读更多...
Kotlin 集合过滤全指南：all、any、filter 及高级用法

Kotlin 集合过滤全指南：all、any、filter 及高级用法

在 Kotlin 中&#xff0c;集合过滤是数据处理的核心操作之一。无论是简单的条件筛选&#xff0c;还是复杂的多条件组合&#xff0c;Kotlin 都提供了丰富的 API。本文将详细介绍 filter、all、any、none 等操作符的用法&#xff0c;并展示如何在实际开发中灵活运用它们。 1. 基础…
阅读更多...
爬虫：一文掌握 curl-cffi 的详细使用（支持 TLS/JA3 指纹仿真的 cURL 库）

爬虫：一文掌握 curl-cffi 的详细使用（支持 TLS/JA3 指纹仿真的 cURL 库）

更多内容请见： 爬虫和逆向教程-专栏介绍和目录 文章目录 一、curl-cffi 概述1.1 curl-cffi介绍1.2 主要特性1.3 适用场景1.4 使用 curl-cffi 的注意事项1.5 与 requests 和 pycurl 对比1.6 curl-cffi 的安装二、基本使用2.1 同步请求2.2 异步请求三、高级功能3.1 模拟浏览器指…
阅读更多...
AllData数据中台升级发布 | 支持K8S数据平台2.0版本

AllData数据中台升级发布 | 支持K8S数据平台2.0版本

&#x1f525;&#x1f525; AllData大数据产品是可定义数据中台&#xff0c;以数据平台为底座&#xff0c;以数据中台为桥梁&#xff0c;以机器学习平台为中层框架&#xff0c;以大模型应用为上游产品&#xff0c;提供全链路数字化解决方案。 ✨杭州奥零数据科技官网&#xf…
阅读更多...
dnf install openssl失败的原因和解决办法

dnf install openssl失败的原因和解决办法

网上有很多编译OpenSSL源码(3.x版本)为RPM包的文章&#xff0c;这些文章在安装RPM包时都是执行rpm -ivh openssl-xxx.rpm --nodeps --force 这个命令能在缺少依赖包的情况下能强行执行安装 其实根据Centos的文档&#xff0c;安装RPM包一般是执行yum install或dnf install。后者…
阅读更多...
从入门到进阶：React 图片轮播 Carousel 的奇妙世界！

从入门到进阶：React 图片轮播 Carousel 的奇妙世界！

全文目录&#xff1a; 开篇语&#x1f590; 前言✨ 目录&#x1f3af; 什么是图片轮播组件&#xff1f;&#x1f528; 初识 React 中的轮播实现示例代码分析 &#x1f4e6; 基于第三方库快速实现轮播示例&#xff1a;用 react-slick优势局限性 &#x1f6e0;️ 自己动手实现一个…
阅读更多...
2025第十六届蓝桥杯PythonB组部分题解

2025第十六届蓝桥杯PythonB组部分题解

一、攻击次数 题目描述 小蓝操控三个英雄攻击敌人&#xff0c;敌人初始血量2025&#xff1a; 第一个英雄每回合固定攻击5点第二个英雄奇数回合攻击15点&#xff0c;偶数回合攻击2点第三个英雄根据回合数除以3的余数攻击&#xff1a;余1攻2点&#xff0c;余2攻10点&#xff0…
阅读更多...
新手宝塔部署thinkphp一步到位

新手宝塔部署thinkphp一步到位

目录 一、下载对应配置 二、加载数据库 三、添加FTP​ 四、上传项目到宝塔​ 五、添加站点​ 六、配置伪静态 七、其他配置 开启监控 八、常见错误 一、打开宝塔页面&#xff0c;下载对应配置。 二、加载数据库 从本地导入数据库文件 三、添加FTP 四、上传项目到宝塔…
阅读更多...
2025年，HarmonyOS认证学习及考试

2025年，HarmonyOS认证学习及考试

HarmonyOS应用开发者认证考试 基础认证 通过系统化的课程学习&#xff0c;熟练掌握 DevEco Studio&#xff0c;ArkTS&#xff0c;ArkUI&#xff0c;预览器&#xff0c;模拟器&#xff0c;SDK 等 HarmonyOS 应用开发的关键概念&#xff0c;具备基础的应用开发能力。 高级认证…
阅读更多...
最新文章

Copyright @ 2022~2023