LongArticleTaskServer/routes/blueprint.py

from quart import Blueprint, jsonify, request
from applications.ab_test import GetCoverService
from applications.utils import generate_task_trace_id

from applications.tasks import TaskScheduler
from applications.service import TaskManagerService


server_blueprint = Blueprint("api", __name__, url_prefix="/api")


def server_routes(pools, log_service):

    @server_blueprint.route("/get_cover", methods=["POST"])
    async def get_cover():
        params = await request.get_json()
        task = GetCoverService(pools, params)
        return jsonify(await task.deal())

    @server_blueprint.route("/run_task", methods=["POST"])
    async def run_task():
        trace_id = generate_task_trace_id()
        data = await request.get_json()
        task_scheduler = TaskScheduler(data, log_service, pools, trace_id)
        response = await task_scheduler.deal()
        return jsonify(response)

    @server_blueprint.route("/health", methods=["GET"])
    async def hello_world():
        # data = await request.get_json()
        return jsonify({"message": "hello world"})

    @server_blueprint.route("/tasks", methods=["POST"])
    async def task_list():
        data = await request.get_json()
        TMS = TaskManagerService(pool=pools, data=data)
        res = await TMS.list_tasks()
        return jsonify(res)

    return server_blueprint