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

Fix inputs JSON parsing in pending_runs database functions

Browse Source 
JSONB columns may return strings in some cases - explicitly parse
inputs field to ensure it's always a list.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-11 20:38:19 +00:00
parent 8ab0f05a7d
commit d92e493689
1 changed files with 15 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
database.py
View File

@@ -1428,12 +1428,16 @@ async def get_pending_run(run_id: str) -> Optional[dict]:
run_id
)
if row:
# Parse inputs if it's a string (JSONB should auto-parse but be safe)
inputs = row["inputs"]
if isinstance(inputs, str):
inputs = _json.loads(inputs)
return {
"run_id": row["run_id"],
"celery_task_id": row["celery_task_id"],
"status": row["status"],
"recipe": row["recipe"],
"inputs": row["inputs"],
"inputs": inputs,
"dag_json": row["dag_json"],
"output_name": row["output_name"],
"actor_id": row["actor_id"],
@@ -1472,21 +1476,25 @@ async def list_pending_runs(actor_id: Optional[str] = None, status: Optional[str
""",
*params
)
return [
{
results = []
for row in rows:
# Parse inputs if it's a string
inputs = row["inputs"]
if isinstance(inputs, str):
inputs = _json.loads(inputs)
results.append({
"run_id": row["run_id"],
"celery_task_id": row["celery_task_id"],
"status": row["status"],
"recipe": row["recipe"],
"inputs": row["inputs"],
"inputs": inputs,
"output_name": row["output_name"],
"actor_id": row["actor_id"],
"error": row["error"],
"created_at": row["created_at"].isoformat() if row["created_at"] else None,
"updated_at": row["updated_at"].isoformat() if row["updated_at"] else None,
}
for row in rows
]
})
return results
async def update_pending_run_status(run_id: str, status: str, error: Optional[str] = None) -> bool: