(job_id)
| 67 | |
| 68 | @app.route('/api/crew/<job_id>', methods=['GET']) |
| 69 | def get_status(job_id): |
| 70 | with jobs_lock: |
| 71 | job = jobs.get(job_id) |
| 72 | if job is None: |
| 73 | abort(404, description="Job not found") |
| 74 | |
| 75 | # Parse the job.result string into a JSON object |
| 76 | try: |
| 77 | result_json = json.loads(job.result) |
| 78 | except json.JSONDecodeError: |
| 79 | # If parsing fails, set result_json to the original job.result string |
| 80 | result_json = job.result |
| 81 | |
| 82 | return jsonify({ |
| 83 | "job_id": job_id, |
| 84 | "status": job.status, |
| 85 | "result": result_json, |
| 86 | "events": [{"timestamp": event.timestamp.isoformat(), "data": event.data} for event in job.events] |
| 87 | }) |
| 88 | |
| 89 | |
| 90 | if __name__ == '__main__': |
nothing calls this directly
no outgoing calls
no test coverage detected