def list_google_storage_directories()

in workflow/__init__.py [0:0]

• 20 lines of code
• 6 McCabe index (conditional complexity)

def list_google_storage_directories(bucket_name, *, prefix=None):
    if bucket_name.startswith(kTestBucket):
        for _, dirs, _ in os.walk(get_test_dir(bucket_name)):
            return dirs

    extra_params = {"projection": "noAcl", "delimiter": "/"}

    if prefix is not None:
        if not prefix.endswith("/"):
            prefix += "/"
        extra_params["prefix"] = prefix

    gcs = storage.Client()

    path = "/b/" + bucket_name + "/o"

    iterator = page_iterator.HTTPIterator(
        client=gcs,
        api_request=gcs._connection.api_request,
        path=path,
        items_key="prefixes",
        item_to_value=_item_to_value,
        extra_params=extra_params,
    )

    return [x for x in iterator]