def view_jobs()

in front/admin_ui/app.py [0:0]
68 lines of code
10 McCabe index (conditional complexity)

                def view_jobs(token):
                    global pending_jobs_df
                    headers = {"Authorization": f"Bearer {token}"}
                    response = requests.get(
                        f"{DV_ENDPOINT}/admin/pending-jobs",
                        headers=headers,
                        timeout=60,
                    )
                    if response.status_code == 200:
                        pending_jobs = response.json()
                        df = pd.DataFrame(
                            [
                                job
                                for job_type in pending_jobs
                                for job_state in pending_jobs[job_type]
                                for job in pending_jobs[job_type][job_state]
                            ]
                        )
                        if "started_at" in df.columns:
                            df["started_at"] = pd.to_datetime(
                                df["started_at"], errors="coerce"
                            )
                        if "last_heartbeat" in df.columns:
                            df["last_heartbeat"] = pd.to_datetime(
                                df["last_heartbeat"],
                                errors="coerce",
                            )
                        if "created_at" in df.columns:
                            df["created_at"] = pd.to_datetime(
                                df["created_at"], errors="coerce"
                            )
                            most_recent = df.nlargest(5, "created_at")
                        else:
                            most_recent = pd.DataFrame()
                        pending_jobs_df = df
                        return {
                            pending_jobs_summary_table: gr.DataFrame(
                                visible=True,
                                value=pd.DataFrame(
                                    {
                                        "Jobs": list(pending_jobs),
                                        "Waiting": [
                                            len(pending_jobs[job_type]["waiting"])
                                            for job_type in pending_jobs
                                        ],
                                        "Started": [
                                            len(pending_jobs[job_type]["started"])
                                            for job_type in pending_jobs
                                        ],
                                    }
                                ),
                            ),
                            recent_pending_jobs_table: gr.DataFrame(value=most_recent),
                        }
                    else:
                        return {
                            pending_jobs_summary_table: gr.DataFrame(
                                visible=True,
                                value=pd.DataFrame(
                                    {
                                        "Error": [
                                            f"❌ Failed to view pending jobs to {DV_ENDPOINT} (error {response.status_code})"
                                        ]
                                    }
                                ),
                            ),
                            recent_pending_jobs_table: gr.DataFrame(value=None),
                        }