| 67 | return record |
| 68 | |
| 69 | def update_status( |
| 70 | self, |
| 71 | job_id: str, |
| 72 | status: str, |
| 73 | error_message: Optional[str] = None, |
| 74 | result_summary: Optional[Dict[str, Any]] = None, |
| 75 | ) -> None: |
| 76 | job = self.get_job(job_id) |
| 77 | if job is None: |
| 78 | return |
| 79 | |
| 80 | now = datetime.now(timezone.utc) |
| 81 | job["status"] = status |
| 82 | |
| 83 | if status == JOB_STATUS_RUNNING: |
| 84 | job["started_at"] = now |
| 85 | elif status in (JOB_STATUS_COMPLETED, JOB_STATUS_FAILED): |
| 86 | job["completed_at"] = now |
| 87 | |
| 88 | if error_message is not None: |
| 89 | job["error_message"] = error_message |
| 90 | if result_summary is not None: |
| 91 | job["result_summary"] = json.dumps(result_summary) |
| 92 | |
| 93 | if "parameters" in job and not isinstance(job["parameters"], str): |
| 94 | job["parameters"] = ( |
| 95 | json.dumps(job["parameters"]) if job["parameters"] else None |
| 96 | ) |
| 97 | |
| 98 | if isinstance(job.get("metric_date"), str): |
| 99 | job["metric_date"] = date.fromisoformat(job["metric_date"]) |
| 100 | |
| 101 | self._offline_store.save_monitoring_metrics(self._config, "job", [job]) |
| 102 | |
| 103 | def execute_job(self, job_id: str, monitoring_service) -> Dict[str, Any]: |
| 104 | """Execute a DQM job synchronously. Manages status transitions.""" |