def _get_job_priority()

in src/hyperpod_cli/service/list_training_jobs.py [0:0]

• 9 lines of code
• 3 McCabe index (conditional complexity)

    def _get_job_priority(self, job) -> Optional[str]:
        worker_template = job.get("spec", {}).get("pytorchReplicaSpecs", {}).get("Worker", {}).get("template", {})
        wl_priority_cls = worker_template.get("metadata", {}).get("labels", {}).get(KUEUE_WORKLOAD_PRIORITY_CLASS_LABEL_KEY, None)
        spec_priority_cls = worker_template.get("spec", {}).get("priorityClassName", None)

        # For reference: https://kueue.sigs.k8s.io/docs/concepts/workload_priority_class/
        # The workload priority class takes precedence over the k8s priority class.
        # Because the cli focuses on the job level which means workload priority is more essential.
        # There is possibility that these two priorities coexist at the same time. In this case,
        # the k8s priority class will be used as pod priority. cli should still take workload
        # priority in this scenario.

        if wl_priority_cls is not None:
            return wl_priority_cls
        elif spec_priority_cls is not None:
            return spec_priority_cls
        
        return None