Redis RQ Python-作业未注册

如何解决Redis RQ Python-作业未注册

在确定如何使Flask与RQ一起使用时，我需要您的帮助。我有一个简单的Flask页面，上面有一堆链接来触发正在/应该输入RQ的脚本。似乎其中只有一个“ run_db_script”（请参阅​​-tasks_combined.py）正在注册并显示在正在运行的作业中。其他人也将运行，但不会注册。另外，我对如何获取工作状态有些困惑，应该使用“ rq.job”还是“ rq.registry”？

感谢您的关注

worker.py：

import os
import sys
import redis
from rq import Worker,Queue,Connection

sys.path.insert(0,"/some_path/")

listen = ["default"]

redis_url = os.getenv("REdisTOGO_URL","redis://localhost:6379")

conn = redis.from_url(redis_url)

if __name__ == "__main__":
    with Connection(conn):
        worker = Worker(list(map(Queue,listen)))
        worker.work()

tasks.py：

    from rq import queue
    from rq.job import Job
    from rq.registry import StartedJobRegistry
    from app.worker import conn
    
    
    @tasks.route("/tasks",methods=["GET","POST"])
    @login_required
    def run_task():
        q = Queue("default",connection=conn)
        workers = Worker.count(connection=conn)
        registry = StartedJobRegistry(queue=q)
        num_of_jobs_in_reg = registry.count
        list_of_jobs_in_reg = registry.get_job_ids()
        jobs = q.jobs
        q_len = len(q)
        title = "Manual Script Runner"
        try:
            if request.method == "POST":
                from app.scripts.tasks.combined_tasks import run_desired_task
                task_name = request.form.get("task")
                job = q.enqueue(run_desired_task,task_name,result_ttl=5000,job_timeout=300)
                return render_template(
                                        "script_pages/manual_scripts.html",title=title,job_id=job.get_id(),job_status=job.get_status(),jobs=jobs,jobs_number=q_len,workers=workers,jobs_in_registry=num_of_jobs_in_reg,job_ids_in_registry=list_of_jobs_in_reg
                                        )
        except Exception as error:
            return render_template("pages/error_page.html",title="Error:" + str(error))

@tasks.route("/status/<job_key>","POST"])
def get_results(job_key):
    job = Job.fetch(job_key,connection=conn)
    if job.is_started:
        return jsonify(f"JOB ID: {job_key} is running,was started at: {job.started_at}"),200
    if job.is_finished:
        return jsonify(f"JOB ID: {job_key} {job.result}"),200
    else:
        return "None",202

tasks_combined.py：

def run_desired_task(task):
    print(f"[+] Received job for {task}")
    if "run_db_script" in task:
        run_db_script()
    elif "initiate_cfg_backup" in task:
        initiate_cfg_backup()
    elif "initiate_reset" in task:
        initiate_reset()