Function get_task_data

app/deploy/jockey/worker/queue.py:203–230

Get task data by task ID. Args: task_id: Task identifier Returns: Task data dictionary or None if not found

(task_id: str)

def get_task_data(task_id: str) -> Optional[JobData]:
    """Get task data by task ID.

    Args:
        task_id: Task identifier

    Returns:
        Task data dictionary or None if not found
    """
    cursor = 0

    while True:
        cursor, fields = _get_redis_client().hscan(
            TASKS_HASH_NAME,
            cursor,
            match=f"*:*:{task_id}",
        )

        for _, task_data in fields.items():
            try:
                return JobData(**json.loads(task_data))
            except (json.JSONDecodeError, KeyError, TypeError):
                continue

        if cursor == 0:
            break

    return None


def get_queue_length() -> int: