Import source: axon/services/jobs
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
import asyncio
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from typing import Awaitable, Callable
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
@dataclass
|
||||
class Job:
|
||||
job_id: str
|
||||
kind: str
|
||||
status: str = "queued"
|
||||
result: dict | None = None
|
||||
error: str | None = None
|
||||
|
||||
|
||||
_jobs: dict[str, Job] = {}
|
||||
|
||||
|
||||
def _dump_result(result) -> dict:
|
||||
if isinstance(result, BaseModel):
|
||||
return result.model_dump()
|
||||
if isinstance(result, dict):
|
||||
return result
|
||||
return {"value": result}
|
||||
|
||||
|
||||
def create_job(kind: str, fn: Callable[[], Awaitable]) -> Job:
|
||||
job_id = str(uuid.uuid4())
|
||||
job = Job(job_id=job_id, kind=kind)
|
||||
_jobs[job_id] = job
|
||||
|
||||
async def _runner() -> None:
|
||||
job.status = "running"
|
||||
try:
|
||||
job.result = _dump_result(await fn())
|
||||
job.status = "succeeded"
|
||||
except Exception as exc:
|
||||
job.error = str(exc)
|
||||
job.status = "failed"
|
||||
|
||||
asyncio.create_task(_runner())
|
||||
return job
|
||||
|
||||
|
||||
def get_job(job_id: str) -> Job | None:
|
||||
return _jobs.get(job_id)
|
||||
Reference in New Issue
Block a user