Repository: vllm-project/aibrix
File: python/aibrix/aibrix/runtime/downloaders.py (Python)

Code before:
        - session_key: Session key (optional)
        - endpoint: TOS endpoint (optional)
        - region: TOS region (optional)
        - enable_crc: Whether to enable CRC check (optional)
        """
        try:
            import tos
            from tos.exceptions import TosClientError, TosServerError
        except ImportError:
            raise ImportError(
                "tos is required for TOS downloads. Install with: pip install tos"
            )

        parsed = urlparse(source_url)
        bucket_name = parsed.netloc
        object_key = parsed.path.lstrip("/")

        if not bucket_name or not object_key:
            raise ValueError(f"Invalid TOS URL: {source_url}")

        if not credentials:
            raise ValueError("TOS credentials are required in credentials dict.")

        ak = (
            credentials.get("ak")
            or credentials.get("access-key")
            or credentials.get("access_key")
            or credentials.get("TOS_ACCESS_KEY")
        )
        sk = (
            credentials.get("sk")
            or credentials.get("secret-key")
            or credentials.get("secret_key")
            or credentials.get("TOS_SECRET_KEY")
        )
        session_key = (
            credentials.get("session_key")
            or credentials.get("tos_session_key")
            or credentials.get("TOS_SESSION_KEY", None)
        )
        if not ak or not sk:
            raise ValueError("TOS ak/sk must be provided in credentials dict.")

        endpoint = (
            credentials.get("endpoint")
            or credentials.get("tos_endpoint")
            # for more endpoint, please refer to https://www.volcengine.com/docs/6349/107356?lang=en
            or os.getenv("TOS_ENDPOINT", "tos-cn-beijing.volces.com")
        )
        region = (
            credentials.get("region")

Review comment (refactor):
em. this is kind of confusing, does TOS support `ak` or `access-key` or `access_key`? let's link a doc if there's official api documentation. If not, let's remove unnecessary envs to clean up the code

Code after:
                raise PermissionError(f"Access denied to GCS object: {source_url}")
            else:
                raise RuntimeError(f"GCS download error: {e}")


class TOSArtifactDownloader(ArtifactDownloader):
    """Download artifacts from Volcengine TOS."""

    async def download(
        self, source_url: str, local_path: str, credentials: Optional[Dict] = None
    ) -> str:
        """
        Download from TOS.

        Credentials dict should contain:
        - ak: Access key
        - sk: Secret key
        - session_key: Session key (optional)
        - endpoint: TOS endpoint (optional)
        - region: TOS region (optional)
        - enable_crc: Whether to enable CRC check (optional)
        """
        try:
            import tos
            from tos.exceptions import TosClientError, TosServerError
        except ImportError:
            raise ImportError(
                "tos is required for TOS downloads. Install with: pip install tos"
            )

        parsed = urlparse(source_url)
        bucket_name = parsed.netloc
        object_key = parsed.path.lstrip("/")

        if not bucket_name or not object_key:
            raise ValueError(f"Invalid TOS URL: {source_url}")

        if not credentials:
            raise ValueError("TOS credentials are required in credentials dict.")

        ak = credentials.get("TOS_ACCESS_KEY")
        sk = credentials.get("TOS_SECRET_KEY")
        session_key = credentials.get("TOS_SESSION_KEY", None)
        if not ak or not sk:
            raise ValueError(
                "TOS access key/secret key must be provided in credentials dict., the name should be TOS_ACCESS_KEY, TOS_SECRET_KEY"
            )

        endpoint = (
            credentials.get("endpoint")
            or credentials.get("tos_endpoint")