art-dag/celery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add _ensure_inputs_list to handle legacy Redis data format

Browse Source 
Inputs stored in old Redis format are JSON strings - this helper
ensures they're always returned as lists regardless of source.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-11 21:27:49 +00:00
parent cca8f74d3c
commit 59de1cf6b5
1 changed files with 20 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
app/services/run_service.py
View File

@@ -83,6 +83,22 @@ class RunService:
self.task_key_prefix = "artdag:task:" # run_id -> task_id mapping only self.task_key_prefix = "artdag:task:" # run_id -> task_id mapping only
self.cache_dir = Path(os.environ.get("CACHE_DIR", "/tmp/artdag-cache")) self.cache_dir = Path(os.environ.get("CACHE_DIR", "/tmp/artdag-cache"))
def _ensure_inputs_list(self, inputs) -> list:
"""Ensure inputs is a list, parsing JSON string if needed."""
if inputs is None:
return []
if isinstance(inputs, list):
return inputs
if isinstance(inputs, str):
try:
parsed = json.loads(inputs)
if isinstance(parsed, list):
return parsed
except json.JSONDecodeError:
pass
return []
return []
async def get_run(self, run_id: str) -> Optional[Dict[str, Any]]: async def get_run(self, run_id: str) -> Optional[Dict[str, Any]]:
"""Get a run by ID. Checks database first, then Celery task state.""" """Get a run by ID. Checks database first, then Celery task state."""
# Check database for completed run # Check database for completed run
@@ -92,7 +108,7 @@ class RunService:
"run_id": run_id, "run_id": run_id,
"status": "completed", "status": "completed",
"recipe": cached.get("recipe"), "recipe": cached.get("recipe"),
"inputs": cached.get("inputs", []), "inputs": self._ensure_inputs_list(cached.get("inputs")),
"output_hash": cached.get("output_hash"), "output_hash": cached.get("output_hash"),
"ipfs_cid": cached.get("ipfs_cid"), "ipfs_cid": cached.get("ipfs_cid"),
"provenance_cid": cached.get("provenance_cid"), "provenance_cid": cached.get("provenance_cid"),
@@ -130,7 +146,7 @@ class RunService:
"celery_task_id": task_id, "celery_task_id": task_id,
"actor_id": pending.get("actor_id"), "actor_id": pending.get("actor_id"),
"recipe": pending.get("recipe"), "recipe": pending.get("recipe"),
"inputs": pending.get("inputs"), "inputs": self._ensure_inputs_list(pending.get("inputs")),
"output_name": pending.get("output_name"), "output_name": pending.get("output_name"),
"created_at": pending.get("created_at"), "created_at": pending.get("created_at"),
"error": pending.get("error"), "error": pending.get("error"),
@@ -154,7 +170,7 @@ class RunService:
"run_id": run_id, "run_id": run_id,
"status": pending.get("status", "pending"), "status": pending.get("status", "pending"),
"recipe": pending.get("recipe"), "recipe": pending.get("recipe"),
"inputs": pending.get("inputs"), "inputs": self._ensure_inputs_list(pending.get("inputs")),
"output_name": pending.get("output_name"), "output_name": pending.get("output_name"),
"actor_id": pending.get("actor_id"), "actor_id": pending.get("actor_id"),
"created_at": pending.get("created_at"), "created_at": pending.get("created_at"),
@@ -215,7 +231,7 @@ class RunService:
"celery_task_id": task_id, "celery_task_id": task_id,
"actor_id": task_actor_id, "actor_id": task_actor_id,
"recipe": task_recipe, "recipe": task_recipe,
"inputs": task_inputs, "inputs": self._ensure_inputs_list(task_inputs),
"output_name": task_output_name, "output_name": task_output_name,
"created_at": task_created_at, "created_at": task_created_at,
} }