当前位置：首页 > wzjs >正文

wordpress 优惠主机没有网站seo怎么做

wzjs 2025/9/10 9:49:24

wordpress 优惠主机,没有网站seo怎么做,个人网站空间大小,wordpress 跳转链接地址目录 Flask Celery 应用项目结构1. 创建app.py2. 创建tasks.py3. 创建celery_worker.py4. 创建templates目录和index.html运行应用测试文件 Flask Celery 应用对于Flask与Celery结合的例子，需要创建几个文件。首先安装必要的依赖： pip install flas…

Flask + Celery 应用

对于Flask与Celery结合的例子，需要创建几个文件。首先安装必要的依赖：

pip install flask celery redis requests

项目结构

让我们创建以下文件结构：

flask_celery/
├── app.py          # Flask应用
├── celery_worker.py # Celery worker
├── tasks.py        # Celery任务定义
└── templates/      # 模板目录└── index.html  # 主页模板

1. 创建app.py

from flask import Flask, request, render_template, jsonify
from celery_worker import celery
import tasks
import timeapp = Flask(__name__)@app.route('/')
def index():return render_template('index.html')@app.route('/process', methods=['POST'])
def process():# 获取表单数据data = request.form.get('data', '')# 启动异步任务task = tasks.process_data.delay(data)return jsonify({'task_id': task.id,'status': '任务已提交','message': f'正在处理数据: {data}'})@app.route('/status/<task_id>')
def task_status(task_id):# 获取任务状态task = tasks.process_data.AsyncResult(task_id)if task.state == 'PENDING':response = {'state': task.state,'status': '任务等待中...'}elif task.state == 'FAILURE':response = {'state': task.state,'status': '任务失败','error': str(task.info)}else:response = {'state': task.state,'status': '任务完成' if task.state == 'SUCCESS' else '任务处理中','result': task.result if task.state == 'SUCCESS' else None}return jsonify(response)if __name__ == '__main__':app.run(debug=True)

2. 创建tasks.py

from celery_worker import celery
import time@celery.task()
def process_data(data):# 模拟耗时操作time.sleep(5)# 处理数据（这里只是简单地转换为大写）result = data.upper()return {'original_data': data,'processed_data': result,'processing_time': '5秒'}

3. 创建celery_worker.py

from celery import Celery# 创建Celery实例
celery = Celery('tasks',broker='redis://localhost:6379/0',backend='redis://localhost:6379/0'
)# 配置Celery
celery.conf.update(task_serializer='json',accept_content=['json'],result_serializer='json',timezone='Asia/Shanghai',enable_utc=True,
)

4. 创建templates目录和index.html

首先创建templates目录：

mkdir templates

然后创建index.html文件：

