dbsmith88/flask_qed

flask_qed 镜像文档

镜像概述和主要用途

flask_qed是一个基于Flask的应用，用于执行qed逻辑和模型数据。通过REST API端点（利用Flask和Flask-RESTful实现）初始化模型，任务交由Celery异步执行并返回唯一task_id。提供任务状态和数据检索端点（仅需task_id），成功完成的任务数据会存入MongoDB，24小时后自动过期。整个应用通过docker-compose部署，包含Flask、Celery、Redis和MongoDB四个容器。

核心功能和特性

• REST API端点：基于Flask和Flask-RESTful提供模型初始化接口
• 异步任务处理：通过Celery实现任务异步执行，支持任务队列管理
• 数据存储与过期：任务结果存储在MongoDB，自动设置24小时过期时间
• 多应用合并：使用werkzeug.wsgi.DispatcherMiddleware整合多个Flask应用，通过URL前缀路由
• 基础镜像支持：构建于quanted/qed_py3镜像之上，包含GDAL、GeoS和Proj4等地理数据处理工具

使用场景和适用范围

• 需要执行qed逻辑和模型数据的应用场景
• 需异步处理任务并通过唯一ID跟踪状态和结果的系统
• 多Flask应用需整合为单一服务的场景
• 开发、测试和生产环境中的容器化部署需求

详细使用方法和配置说明

Docker镜像信息

flask_qed镜像基于https://hub.docker.com/r/quanted/qed_py3/%E6%9E%84%E5%BB%BA%EF%BC%8C%E8%AF%A5%E5%9F%BA%E7%A1%80%E9%95%9C%E5%83%8F%E6%8F%90%E4%BE%9BGDAL%E3%80%81GeoS%E5%92%8CProj4%E5%B7%A5%E5%85%B7%E3%80%82%E9%BB%98%E8%AE%A4%E4%BD%BF%E7%94%A8%E6%9C%80%E6%96%B0%E7%89%88%E6%9C%AC%EF%BC%88master%E5%88%86%E6%94%AF%E6%9E%84%E5%BB%BA%EF%BC%89%EF%BC%8C%E5%A6%82%E9%9C%80%E4%BD%BF%E7%94%A8%E7%89%B9%E5%AE%9A%E5%88%86%E6%94%AF%EF%BC%88%E5%A6%82'staging'%E6%88%96'dev'%EF%BC%89%EF%BC%8C%E5%8F%AF%E9%80%9A%E8%BF%87%E4%BB%A5%E4%B8%8B%E5%91%BD%E4%BB%A4%E6%9E%84%E5%BB%BA%EF%BC%9A

bash
docker-compose build --build-arg version=dev SERVICE

多Flask应用合并

flask_qed通过flask_cgi.py执行，从pram_flask和hms_flask导入Flask应用，使用werkzeug.wsgi.DispatcherMiddleware合并为单一应用，通过URL前缀路由不同应用。示例：

python
# flask_cgi.py
from werkzeug.wsgi import DispatcherMiddleware
import hms_flask.flask_hms as hms
import pram_flask.flask_pram as pram

app = DispatcherMiddleware(hms.app, {
    '/pram': pram.app  # pram应用通过'/pram'前缀访问
})

添加新Flask应用

1. 将新Flask应用作为子模块添加到flask_qed仓库（类似hms_flask和pram_flask）
2. 更新Celery配置：在celery_cgi.py中，将新应用的Celery任务路径添加到celery_tasks列表：

   python
   celery_tasks = [
       'hms_flask.modules.hms_controller',
       'pram_flask.modules.pram_controller',
       'new_flask_app.modules.new_controller'  # 新应用任务路径
   ]
   

3. 更新Flask配置：在flask_cgi.py中导入新应用并添加到DispatcherMiddleware，设置URL前缀：

   python
   import new_flask_app.flask_new as new_app  # 导入新应用
   
   app = DispatcherMiddleware(hms.app, {
       '/pram': pram.app,
       '/new': new_app.app  # 新应用通过'/new'前缀访问
   })
   

添加新端点（使用Celery）

以下以hms_flask应用为例，说明添加新端点的步骤：

1. 注册端点资源

在创建Flask应用的文件（如flask_hms.py）中，使用Flask-RESTful注册端点：

python
from flask_restful import Api
api = Api(app)
from modules.new_module import MyNewClass  # 导入新模块类
api.add_resource(MyNewClass, '/new/endpoint/url')  # 注册端点URL

