<gh_stars>1-10
import json

from arq.jobs import Job
from fastapi import APIRouter
from starlette.requests import Request

router = APIRouter()


async def test(request: Request):
    job = await request.app.state.arq_redis.enqueue_job('say_hello', name="wt", _queue_name="arq:queue1")
    job_ = await job.info()
    return {"job_": job_}


async def get_queue_health_check(request: Request, queue_name="arq:queue3"):
    workers = await request.app.state.arq_redis.all_workers()
    queues_ = set([json.loads(v).get("queue_name") for k, v in workers.items()])
    # print(workers)
    # print(queues_)
    __ = []
    for queue_name in queues_:
        r = await request.app.state.arq_redis._get_health_check(queue_name=f"{queue_name}:health-check")
        r_ = json.loads(r)

        workers_ = [json.loads(v).get("worker_consumer_name") for k, v in workers.items() if json.loads(v).get("queue_name") == queue_name]

        r_.update({"queue_name": queue_name, "workers": workers_})
        __.append(r_)

    return {"code": 20000, "result": __,"total":len(__)}


async def queued_jobs(request: Request, queue_name="arq:queue3"):
    queued_jobs_ = await request.app.state.arq_redis.queued_jobs(queue_name=queue_name)
    queued_jobs__ = []
    for queued_job_ in queued_jobs_:
        state = await Job(job_id=queued_job_.__dict__.get("job_id"), redis=request.app.state.arq_redis,
                          _queue_name=queue_name).status()
        queued_job_.__dict__.update({"state": state})
        queued_jobs__.append(queued_job_)

    results = await request.app.state.arq_redis.all_job_results()
    results_ = []
    for result in results:
        if result.__dict__.get("queue_name") == queue_name:
            result.__dict__.update({"state": "complete"})
            results_.append(result)

    __ = results_ + queued_jobs__

    # queued_num = len([job_ for job_ in __ if job_.__dict__.get("state") == "queued"])
    # in_progress_num = len([job_ for job_ in __ if job_.__dict__.get("state") == "in_progress"])
    # deferred_num = len([job_ for job_ in __ if job_.__dict__.get("state") == "deferred"])
    # complete_num = len([job_ for job_ in __ if job_.__dict__.get("state") == "complete"])
    # print(queued_num, in_progress_num, deferred_num, complete_num)
    return {
        "code": 20000,
        "queued_jobs__": __
    }


router.add_api_route(methods=['GET'], path="/test", endpoint=test, summary="test")
router.add_api_route(methods=['GET'], path="/get_queue_health_check", endpoint=get_queue_health_check,
                     summary="get_queue_health_check")
router.add_api_route(methods=['GET'], path="/queued_jobs", endpoint=queued_jobs, summary="queued_jobs")