<!DOCTYPE html>
<html>
<head><title>Flask + Celery 示例</title><style>body { font-family: Arial, sans-serif; margin: 40px; line-height: 1.6; }h1 { color: #4285f4; }.container { max-width: 800px; margin: 0 auto; }.form-group { margin-bottom: 15px; }input[type="text"] { padding: 8px; width: 300px; }button { padding: 8px 15px; background-color: #4285f4; color: white; border: none; cursor: pointer; }button:hover { background-color: #3b78e7; }#result { margin-top: 20px; padding: 15px; background-color: #f5f5f5; border-radius: 5px; display: none; }#status { margin-top: 10px; font-style: italic; }</style>
</head>
<body><div class="container"><h1>Flask + Celery 异步任务示例</h1><div class="form-group"><label for="data">输入要处理的数据:</label><br><input type="text" id="data" name="data" placeholder="输入一些文本..."><button onclick="submitTask()">提交任务</button></div><div id="status"></div><div id="result"><h3>任务结果:</h3><pre id="result-data"></pre></div></div><script>function submitTask() {const data = document.getElementById('data').value;if (!data) {alert('请输入数据!');return;}const statusDiv = document.getElementById('status');statusDiv.textContent = '提交任务中...';// 提交任务fetch('/process', {method: 'POST',headers: {'Content-Type': 'application/x-www-form-urlencoded',},body: `data=${encodeURIComponent(data)}`}).then(response => response.json()).then(data => {statusDiv.textContent = data.status;// 开始轮询任务状态pollTaskStatus(data.task_id);}).catch(error => {statusDiv.textContent = `错误: ${error}`;});}function pollTaskStatus(taskId) {const statusDiv = document.getElementById('status');const resultDiv = document.getElementById('result');const resultDataPre = document.getElementById('result-data');// 定期检查任务状态const interval = setInterval(() => {fetch(`/status/${taskId}`).then(response => response.json()).then(data => {statusDiv.textContent = `状态: ${data.status}`;if (data.state === 'SUCCESS') {clearInterval(interval);resultDiv.style.display = 'block';resultDataPre.textContent = JSON.stringify(data.result, null, 2);} else if (data.state === 'FAILURE') {clearInterval(interval);statusDiv.textContent = `错误: ${data.error}`;}}).catch(error => {clearInterval(interval);statusDiv.textContent = `轮询错误: ${error}`;});}, 1000);}</script>
</body>
</html>

运行应用

首先，确保Redis服务器正在运行（Celery需要它作为消息代理）可以用docker启动：

docker run --name redis-server -p 6379:6379 -d redis

在这里插入图片描述

启动Celery worker：

celery -A tasks worker --loglevel=info

在这里插入图片描述

在另一个终端中启动Flask应用：

python app.py

在这里插入图片描述

测试文件

创建一个新文件test_app.py：

import requests
import time
import json# 应用服务器地址
BASE_URL = 'http://localhost:5000'def test_process_endpoint():"""测试/process端点"""print("测试1: 提交任务处理")# 准备测试数据test_data = "hello world"# 发送POST请求到/process端点response = requests.post(f"{BASE_URL}/process",data={"data": test_data})# 检查响应状态码if response.status_code == 200:print("✓ 状态码正确: 200")else:print(f"✗ 状态码错误: {response.status_code}")# 解析响应JSONresult = response.json()# 检查响应内容if 'task_id' in result:print(f"✓ 成功获取任务ID: {result['task_id']}")return result['task_id']else:print("✗ 未能获取任务ID")return Nonedef test_status_endpoint(task_id):"""测试/status/<task_id>端点"""print("\n测试2: 检查任务状态")if not task_id:print("✗ 无法测试状态端点: 缺少任务ID")return# 轮询任务状态，最多等待10秒max_attempts = 10for attempt in range(1, max_attempts + 1):print(f"\n轮询 {attempt}/{max_attempts}...")# 发送GET请求到/status/<task_id>端点response = requests.get(f"{BASE_URL}/status/{task_id}")# 检查响应状态码if response.status_code == 200:print("✓ 状态码正确: 200")else:print(f"✗ 状态码错误: {response.status_code}")continue# 解析响应JSONresult = response.json()print(f"当前状态: {result.get('state', '未知')}")# 如果任务完成，显示结果并退出循环if result.get('state') == 'SUCCESS':print("\n✓ 任务成功完成!")print("结果:")print(json.dumps(result.get('result', {}), indent=2, ensure_ascii=False))return True# 如果任务失败，显示错误并退出循环if result.get('state') == 'FAILURE':print(f"\n✗ 任务失败: {result.get('error', '未知错误')}")return False# 等待1秒后再次检查time.sleep(1)print("\n✗ 超时: 任务未在预期时间内完成")return Falsedef test_index_endpoint():"""测试首页端点"""print("\n测试3: 访问首页")# 发送GET请求到/端点response = requests.get(BASE_URL)# 检查响应状态码if response.status_code == 200:print("✓ 状态码正确: 200")print("✓ 成功访问首页")else:print(f"✗ 状态码错误: {response.status_code}")def run_all_tests():"""运行所有测试"""print("开始测试Flask+Celery应用...\n")# 测试1: 提交任务task_id = test_process_endpoint()# 测试2: 检查任务状态if task_id:test_status_endpoint(task_id)# 测试3: 访问首页test_index_endpoint()print("\n测试完成!")if __name__ == "__main__":run_all_tests()

然后，在另一个终端中运行测试：