2. 定义端点处理类

在新模块中定义继承Resource的处理类，解析请求参数并实现HTTP方法：

python
from flask import Response
from flask_restful import Resource, reqparse

# 参数解析器
parser = reqparse.RequestParser()
parser.add_argument('ARGUMENT1')  # 请求参数1
parser.add_argument('ARGUMENT2')  # 请求参数2

class MyNewClass(Resource):
    def post(self):
        """处理POST请求"""
        args = parser.parse_args()
        # 调用Celery任务，指定队列
        task_id = self.run_task.apply_async(queue="qed", args=(args['ARGUMENT1'], args['ARGUMENT2']))
        return Response(json.dumps({'job_id': task_id.id}))  # 返回task_id

3. 实现Celery任务

从父目录导入Celery，使用@celery.task装饰器定义异步任务，并连接MongoDB存储结果：

python
from celery_cgi import celery
import json
from datetime import datetime
from pymongo import MongoClient

class MyNewClass(Resource):
    # ... 其他方法 ...
    
    @celery.task(name="new_flask_task", bind=True)
    def run_task(self, arg1, arg2):
        task_id = celery.current_task.request.id  # 获取任务ID
        mongo_db = self.connect_to_mongoDB()  # 连接MongoDB
        posts = mongo_db.posts
        time_stamp = datetime.utcnow()  # UTC时间戳
        # 处理数据
        data_value = {"arg1": arg1, "arg2": arg2, "request_time": str(time_stamp)}
        # 存入MongoDB，包含任务ID和过期时间戳
        data = {'_id': task_id, 'date': time_stamp, 'data': json.dumps(data_value)}
        posts.insert_one(data)
        return data_value
    
    def connect_to_mongoDB(self):
        """连接MongoDB（根据环境判断地址）"""
        if IN_DOCKER == "False":
            # 开发环境
            mongo = MongoClient(host='mongodb://localhost:27017/0')
        else:
            # 生产环境（Docker内部）
            mongo = MongoClient(host='mongodb://mongodb:27017/0')
        mongo_db = mongo['flask_new']
        # 设置数据24小时过期（86400秒）
        mongo.flask_new.Collection.create_index([("date", pymongo.DESCENDING)], expireAfterSeconds=86400)
        return mongo_db

4. 任务状态和数据检索端点

创建端点用于查询任务状态和结果：

python
class NewTaskData(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('job_id')  # 接收task_id参数
    
    def get(self):
        args = self.parser.parse_args()
        task_id = args.job_id
        if task_id:
            task = celery.AsyncResult(task_id)  # 获取任务状态
            if task.status == "SUCCESS":
                # 任务成功，从MongoDB获取数据
                mongo_db = connect_to_mongoDB()
                posts = mongo_db.posts
                data = json.loads(posts.find_one({'_id': task_id})['data'])
                return Response(json.dumps({'id': task.id, 'status': task.status, 'data': data}))
            else:
                # 返回任务状态
                return Response(json.dumps({'id': task.id, 'status': task.status}))
        else:
            return Response(json.dumps({'error': '无效的job_id'}))

api.add_resource(NewTaskData, '/new/task/data')  # 注册检索端点

Docker部署示例

flask_qed通过docker-compose部署，包含以下服务：

yaml
# docker-compose.yml示例
version: '3'
services:
  flask:
    build:
      context: .
      args:
        version: dev  # 版本：dev/staging/latest
    ports:
      - "5000:5000"
    depends_on:
      - redis
      - mongodb
    environment:
      - IN_DOCKER=True  # 标识Docker环境
  
  celery:
    build:
      context: .
      args:
        version: dev
    command: celery -A celery_cgi worker --loglevel=info -Q qed  # 启动Celery worker
    depends_on:
      - redis
      - mongodb
    environment:
      - IN_DOCKER=True
  
  redis:
    image: redis:latest  # Redis用于Celery消息队列
    ports:
      - "6379:6379"
  
  mongodb:
    image: mongo:latest  # MongoDB用于存储任务结果
    ports:
      - "27017:27017"

启动命令：

bash
docker-compose up -d

示例代码和资源

完整示例可参考：

• https://github.com/quanted/qed_flask
• https://github.com/quanted/hms_flask/blob/dev/modules/hms_controller.py