def prompt_for_oss_bucket()

in pai/toolkit/config.py [0:0]

• 111 lines of code
• 16 McCabe index (conditional complexity)

def prompt_for_oss_bucket(user_profile: UserProfile, workspace_id: str):
    default_storage_uri, endpoint = user_profile.get_default_oss_storage_uri(
        workspace_id=workspace_id
    )
    print(
        localized_text(
            "Please configure the OSS Bucket to be used.", "请配置使用的OSS Bucket"
        )
    )
    if user_profile.is_ram_user:
        print_warning(
            localized_text(
                "Please confirm that current RAM user has been granted permission to"
                " list buckets and access the OSS Bucket that is needed.",
                "请确认当前的RAM账号 ListBuckets 权限以及 需要使用的OSS Bucket的读写权限。",
            )
        )

    bucket_name = None
    # use the default OSS storage URI configured in the Workspace.
    if default_storage_uri and confirm(
        localized_text(
            f"Do you want to use the default OSS Bucket of the current workspace"
            f" ({default_storage_uri})?",
            f"是否使用当前工作空间的默认OSS Bucket ({default_storage_uri}) ?",
        )
    ):
        bucket_name = OssUriObj(default_storage_uri).bucket_name

    if not bucket_name:
        buckets: List[SimplifiedBucketInfo] = user_profile.list_oss_buckets()
        if not buckets:
            print(
                localized_text(
                    f"You do not have any available OSS Bucket in the region"
                    f" ({user_profile.region_id}).",
                    f"您在当前Region （{user_profile.region_id}） 下没有可用的OSS Bucket。",
                )
            )
            bucket_name = prompt_for_create_oss_bucket(user_profile, workspace_id)
        else:
            index = radio_list_prompt(
                localized_text(
                    "Please select the OSS Bucket you want to use:",
                    "请选择您需要使用的OSS Bucket：",
                ),
                values=[(idx, b.name) for idx, b in enumerate(buckets)],
                erase_when_done=True,
            )
            bucket_name = buckets[index].name

    try:
        bucket_info = user_profile.get_bucket_info(bucket_name=bucket_name)
    except oss2.exceptions.AccessDenied:
        # try to get bucket info with ListBuckets API if the user has no permission to
        # GetBucketInfo API.
        buckets = user_profile.list_oss_buckets(prefix=bucket_name)
        bucket_info = next((b for b in buckets if b.name == bucket_name), None)

    if not bucket_info:
        print_warning(
            localized_text(
                "Failed to get bucket info, use default endpoint.",
                "获取 Bucket 信息失败，使用默认 Endpoint。",
            )
        )
        region_id = user_profile.region_id
        extranet_endpoint, intranet_endpoint = (
            f"oss-{region_id}.aliyuncs.com",
            f"oss-{region_id}-internal.aliyuncs.com",
        )
    else:
        extranet_endpoint, intranet_endpoint = (
            bucket_info.extranet_endpoint,
            bucket_info.intranet_endpoint,
        )

    # If Workspace has no default OSS storage URI and user has permission to edit,
    # prompt to set the default OSS storage URI.
    if not default_storage_uri and user_profile.has_permission_edit_config(
        workspace_id=workspace_id
    ):
        prompt_for_set_default_oss_storage(
            user_profile, workspace_id, bucket_name, intranet_endpoint=intranet_endpoint
        )

    row_format = "{:<60}{}"
    intra_endpoint_connectable = is_domain_connectable(intranet_endpoint, timeout=1)
    candidates = [
        (
            intranet_endpoint,
            row_format.format(
                intranet_endpoint,
                localized_text(
                    "Internal endpoint (Please use in PAI-DSW Notebook, ECS and other "
                    "intranet environment)",
                    "内网Endpoint(请在PAI-DSW Notebook, ECS等内网环境中使用)",
                ),
            ),
        ),
        (
            extranet_endpoint,
            row_format.format(
                extranet_endpoint,
                localized_text(
                    "Public endpoint",
                    "外网Endpoint",
                ),
            ),
        ),
    ]

    if not intra_endpoint_connectable:
        candidates = candidates[::-1]

    endpoint = radio_list_prompt(
        localized_text(
            "Please select the Endpoint to access OSS:",
            "请选择访问OSS使用的Endpoint：",
        ),
        values=candidates,
        erase_when_done=True,
    )
    return bucket_name, endpoint