def parse_oss_uri()

in oss-torch-connector/osstorchconnector/_oss_bucket_iterable.py [0:0]

• 18 lines of code
• 9 McCabe index (conditional complexity)

def parse_oss_uri(uri: str) -> Tuple[str, str]:
    if not uri or not (uri.startswith("oss://") or uri.startswith("/")):
        raise ValueError("Only oss:// URIs are supported")
    if uri.startswith("oss://"):
        uri = uri[len("oss://"):]
    elif uri.startswith("/"):
        uri = uri[1:]
    if not uri:
        raise ValueError("Bucket name must be non-empty")
    split = uri.split("/", maxsplit=1)
    if len(split) == 1:
        bucket = split[0]
        prefix = ""
    else:
        bucket, prefix = split
    if not bucket:
        raise ValueError("Bucket name must be non-empty")
    return bucket, prefix