def format_replica_status()

in torchx/cli/cmd_status.py [0:0]

• 22 lines of code
• 7 McCabe index (conditional complexity)

def format_replica_status(replica_status: api.ReplicaStatus) -> str:
    if replica_status.structured_error_msg != NONE:
        error_data = json.loads(replica_status.structured_error_msg)
        error_message = format_error_message(
            msg=error_data["message"]["message"], header="    error_msg: "
        )
        timestamp = int(error_data["message"]["extraInfo"]["timestamp"])
        exitcode = error_data["message"]["errorCode"]
        if not exitcode:
            exitcode = "<N/A>"
        data = f"""{str(replica_status.state)} (exitcode: {exitcode})
    timestamp: {datetime.fromtimestamp(timestamp)}
    hostname: {replica_status.hostname}
{error_message}"""
    else:
        data = f"{str(replica_status.state)}"
        if replica_status.state in [
            api.ReplicaState.CANCELLED,
            api.ReplicaState.FAILED,
        ]:
            data += " (no reply file)"

    # mark index 0 for each role with a "*" for a visual queue on role boundaries
    header = " "
    if replica_status.id == 0:
        header = "*"

    return f"\n {header}{replica_status.role}[{replica_status.id}]:{data}"