(job_id: str, *, task: str, params: dict[str, Any])
| 31 | |
| 32 | |
| 33 | def create_job(job_id: str, *, task: str, params: dict[str, Any]) -> Job: |
| 34 | now = time.time() |
| 35 | job = Job(id=job_id, task=task, params=params, created_at=now) |
| 36 | r.set(JOB_KEY.format(job_id), job.model_dump_json()) |
| 37 | r.zadd(JOBS_SET, {job_id: now}) |
| 38 | return job |
| 39 | |
| 40 | |
| 41 | def get_job(job_id: str) -> Job | None: |
no test coverage detected