Format job data for display in the UI
(self, job: dict[str, Any])
| 74 | return all_jobs |
| 75 | |
| 76 | def format_job_for_display(self, job: dict[str, Any]) -> dict[str, Any]: |
| 77 | """Format job data for display in the UI""" |
| 78 | job_id = job.get("jobId", "") |
| 79 | job_name = job.get("jobName", "") |
| 80 | status = job.get("status", "UNKNOWN") |
| 81 | created_at = job.get("createdAt") |
| 82 | started_at = job.get("startedAt") |
| 83 | stopped_at = job.get("stoppedAt") |
| 84 | |
| 85 | if isinstance(created_at, datetime): |
| 86 | created_str = created_at.strftime("%m/%d %H:%M") |
| 87 | created_timestamp = created_at.timestamp() |
| 88 | elif created_at: |
| 89 | created_timestamp = created_at / 1000 |
| 90 | created_str = datetime.fromtimestamp(created_timestamp).strftime("%m/%d %H:%M") |
| 91 | else: |
| 92 | created_str = "" |
| 93 | created_timestamp = 0 |
| 94 | |
| 95 | if isinstance(started_at, datetime): |
| 96 | started_str = started_at.strftime("%m/%d %H:%M") |
| 97 | started_timestamp = started_at.timestamp() |
| 98 | elif started_at: |
| 99 | started_timestamp = started_at / 1000 |
| 100 | started_str = datetime.fromtimestamp(started_timestamp).strftime("%m/%d %H:%M") |
| 101 | else: |
| 102 | started_str = "" |
| 103 | started_timestamp = 0 |
| 104 | |
| 105 | time_running_str, time_running_seconds = self._calculate_time_running(started_at, stopped_at, status) |
| 106 | |
| 107 | aws_link = self._generate_aws_console_link(job_id) |
| 108 | emagedoc_link = self._generate_emagedoc_link(job_id) |
| 109 | s3_link = self._generate_s3_link(job_id) |
| 110 | round_info = self._get_round_info(job_id) |
| 111 | aws_command = self._get_aws_command(job_id) |
| 112 | |
| 113 | return { |
| 114 | "job_id": job_id, |
| 115 | "job_name": job_name, |
| 116 | "status": status, |
| 117 | "created_at": created_str, |
| 118 | "created_timestamp": created_timestamp, |
| 119 | "started_at": started_str, |
| 120 | "started_timestamp": started_timestamp, |
| 121 | "time_running": time_running_str, |
| 122 | "time_running_seconds": time_running_seconds, |
| 123 | "aws_link": aws_link, |
| 124 | "emagedoc_link": emagedoc_link, |
| 125 | "s3_link": s3_link, |
| 126 | "round_info": round_info, |
| 127 | "aws_command": aws_command, |
| 128 | } |
| 129 | |
| 130 | def _calculate_time_running( |
| 131 | self, started_at: int | datetime | None, stopped_at: int | datetime | None, status: str |
no test coverage detected