Arvados

A lot of code in the Python SDK cookbook that coordinates low-level API calls should just be methods that are defined in the SDK itself. Why force every user to copy-paste these code snippets when we can just give them the method directly?

Here are common general-purpose container/request functions that could be useful. I'm not sure what module they should go in. It might make sense to add them to a new arvados.container module.

def lookup(
        client: ArvadosAPIClient,
        container: str | Container | ContainerRequest,
) -> Container | None: ...
# If container is a str, it's a UUID for one of the other types, fetch that.
# If it's a container request, return None if container_uuid is None,
# else fetch that.
# Return the resulting container.
# Note that lookup should return None *only* if it handles a container
# request with no container_uuid set.
# Other problems like malformed UUID, other object types, container request
# without container_uuid selected, etc. should all raise their corresponding
# exceptions.

def container_started(
        client: ArvadosAPIClient,
        container: str | Container | ContainerRequest,
) -> bool:
    container_obj = lookup(client, container)
    if container_obj is None:
        return False
    else:
        return container_obj['status'] not in {'Queued', 'Locked'}

def container_finished(
        client: ArvadosAPIClient,
        container: str | Container | ContainerRequest,
) -> bool:
    container_obj = lookup(client, container)
    if container_obj is None:
        return False
    else:
        return container_obj['status'] in {'Cancelled', 'Complete'}

def container_succeeded(
        client: ArvadosAPIClient,
        container: str | Container | ContainerRequest,
        success: typing.Container[int]=frozenset([0]),
) -> bool:
    container_obj = lookup(client, container)
    return (
        container_obj is not None
        and container_obj['status'] == 'Complete'
        and container_obj['exit_code'] in success
    )

def child_requests(
        client: ArvadosAPIClient,
        container: str | Container | ContainerRequest,
        filters: list[list[str]]=[],
        select: list[str]=[],
        depth: int | None=None,
) -> Iterator[ContainerRequest]: ...
# lookup container, yield from `keyset_list_all` calls, down to depth if given

def child_containers(
        client: ArvadosAPIClient,
        container: str | Container | ContainerRequest,
        request_filters: list[list[str]]=[],
        container_filters: list[list[str]]=[],
        select: list[str]=[],
        depth: int | None=None,
) -> Iterator[Container]: ...
# Same dance as above, call `child_requests` to get container_uuids, then
# yield from `keyset_list_all` calls with those UUIDs, down to depth if given.
# Note that `child_requests` can be called with a very narrow `select` for
# optimization, and can add the filter `container_uuid is not null`.