"""High-level public API facade for the ModelScope Hub SDK. :class:`HubApi` is the **only** entry point users should construct. It composes the low-level :class:`OpenAPIClient`, :class:`LegacyClient`, :class:`DownloadManager`, :class:`UploadManager` and the cache helpers into a unified, OpenAPI-first surface. Design principles ----------------- * **OpenAPI-first** — every operation that has an OpenAPI counterpart goes through :mod:`._openapi`. Legacy endpoints are used only as a transparent fallback when no OpenAPI equivalent exists. * **Unified repo pattern** — every repository operation accepts a ``repo_type`` parameter; there are no type-specific methods like ``create_model`` or ``get_dataset``. * **Transparent fallback** — callers do not need to know which path served their request. * **Lazy clients** — the underlying HTTP clients are instantiated on demand so that ``HubApi()`` never fails just because no token is present. * **SOLID** — :class:`HubApi` only routes and orchestrates; concrete network logic lives in the injected dependencies. """ from __future__ import annotations from collections.abc import Iterable, Mapping from pathlib import Path from typing import Any, BinaryIO, TypeAlias from urllib.parse import urlparse from requests.cookies import RequestsCookieJar from ._cache_manager import _resolve_verification_root from ._cache_manager import clear_cache as _clear_cache from ._cache_manager import scan_cache as _scan_cache from ._cache_manager import verify_cache as _verify_cache from ._download import DownloadManager, ProgressCallback from ._legacy_api import LegacyClient from ._openapi import OpenAPIClient from ._upload import UploadManager from .config import HubConfig, get_default_config from .constants import DEFAULT_ENDPOINT, RepoType, Visibility from .errors import ( AuthenticationError, HubError, InvalidParameter, NetworkError, NotExistError, NotSupportedError, ) from .types import CacheInfo, CacheVerification, FileInfo, PagedResult, RepoInfo, UserInfo from .utils.logger import get_logger __all__ = ["HubApi"] logger = get_logger("api") RepoTypeLike: TypeAlias = "str | RepoType" # Routing tables — declarative dispatch keeps :class:`HubApi` free of long # if/elif chains and makes adding new repo types a one-line change. _CREATABLE_TYPES: frozenset[RepoType] = frozenset({RepoType.MODEL, RepoType.DATASET, RepoType.STUDIO, RepoType.SKILL}) _OPENAPI_CREATE_TYPES: frozenset[RepoType] = frozenset({RepoType.STUDIO, RepoType.SKILL}) _OPENAPI_DETAIL_TYPES: frozenset[RepoType] = frozenset( {RepoType.MODEL, RepoType.DATASET, RepoType.STUDIO, RepoType.SKILL} ) # Mapping of common license display names to their SPDX identifiers. The Hub # backend rejects display names like ``"Apache License 2.0"`` — we translate # them transparently while passing unknown values (already SPDX) through. _LICENSE_DISPLAY_TO_SPDX: dict[str, str] = { "Apache License 2.0": "apache-2.0", "MIT License": "mit", "GPL-2.0": "gpl-2.0", "GPL-3.0": "gpl-3.0", "LGPL-2.1": "lgpl-2.1", "LGPL-3.0": "lgpl-3.0", "AFL-3.0": "afl-3.0", "ECL-2.0": "ecl-2.0", "BSD-2-Clause": "bsd-2-clause", "BSD-3-Clause": "bsd-3-clause", "CC-BY-4.0": "cc-by-4.0", "CC-BY-SA-4.0": "cc-by-sa-4.0", "CC-BY-NC-4.0": "cc-by-nc-4.0", "CC0-1.0": "cc0-1.0", "Unlicense": "unlicense", } _STUDIO_FIELD_RENAMES: dict[str, str] = { "cover_image": "coverImage", } # Reserved fields that are controlled by create_repo method parameters. # These MUST NOT be overridden via extra kwargs to avoid silent conflicts. _RESERVED_EXTRA_FIELDS: frozenset[str] = frozenset({"Path", "Owner", "Name"}) class HubApi: """Unified client for ModelScope Hub operations. Provides a high-level interface for repository management, file operations, deployment, secret management and local caching. All repo-type-specific operations use a unified ``repo_type`` parameter following the OpenAPI-First design — there are no type-specific methods like ``create_model`` or ``get_dataset``. Internally the class composes :class:`OpenAPIClient`, :class:`LegacyClient`, :class:`DownloadManager`, :class:`UploadManager` and the cache helpers. HTTP clients are instantiated lazily, so ``HubApi()`` never fails just because no token is present. Parameters ---------- config : HubConfig or None, optional Pre-built configuration. When omitted, the process-wide default from :func:`get_default_config` is used (which reads the ``MODELSCOPE_API_TOKEN`` env var and the local config file). endpoint : str or None, optional Override the API endpoint. Takes precedence over ``config.endpoint``. Defaults to ``https://modelscope.cn``. token : str or None, optional Override the API token. Takes precedence over ``config.token``. Examples -------- >>> from modelscope_hub import HubApi >>> api = HubApi(token="ms-xxxxxxxx") >>> user = api.whoami() >>> user.username 'alice' Create and manage repositories: >>> api.create_repo("alice/my-model", repo_type="model", visibility="private") >>> api.upload_file("alice/my-model", "model", "./weights.bin", "weights.bin") >>> path = api.download_file("alice/my-model", "model", "weights.bin") """ def __init__( self, config: HubConfig | None = None, *, endpoint: str | None = None, token: str | None = None, ) -> None: base = config or get_default_config() if config is None and (endpoint is not None or token is not None): from dataclasses import replace was_overridden = base._endpoint_overridden base = replace(base) # replace() re-runs __post_init__ which sees the inherited # endpoint string as "explicit" and sets _endpoint_overridden. # Restore the original state so resolve_endpoint_for_read works. base._endpoint_overridden = was_overridden self._config = base if endpoint is not None: self._config.endpoint = HubConfig.normalize_endpoint(endpoint) self._config._endpoint_overridden = True if token is not None: self._config.token = token self._config._token_overridden = True self._openapi: OpenAPIClient | None = None self._legacy: LegacyClient | None = None self._downloader: DownloadManager | None = None self._uploader: UploadManager | None = None # ================================================================== # Lazy client accessors # ================================================================== @property def openapi(self) -> OpenAPIClient: """Lazily-constructed OpenAPI client.""" if self._openapi is None: self._openapi = OpenAPIClient(self._config) return self._openapi @property def legacy(self) -> LegacyClient: """Lazily-constructed legacy ``/api/v1`` client.""" if self._legacy is None: from .utils import build_user_agent self._legacy = LegacyClient( token=self._config.token, endpoint=self._config.endpoint or DEFAULT_ENDPOINT, user_agent=build_user_agent(self._config.get_session_id()), ) elif self._legacy.token != self._config.token: # Clears propagate as well as changes: a cached client left holding a # revoked token would keep authenticating with it. self._legacy.token = self._config.token return self._legacy @property def downloader(self) -> DownloadManager: """Lazily-constructed :class:`DownloadManager`.""" if self._downloader is None: self._downloader = DownloadManager(self.legacy, self._config) return self._downloader @property def uploader(self) -> UploadManager: """Lazily-constructed :class:`UploadManager`. The OpenAPI client is injected so small files (≤ 5 MiB) flow through ``POST /files/upload`` instead of the legacy commit endpoint. """ if self._uploader is None: self._uploader = UploadManager( self.legacy, self._config, self.openapi, create_repo_fn=self._create_repo_exist_ok, ) return self._uploader # ================================================================== # Static helpers # ================================================================== @staticmethod def _parse_repo_id(repo_id: str) -> tuple[str, str]: """Split a canonical ``owner/name`` identifier into its two halves.""" if not repo_id or "/" not in repo_id: raise InvalidParameter(f"repo_id {repo_id!r} should be in format of 'owner/name'.") owner, _, name = repo_id.partition("/") if not owner or not name: raise InvalidParameter( f"repo_id {repo_id!r} should be in format of 'owner/name': owner and name must both be non-empty." ) return owner, name def _create_repo_exist_ok(self, repo_id: str, repo_type: str) -> None: """Auto-create the repo if it doesn't exist, silently ignore if it does.""" try: self.create_repo(repo_id, repo_type) except HubError: pass @staticmethod def _normalize_repo_type(repo_type: RepoTypeLike) -> RepoType: """Coerce a ``str`` or :class:`RepoType` value to a :class:`RepoType`.""" if isinstance(repo_type, RepoType): return repo_type try: return RepoType(str(repo_type).lower()) except ValueError as exc: allowed = ", ".join(t.value for t in RepoType) raise InvalidParameter(f"Unknown repo_type {repo_type!r}. Expected one of: {allowed}.") from exc @staticmethod def _normalize_visibility(visibility: int | str | Visibility | None) -> int | None: """Normalise visibility input to its integer wire encoding.""" if visibility is None: return None if isinstance(visibility, Visibility): return int(visibility) if isinstance(visibility, int): return visibility return int(Visibility.from_label(str(visibility))) _PAGED_ITEM_KEYS = ( "items", "list", "data", "results", "models", "datasets", "skills", "servers", "mcp_server_list", "Models", "Datasets", "Skills", "Servers", ) _PAGED_META_KEYS = frozenset( { "total_count", "total", "page_number", "page", "page_size", "size", "TotalCount", "Total", "PageNumber", "PageSize", } ) @staticmethod def _extract_paged(payload: Any) -> tuple[list[Any], int, int, int]: """Decode a paginated OpenAPI response into ``(items, total, page, size)``. The ModelScope API returns item arrays under type-specific keys (``models``, ``datasets``, ``skills``, ``servers``). This method checks known keys first, then falls back to the first list-valued key that is not pagination metadata. """ if isinstance(payload, list): return payload, len(payload), 1, len(payload) if not isinstance(payload, dict): return [], 0, 1, 0 items: list[Any] = [] for key in HubApi._PAGED_ITEM_KEYS: if isinstance(payload.get(key), list): items = payload[key] break else: for key, value in payload.items(): if isinstance(value, list) and key not in HubApi._PAGED_META_KEYS: items = value break def _first(keys: tuple[str, ...], default: int) -> int: for k in keys: v = payload.get(k) if v is not None: return int(v) return default total = _first(("total_count", "TotalCount", "total"), len(items)) page = _first(("page_number", "PageNumber", "page"), 1) size = _first(("page_size", "PageSize", "size"), len(items)) return items, total, page, size @staticmethod def _repo_info_from_payload( data: Mapping[str, Any] | None, repo_type: RepoType, *, owner_hint: str | None = None, name_hint: str | None = None, ) -> RepoInfo: """Build a :class:`RepoInfo` from an arbitrary API payload. The legacy and OpenAPI surfaces use different field-naming conventions (PascalCase vs snake_case). This helper normalises both into the SDK's canonical dataclass. """ data = dict(data or {}) # PascalCase → snake_case shims for legacy responses. normalised: dict[str, Any] = {} aliases = { "Id": "id", "Path": "owner", "Name": "name", "Owner": "owner", "Visibility": "visibility", "License": "license", "Description": "description", "Downloads": "downloads", "Likes": "likes", "CreatedAt": "created_at", "UpdatedAt": "last_modified", "LastModified": "last_modified", "last_modified": "last_modified", "updated_at": "last_modified", "Tags": "tags", } for key, value in data.items(): normalised[aliases.get(key, key)] = value # The OpenAPI surface uses ``private`` bool for visibility. # gated is orthogonal and does not affect visibility mapping. if normalised.get("visibility") is None: private_flag = normalised.get("private") if isinstance(private_flag, bool): if private_flag: normalised["visibility"] = Visibility.PRIVATE else: normalised["visibility"] = Visibility.PUBLIC elif normalised.get("gated"): # Rule 4: visibility unknown + gated=True → imply PRIVATE normalised["visibility"] = Visibility.PRIVATE # The OpenAPI list endpoints return ``id`` as "owner/name". # Split it so the computed ``repo_id`` property works. id_val = normalised.get("id") if isinstance(id_val, str) and "/" in id_val: parts = id_val.split("/", 1) if not normalised.get("owner"): normalised["owner"] = parts[0] if not normalised.get("name"): normalised["name"] = parts[1] if not normalised.get("owner"): normalised["owner"] = owner_hint if not normalised.get("name"): normalised["name"] = name_hint normalised["repo_type"] = repo_type return RepoInfo.from_dict(normalised) # ================================================================== # Authentication # ================================================================== def get_cookies( self, access_token: str | None = None, *, cookies_required: bool = False, ) -> RequestsCookieJar | None: """Get cookies for authentication from token or local cache. Resolution order: 1. Explicit ``access_token`` argument 2. Token from config (explicit arg > env var > persisted cookie) 3. Saved cookies from ``~/.modelscope/credentials/cookies`` When a token is available (steps 1-2), a fresh :class:`~requests.cookies.RequestsCookieJar` with ``m_session_id`` is built. Otherwise the locally cached cookies from a prior ``login()`` call are loaded. Parameters ---------- access_token : str, optional Explicit token override. cookies_required : bool, optional When ``True``, raise :class:`AuthenticationError` if no credentials are available. Default is ``False``. Returns ------- RequestsCookieJar or None Cookie jar for authentication, or ``None`` when no credentials are available and ``cookies_required`` is ``False``. Raises ------ AuthenticationError When ``cookies_required`` is ``True`` and no credentials found. Examples -------- >>> cookies = api.get_cookies() >>> cookies['m_session_id'] 'ms-xxxxxxxx' """ token = access_token or self._config.token if token: domain = urlparse(self._config.endpoint).hostname or "" jar = RequestsCookieJar() jar.set("m_session_id", token, domain=domain, path="/") return jar # An explicitly overridden (empty) token means "run without local # credentials" -- never silently fall back to the persisted cookies. if not getattr(self._config, "_token_overridden", False): cookies = self._config.load_cookies() if cookies is not None: return cookies if cookies_required: raise AuthenticationError( "No credentials found. " "Pass --token, call HubApi.login(), or set MODELSCOPE_API_TOKEN. " "Your token is available at https://modelscope.cn/my/myaccesstoken" ) return None def login(self, token: str) -> UserInfo: """Authenticate and persist credentials locally. Calls ``POST /api/v1/login`` to obtain server-issued session cookies and a git access token, then saves them to ``~/.modelscope/credentials/`` (compatible with the old modelscope SDK). Parameters ---------- token : str ModelScope API token. Must be non-empty after stripping. Returns ------- UserInfo Profile of the authenticated user. Raises ------ InvalidParameter When ``token`` is empty or whitespace-only. AuthenticationError When the server rejects the token. The server's own explanation is preserved, and an endpoint hint is appended when the token turns out to be valid on the peer ModelScope site. HubError Transport, timeout and server-side failures propagate unchanged -- they are never reported as a rejected token. Notes ----- A failed attempt leaves persisted credentials untouched. Until the server has accepted the new token, the stored credential is still the caller's only working one, so revoking it on failure would turn a mistyped token into an unintended logout. Examples -------- >>> api = HubApi() >>> user = api.login("ms-xxxxxxxx") >>> user.username 'alice' """ if not token or not token.strip(): raise InvalidParameter("token must be a non-empty string") token = token.strip() previous_token = self._config.token previous_logged_out = self._config._logged_out self._config.token = token self._config._logged_out = False self._openapi = None if self._legacy is not None: self._legacy.token = token try: data, cookies = self.legacy.login(token) except HubError as exc: self._restore_credential_state(previous_token, previous_logged_out) explained = self._explain_login_failure(token, exc) if explained is exc: raise raise explained from exc git_token = data.get("AccessToken", "") username = data.get("Username", "") email = data.get("Email", "") self._config.save_cookies(cookies) if git_token: self._config.save_git_token(git_token) if username: self._config.save_user_info(username, email or "") return self.whoami() def _restore_credential_state(self, token: str | None, logged_out: bool) -> None: """Roll the in-memory credential back to its pre-login value. Persisted credentials are deliberately left alone; only this instance's transient state is rewound, so a failed attempt leaves the object exactly as it was found instead of poisoning it with a rejected token. """ self._config.token = token self._config._logged_out = logged_out self._openapi = None if self._legacy is not None: self._legacy.token = token def _explain_login_failure(self, token: str, exc: HubError) -> HubError: """Return the exception to surface for a failed login attempt. Only authentication failures are re-worded. Network, timeout and server-side errors are handed back untouched, because presenting them as a rejected token would send the caller after the wrong remedy. The two ModelScope sites keep separate account systems and answer an unknown token with the same business code, so the server cannot tell "invalid token" apart from "token issued by the other site". Only the client knows which site it addressed, which is why that disambiguation has to happen here. """ if not isinstance(exc, AuthenticationError): return exc peer = self._peer_site_endpoint() if peer is None or not self._token_valid_on(token, peer): return exc return AuthenticationError( f"{exc.message} This token is valid on {peer} instead; retry with " f"--endpoint {peer} (or set MODELSCOPE_ENDPOINT={peer}).", status_code=exc.status_code, request_id=exc.request_id, response_body=exc.response_body, url=exc.url, method=exc.method, ) def _peer_site_endpoint(self) -> str | None: """Return the sibling ModelScope site, or ``None`` when not applicable. An explicitly configured endpoint is always respected, mirroring :meth:`resolve_endpoint_for_read`: when the caller has pinned a site we do not second-guess it. """ if self._config._endpoint_overridden: return None from .constants import DEFAULT_INTL_ENDPOINT def site_key(url: str) -> str: host = (urlparse(url).hostname or "").lower() return host[4:] if host.startswith("www.") else host current = site_key(self._config.endpoint or DEFAULT_ENDPOINT) for candidate in (DEFAULT_ENDPOINT, DEFAULT_INTL_ENDPOINT): if site_key(candidate) != current: return candidate return None @staticmethod def _token_valid_on(token: str, endpoint: str) -> bool: """Best-effort check of whether *token* authenticates against *endpoint*. Runs on the failure path only and is strictly advisory: any error means "cannot confirm", so a probe outage degrades to the plain server message rather than producing a misleading hint. Retries are disabled to keep the failure path responsive. """ from .constants import API_CONNECT_TIMEOUT probe = LegacyClient( token=None, endpoint=endpoint, timeout=API_CONNECT_TIMEOUT, max_retries=0, ) try: probe.login(token) except Exception: # advisory only -- never mask the original failure return False return True def logout(self) -> None: """Clear the locally persisted token. Cached HTTP clients are reset so subsequent calls behave as if no credential was ever provided. Examples -------- >>> api.logout() """ self._config.clear_token() self._openapi = None if self._legacy is not None: self._legacy.token = None def whoami(self) -> UserInfo: """Return the profile for the currently authenticated user. Returns ------- UserInfo Authenticated user profile (username, email, avatar, ...). Raises ------ AuthenticationError When no token is configured or the token is invalid. Examples -------- >>> from modelscope_hub import HubApi >>> api = HubApi(token="ms-xxxxxxxx") >>> user = api.whoami() >>> print(user.username, user.email) alice alice@example.com """ payload = self.openapi.get_current_user() return UserInfo.from_dict(payload if isinstance(payload, dict) else {}) # ================================================================== # Unified repo CRUD # ================================================================== def create_repo( self, repo_id: str, repo_type: RepoTypeLike, *, visibility: int | str | Visibility | None = None, license: str | None = None, chinese_name: str | None = None, description: str | None = None, gated_mode: bool | None = None, **extra: Any, ) -> RepoInfo: """Create a new repository. Routing is decided by ``repo_type``: * ``studio`` / ``skill`` → OpenAPI ``POST /studios`` / ``POST /skills`` * ``model`` / ``dataset`` → legacy ``POST /api/v1/{type}s`` Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType One of ``"model"``, ``"dataset"``, ``"studio"``, ``"skill"``. visibility : int, str or Visibility, optional Visibility level. Accepts the integer wire encoding, a label (``"public"`` / ``"private"``) or a :class:`Visibility` value. Defaults to public. license : str, optional SPDX-style license identifier (e.g. ``"apache-2.0"``). chinese_name : str, optional Chinese display name shown on the Hub UI. description : str, optional Short description of the repository. gated_mode : bool, optional Enable gated (application-based download) mode for private repos. True = gated, False = normal private. Only effective when visibility is PRIVATE; ignored otherwise. **extra : Any Additional fields forwarded verbatim to the underlying client. Returns ------- RepoInfo Metadata of the newly created repository. Raises ------ InvalidParameter When ``repo_id`` does not have the ``owner/name`` shape. AuthenticationError When the token is missing or invalid. Examples -------- Create a private model repository: >>> info = api.create_repo( ... "alice/llama-7b-finetuned", ... repo_type="model", ... visibility="private", ... license="apache-2.0", ... description="A LoRA fine-tune of LLaMA-7B", ... ) >>> info.repo_id 'alice/llama-7b-finetuned' Create a private gated dataset: >>> api.create_repo("alice/my-data", "dataset", visibility="private", gated_mode=True) Create a public Studio space: >>> api.create_repo("alice/chat-demo", repo_type="studio", visibility="public") """ rt = self._normalize_repo_type(repo_type) if rt not in _CREATABLE_TYPES: supported = ", ".join(sorted(t.value for t in _CREATABLE_TYPES)) raise NotSupportedError( f"create_repo does not support repo_type={rt.value!r}. Supported types: {supported}." ) owner, name = self._parse_repo_id(repo_id) vis = self._normalize_visibility(visibility) if license is not None: license = _LICENSE_DISPLAY_TO_SPDX.get(license, license) if rt in _OPENAPI_CREATE_TYPES: is_private = vis is not None and vis == int(Visibility.PRIVATE) if rt is RepoType.STUDIO: payload: dict[str, Any] = { "owner": owner, "repo_name": name, } if vis is not None: payload["private"] = is_private if chinese_name is not None: payload["display_name"] = chinese_name else: payload = { "owner": owner, "skill_name": name, } if vis is not None: payload["private"] = is_private if chinese_name is not None: payload["display_name"] = chinese_name if license is not None: payload["license"] = license if description is not None: payload["description"] = description for old_key, new_key in _STUDIO_FIELD_RENAMES.items(): if old_key in extra: extra[new_key] = extra.pop(old_key) payload.update(extra) data = self.openapi.create_studio(payload) if rt is RepoType.STUDIO else self.openapi.create_skill(payload) return self._repo_info_from_payload(data, rt, owner_hint=owner, name_hint=name) if rt is RepoType.DATASET: body: dict[str, Any] = { "Owner": owner, "Name": name, "Visibility": vis if vis is not None else int(Visibility.PUBLIC), "License": license or "Apache-2.0", } else: body = { "Path": owner, "Name": name, "Visibility": vis if vis is not None else int(Visibility.PUBLIC), "License": license or "Apache-2.0", } if chinese_name is not None: body["ChineseName"] = chinese_name if description is not None: body["Description"] = description # gated_mode → ProtectedMode wire field (1=gated, 2=off). # gated only effective with PRIVATE; when vis=None + gated=True, # implicitly set visibility to PRIVATE (user intent: gated repo). if gated_mode is not None: if vis is None: vis = int(Visibility.PRIVATE) body["Visibility"] = vis body["ProtectedMode"] = 1 if gated_mode else 2 elif vis == int(Visibility.PRIVATE): body["ProtectedMode"] = 1 if gated_mode else 2 else: logger.warning("gated_mode is only effective when visibility is PRIVATE, ignored.") # Blocklist filtering: only reject fields controlled by method params. filtered: dict[str, Any] = {} for k, v in extra.items(): if k in _RESERVED_EXTRA_FIELDS: logger.warning("Reserved field %r in extra ignored (controlled by method params)", k) continue filtered[k] = v if "ProtectedMode" in filtered: pm = filtered["ProtectedMode"] if not isinstance(pm, int) or isinstance(pm, bool) or pm not in (1, 2): raise ValueError("ProtectedMode must be int 1 (gated) or 2 (off); use gated_mode=True/False instead") body.update(filtered) data = self.legacy.create_repo(repo_type=str(rt), body=body) return self._repo_info_from_payload(data, rt, owner_hint=owner, name_hint=name) def get_repo( self, repo_id: str, repo_type: RepoTypeLike, *, revision: str | None = None, # noqa: ARG002 - reserved for future use ) -> RepoInfo: """Fetch a repository's metadata via the OpenAPI surface. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType One of ``"model"``, ``"dataset"``, ``"studio"``, ``"skill"``, ``"mcp"``. revision : str, optional Reserved for future use; currently ignored. Returns ------- RepoInfo Repository metadata (id, owner, name, visibility, stats, ...). Raises ------ NotExistError When the repository does not exist or is not visible to the caller. AuthenticationError When the request requires auth and the token is missing or invalid. Examples -------- >>> info = api.get_repo("alice/llama-7b", repo_type="model") >>> info.visibility 'public' >>> info.downloads 1234 """ rt = self._normalize_repo_type(repo_type) owner, name = self._parse_repo_id(repo_id) if rt is RepoType.MODEL: try: data = self.openapi.get_model(owner, name) except NotExistError: data = self.legacy.get_repo_info(repo_id, str(rt)) elif rt is RepoType.DATASET: try: data = self.openapi.get_dataset(owner, name) except NotExistError: logger.debug( "Dataset %s/%s not found in OpenAPI, falling back to legacy API", owner, name, ) data = self.legacy.get_repo_info(repo_id, str(rt)) elif rt is RepoType.STUDIO: data = self.openapi.get_studio(owner, name) elif rt is RepoType.SKILL: data = self.openapi.get_skill(f"{owner}/{name}") elif rt is RepoType.MCP: data = self.openapi.get_mcp_server(f"{owner}/{name}") else: # pragma: no cover - defensive raise NotSupportedError(f"get_repo not supported for {rt}") return self._repo_info_from_payload(data, rt, owner_hint=owner, name_hint=name) def list_repos( self, repo_type: RepoTypeLike, *, owner: str | None = None, search: str | None = None, sort: str | None = None, page_number: int = 1, page_size: int = 10, **filters: Any, ) -> PagedResult[RepoInfo]: """List repositories of the given type via OpenAPI. Parameters ---------- repo_type : str or RepoType One of ``"model"``, ``"dataset"``, ``"skill"``, ``"mcp"``. ``"studio"`` raises :class:`NotSupportedError` (no list endpoint). owner : str, optional Restrict results to repositories owned by this user/org. search : str, optional Free-text search query. sort : str, optional Sort key understood by the upstream endpoint (e.g. ``"downloads"``). page_number : int, optional 1-based page index. Default is 1. page_size : int, optional Items per page. Default is 10. **filters : Any Additional filter fields. ``None`` values are dropped. Returns ------- PagedResult[RepoInfo] Paginated repository listing. Raises ------ NotSupportedError When ``repo_type`` is ``"studio"`` (no list endpoint yet). Examples -------- Browse public LLaMA models: >>> page = api.list_repos("model", search="llama", page_size=5) >>> page.total_count 42 >>> [r.repo_id for r in page.items] ['meta-llama/Llama-2-7b', ...] List datasets owned by an organisation: >>> api.list_repos("dataset", owner="my_org", page_number=2) """ rt = self._normalize_repo_type(repo_type) clean_filters: dict[str, Any] = {k: v for k, v in filters.items() if v is not None} if rt is RepoType.MODEL: payload = self.openapi.list_models( search=search, owner=owner, sort=sort, page_number=page_number, page_size=page_size, filters=clean_filters or None, ) elif rt is RepoType.DATASET: payload = self.openapi.list_datasets( search=search, owner=owner, sort=sort, page_number=page_number, page_size=page_size, filters=clean_filters or None, ) elif rt is RepoType.SKILL: if owner: clean_filters.setdefault("owner", owner) payload = self.openapi.list_skills( search=search, page_number=page_number, page_size=page_size, filters=clean_filters or None, ) elif rt is RepoType.MCP: payload = self.openapi.list_mcp_servers( search=search, page_number=page_number, page_size=page_size, filter=clean_filters or None, ) elif rt is RepoType.STUDIO: raise NotSupportedError("Listing studios is not supported by the OpenAPI surface yet.") else: # pragma: no cover - defensive raise NotSupportedError(f"list_repos not supported for {rt}") items, total, page, size = self._extract_paged(payload) # MCP response omits page_number/page_size — use requested values. if rt is RepoType.MCP: page = page_number size = page_size infos = [self._repo_info_from_payload(item, rt) for item in items] # Determine collection key for OpenAPI-aligned to_dict() output _COLLECTION_KEYS = { RepoType.MODEL: "models", RepoType.DATASET: "datasets", RepoType.SKILL: "skills", RepoType.MCP: "servers", } key = _COLLECTION_KEYS.get(rt, "items") return PagedResult(items=infos, total_count=total, page_number=page, page_size=size, collection_key=key) def delete_repo(self, repo_id: str, repo_type: RepoTypeLike) -> None: """Delete a repository. .. deprecated:: Programmatic repository deletion is not currently supported by the Hub API for security reasons. This method will be restored in a future release once proper token-scoped authentication is available. To delete a repository now, use the web console at https://modelscope.cn. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type (``"model"``, ``"dataset"``, etc.). """ import warnings warnings.warn( "This function is deprecated due to security reasons, " "and will be recovered in future versions with proper token authentication. " "Please go to https://modelscope.cn to delete repositories via the web console.", DeprecationWarning, stacklevel=2, ) rt = self._normalize_repo_type(repo_type) self._parse_repo_id(repo_id) self.legacy.delete_repo(repo_id=repo_id, repo_type=str(rt)) def repo_exists(self, repo_id: str, repo_type: RepoTypeLike) -> bool: """Return ``True`` iff the repository exists and is visible to the caller. This is a thin wrapper around :meth:`get_repo` that converts a :class:`NotExistError` into a boolean. Examples -------- >>> api.repo_exists("alice/my-model", "model") True """ try: self.get_repo(repo_id, repo_type) return True except NotExistError: return False def resolve_endpoint_for_read( self, repo_id: str, *, repo_type: RepoTypeLike = "model", token: str | None = None, ) -> str: """Resolve the best endpoint for read operations (download, list, get). 1. If the endpoint was explicitly configured (via constructor arg, ``MODELSCOPE_ENDPOINT``, or the deprecated ``MODELSCOPE_DOMAIN``), trust the user's configuration and return it directly (no probe). 2. If ``MODELSCOPE_PREFER_AI_SITE=true``, check ``.ai`` first, then fall back to ``.cn``. 3. Otherwise (default), check ``.cn`` first, then fall back to ``.ai``. Parameters ---------- token : str, optional Explicit token for the probe requests. Falls back to the token stored in this instance's config. Returns ------- str The endpoint URL where the repo exists. Raises ------ NotExistError If the repo is not found on any checked endpoint. """ from .constants import ( DEFAULT_ENDPOINT, DEFAULT_INTL_ENDPOINT, ENV_PREFER_AI_SITE, _env_bool, ) if self._config._endpoint_overridden: return self._config.endpoint or DEFAULT_ENDPOINT effective_token = token or self._config.token prefer_ai = _env_bool(ENV_PREFER_AI_SITE, False) primary = DEFAULT_INTL_ENDPOINT if prefer_ai else DEFAULT_ENDPOINT fallback = DEFAULT_ENDPOINT if prefer_ai else DEFAULT_INTL_ENDPOINT primary_probe = HubApi(endpoint=primary, token=effective_token) if primary_probe.repo_exists(repo_id, repo_type): return primary fallback_probe = HubApi(endpoint=fallback, token=effective_token) if fallback_probe.repo_exists(repo_id, repo_type): logger.warning( "Repo %s not found on %s, using %s instead.", repo_id, primary, fallback, ) return fallback raise NotExistError(f"Repo {repo_id} not found on either {primary} or {fallback}") # ================================================================== # Files # ================================================================== def upload_file_to_openapi(self, file: str | Path | BinaryIO) -> str: """Upload a file (max 5 MiB) via OpenAPI and return the file ID. This is a generic upload not tied to any repository. The returned ID can be used in subsequent API calls (e.g. ``skill_file`` when creating a skill). """ data = self.openapi.upload_file(file=file) return data["id"] def upload_file( self, repo_id: str, repo_type: RepoTypeLike, path_or_fileobj: str | Path | bytes | BinaryIO, path_in_repo: str, *, commit_message: str | None = None, commit_description: str | None = None, revision: str | None = None, buffer_size_mb: int = 16, disable_tqdm: bool = False, ) -> dict: """Upload a single file to a repository. LFS files upload/reuse their blob first, then commit an LFS pointer. Normal files are committed directly with inline base64 content. LFS mode is determined by file suffix and size threshold. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type (``"model"``, ``"dataset"``, ...). path_or_fileobj : str, Path, bytes or BinaryIO Local path, raw bytes, or a binary file-like object. path_in_repo : str Destination path inside the repository. commit_message : str, optional Commit message. Defaults to ``"Upload file"``. commit_description : str, optional Extended commit description. revision : str, optional Branch to commit on. Defaults to ``"master"``. buffer_size_mb : int, optional Buffer size in MiB for reading file data. Default 16. disable_tqdm : bool, optional Disable progress bar. Default False. Returns ------- dict Commit info from the server. Raises ------ AuthenticationError When the token is missing or invalid. NotExistError When the target repository does not exist. Examples -------- >>> api.upload_file( ... "alice/llama-7b", ... repo_type="model", ... path_or_fileobj="./pytorch_model.bin", ... path_in_repo="pytorch_model.bin", ... commit_message="Add fine-tuned weights", ... ) """ rt = self._normalize_repo_type(repo_type) return self.uploader.upload_file( repo_id=repo_id, repo_type=str(rt), path_or_fileobj=path_or_fileobj, path_in_repo=path_in_repo, commit_message=commit_message or "Upload file", commit_description=commit_description, revision=revision or "master", buffer_size_mb=buffer_size_mb, disable_tqdm=disable_tqdm, ) def upload_folder( self, repo_id: str, repo_type: RepoTypeLike, folder_path: str | Path, *, path_in_repo: str = "", commit_message: str | None = None, commit_description: str | None = None, revision: str | None = None, allow_patterns: list[str] | None = None, ignore_patterns: list[str] | None = None, max_workers: int | None = None, use_cache: bool = True, disable_tqdm: bool = False, sync_remote_repo: bool = False, ) -> dict | list[dict] | None: """Upload an entire folder to a repository with resumable support. Files are walked recursively from ``folder_path`` and uploaded in parallel with adaptive batching, per-file retry, and ReAct progressive retry fallback. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type. folder_path : str or Path Local directory whose contents will be uploaded. path_in_repo : str, optional Destination prefix inside the repository. Defaults to the repo root. commit_message : str, optional Commit message. Defaults to ``"Upload folder"``. commit_description : str, optional Extended commit description. revision : str, optional Branch to commit on. Defaults to ``"master"``. allow_patterns : list of str, optional If given, only files matching at least one pattern are uploaded. ignore_patterns : list of str, optional Files matching any pattern are skipped. max_workers : int, optional Concurrency for parallel uploads. Defaults to adaptive. use_cache : bool, optional Use ``.ms_upload_cache`` for resumable uploads. Default True. disable_tqdm : bool, optional Disable progress bars. Default False. sync_remote_repo : bool, optional If True, delete remote files that are not present locally after a successful upload (sync semantics). Default False. Returns ------- None If all files were already committed (nothing to do). dict If only one batch was committed. list of dict If multiple batches were committed. Examples -------- >>> api.upload_folder( ... "alice/llama-7b", ... repo_type="model", ... folder_path="./checkpoint-1000", ... ignore_patterns=["*.optim", "events.out.*"], ... max_workers=8, ... ) """ rt = self._normalize_repo_type(repo_type) return self.uploader.upload_folder( repo_id=repo_id, repo_type=str(rt), folder_path=folder_path, path_in_repo=path_in_repo, commit_message=commit_message or "Upload folder", commit_description=commit_description, revision=revision or "master", allow_patterns=allow_patterns, ignore_patterns=ignore_patterns, max_workers=max_workers, use_cache=use_cache, disable_tqdm=disable_tqdm, sync_remote_repo=sync_remote_repo, ) def download_file( self, repo_id: str, repo_type: RepoTypeLike, file_path: str, *, revision: str | None = None, cache_dir: str | Path | None = None, local_dir: str | Path | None = None, force: bool = False, expected_sha256: str | None = None, local_files_only: bool = False, user_agent: dict | str | None = None, ) -> Path: """Download a single file from a repository. The file is fetched into the local cache and a path pointing at the cached blob is returned. Subsequent calls reuse the cached copy unless ``force=True``. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type. file_path : str Path of the file inside the repository. revision : str, optional Branch, tag or commit SHA. Defaults to ``"master"``. cache_dir : str or Path, optional Override the default cache directory. local_dir : str or Path, optional When set, download directly into this directory instead of cache. force : bool, optional Re-download even if a cached copy exists. Default is ``False``. expected_sha256 : str, optional When provided, verify downloaded file hash and use it for cache hit validation. On mismatch, re-download up to 3 times. local_files_only : bool, optional When ``True``, return the cached path without network access. Raises :class:`CacheNotFound` if the file is not cached. user_agent : dict, str or None, optional Custom user-agent info appended to the default UA string. Returns ------- Path Absolute path to the downloaded file on disk. Raises ------ NotExistError When the file or repository does not exist. Examples -------- >>> path = api.download_file( ... "alice/llama-7b", ... repo_type="model", ... file_path="config.json", ... ) >>> path.read_text()[:30] '{\n "architectures": [\n "Ll' """ rt = self._normalize_repo_type(repo_type) if rt is RepoType.STUDIO: raise NotSupportedError( "File download is not supported for studio repositories. " "Studios are application containers without a file listing API. " f"To access studio source code, use: git clone https://modelscope.cn/studios/{repo_id}.git" ) return self.downloader.download_file( repo_id=repo_id, repo_type=str(rt), file_path=file_path, revision=revision or "master", cache_dir=Path(cache_dir) if cache_dir else None, local_dir=Path(local_dir) if local_dir else None, force=force, expected_sha256=expected_sha256, local_files_only=local_files_only, user_agent=user_agent, ) def download_repo( self, repo_id: str, repo_type: RepoTypeLike, *, revision: str | None = None, cache_dir: str | Path | None = None, local_dir: str | Path | None = None, allow_patterns: list[str] | None = None, ignore_patterns: list[str] | None = None, max_workers: int = 4, local_files_only: bool = False, user_agent: dict | str | None = None, progress_callbacks: list[type[ProgressCallback]] | None = None, ) -> Path: """Download an entire repository snapshot. All files at the given ``revision`` are fetched into the local cache in parallel. The returned path is the snapshot root directory. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type. revision : str, optional Branch, tag or commit SHA. Defaults to ``"master"``. cache_dir : str or Path, optional Override the default cache directory. local_dir : str or Path, optional When set, download directly into this directory instead of cache. allow_patterns : list of str, optional If given, only matching files are downloaded. ignore_patterns : list of str, optional Matching files are skipped. max_workers : int, optional Concurrency for parallel downloads. Default is 4. local_files_only : bool, optional When ``True``, return the cached snapshot path without network. user_agent : dict, str or None, optional Custom user-agent info for download headers. progress_callbacks : list of ProgressCallback subclasses, optional Callback *classes* (not instances); each is instantiated per file to report byte-level download progress. Returns ------- Path Absolute path to the snapshot/local directory. Examples -------- Download only the tokenizer assets of a model: >>> root = api.download_repo( ... "alice/llama-7b", ... repo_type="model", ... allow_patterns=["tokenizer*", "*.json"], ... max_workers=8, ... ) >>> sorted(p.name for p in root.iterdir()) ['config.json', 'tokenizer.json', 'tokenizer_config.json'] """ rt = self._normalize_repo_type(repo_type) if rt is RepoType.STUDIO: raise NotSupportedError( "File download is not supported for studio repositories. " "Studios are application containers without a file listing API. " f"To access studio source code, use: git clone https://modelscope.cn/studios/{repo_id}.git" ) return self.downloader.download_repo( repo_id=repo_id, repo_type=str(rt), revision=revision or "master", cache_dir=Path(cache_dir) if cache_dir else None, local_dir=Path(local_dir) if local_dir else None, allow_patterns=allow_patterns, ignore_patterns=ignore_patterns, max_workers=max_workers, local_files_only=local_files_only, user_agent=user_agent, progress_callbacks=progress_callbacks, ) def list_repo_files( self, repo_id: str, repo_type: RepoTypeLike, *, revision: str | None = None, recursive: bool = True, ) -> list[FileInfo]: """List files inside a repository (legacy — no OpenAPI equivalent). Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type. revision : str, optional Branch, tag or commit SHA. Defaults to ``"master"``. recursive : bool, optional Walk subdirectories recursively. Default is ``True``. Returns ------- list of FileInfo File metadata entries (path, size, blob id, last modified, ...). Examples -------- >>> files = api.list_repo_files("alice/llama-7b", "model") >>> [f.path for f in files][:3] ['README.md', 'config.json', 'pytorch_model.bin'] """ rt = self._normalize_repo_type(repo_type) raw = self.legacy.list_repo_files( repo_id=repo_id, repo_type=str(rt), revision=revision or "master", recursive=recursive, ) files: list[FileInfo] = [] for item in raw: normalised = { "path": item.get("Path") or item.get("path") or item.get("Name") or "", "size": int(item.get("Size") or item.get("size") or 0), "blob_id": item.get("BlobId") or item.get("blob_id") or item.get("Sha256"), "sha256": item.get("Sha256") or item.get("sha256"), "type": item.get("Type") or item.get("type") or "blob", "last_modified": item.get("CommittedDate") or item.get("last_modified"), "lfs": item.get("Lfs") or item.get("lfs"), } files.append(FileInfo.from_dict(normalised)) return files def delete_files( self, repo_id: str, repo_type: RepoTypeLike, file_paths: Iterable[str], *, commit_message: str | None = None, revision: str | None = None, ) -> dict: """Delete one or more files from a repository. .. note:: File deletion is restricted by the server to cookie-based session auth (interactive login). API tokens (``ms-...``) may receive a 401 "token no longer supports deletion operations" error. Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Repository type. file_paths : iterable of str Paths of files to remove. Empty entries are ignored. commit_message : str, optional Unused (kept for API compatibility). revision : str, optional Branch to delete from. Defaults to ``"master"``. Returns ------- dict Summary with ``deleted_files`` and ``failed_files`` lists. Raises ------ InvalidParameter When ``file_paths`` resolves to an empty list. Examples -------- >>> api.delete_files( ... "alice/llama-7b", ... "model", ... ["old_weights.bin", "deprecated/config.json"], ... ) """ rt = self._normalize_repo_type(repo_type) paths = [p for p in file_paths if p] if not paths: raise InvalidParameter("file_paths must contain at least one non-empty path.") deleted, failed = [], [] for p in paths: try: self.legacy.delete_file( repo_id=repo_id, repo_type=str(rt), file_path=p, revision=revision or "master", ) deleted.append(p) except (AuthenticationError, NetworkError): failed.append(p) raise except Exception: failed.append(p) return {"deleted_files": deleted, "failed_files": failed, "total_files": len(paths)} # ================================================================== # Versioning # ================================================================== def list_repo_revisions(self, repo_id: str, repo_type: RepoTypeLike) -> list[dict]: """Return branches and tags of a repository (legacy). Examples -------- >>> revs = api.list_repo_revisions("alice/llama-7b", "model") >>> [r["name"] for r in revs] ['master', 'v1.0', 'experimental'] """ rt = self._normalize_repo_type(repo_type) return self.legacy.list_revisions(repo_id=repo_id, repo_type=str(rt)) def create_repo_tag( self, repo_id: str, repo_type: RepoTypeLike, tag: str, *, revision: str | None = None, ) -> dict: """Create a tag pointing at ``revision`` (defaults to ``master``). Examples -------- >>> api.create_repo_tag("alice/llama-7b", "model", "v1.0") """ rt = self._normalize_repo_type(repo_type) return self.legacy.create_tag( repo_id=repo_id, repo_type=str(rt), tag=tag, revision=revision or "master", ) # ================================================================== # Lifecycle (Studio / MCP) # ================================================================== def deploy_repo( self, repo_id: str, repo_type: RepoTypeLike = RepoType.STUDIO, *, payload: Mapping[str, Any] | None = None, ) -> dict: """Deploy a Studio space or an MCP server. Parameters ---------- repo_id : str For Studios, an ``owner/name`` pair. For MCP servers, the server identifier accepted by the OpenAPI surface. repo_type : str or RepoType, optional ``"studio"`` (default) or ``"mcp"``. payload : Mapping, optional Deployment configuration forwarded verbatim to the backend (hardware tier, env vars, ...). Returns ------- dict Deployment response payload (deployment id, status URL, ...). Raises ------ NotSupportedError When ``repo_type`` is neither ``"studio"`` nor ``"mcp"``. Examples -------- Deploy a Studio space on a GPU instance: >>> api.deploy_repo( ... "alice/chat-demo", ... repo_type="studio", ... payload={"instance_type": "GPU-A10", "min_replicas": 1}, ... ) """ rt = self._normalize_repo_type(repo_type) if rt is RepoType.STUDIO: owner, name = self._parse_repo_id(repo_id) return self.openapi.deploy_studio(owner, name, payload) if rt is RepoType.MCP: return self.openapi.deploy_mcp_server(repo_id, payload) raise NotSupportedError(f"deploy_repo is not supported for repo_type={rt.value!r}.") def stop_repo( self, repo_id: str, repo_type: RepoTypeLike = RepoType.STUDIO, ) -> dict: """Stop a running Studio or undeploy an MCP server. Examples -------- >>> api.stop_repo("alice/chat-demo", repo_type="studio") """ rt = self._normalize_repo_type(repo_type) if rt is RepoType.STUDIO: owner, name = self._parse_repo_id(repo_id) return self.openapi.stop_studio(owner, name) if rt is RepoType.MCP: return self.openapi.undeploy_mcp_server(repo_id) raise NotSupportedError(f"stop_repo is not supported for repo_type={rt.value!r}.") def get_repo_logs( self, repo_id: str, repo_type: RepoTypeLike = RepoType.STUDIO, *, log_type: str = "run", page_num: int = 1, page_size: int = 20, keyword: str | None = None, start_timestamp: int | None = None, end_timestamp: int | None = None, ) -> dict: """Fetch paginated runtime/build logs for a Studio space. Parameters ---------- repo_id : str Studio identifier in ``owner/name`` form. repo_type : str or RepoType, optional Must be ``"studio"``. Defaults to :class:`RepoType.STUDIO`. log_type : str, optional Either ``"run"`` (default) or ``"build"``. page_num : int, optional 1-based page index. Default is 1. page_size : int, optional Lines per page. Default is 20. keyword : str, optional Filter logs containing this substring. start_timestamp, end_timestamp : int, optional Unix timestamps (seconds) bounding the log window. Returns ------- dict Paginated log payload. Raises ------ NotSupportedError When ``repo_type`` is not ``"studio"``. Examples -------- >>> logs = api.get_repo_logs( ... "alice/chat-demo", ... log_type="run", ... keyword="ERROR", ... page_size=50, ... ) """ rt = self._normalize_repo_type(repo_type) if rt is not RepoType.STUDIO: raise NotSupportedError(f"get_repo_logs is currently only supported for studio (got {rt.value!r}).") owner, name = self._parse_repo_id(repo_id) return self.openapi.get_studio_logs( owner, name, log_type, page_num=page_num, page_size=page_size, keyword=keyword, start_timestamp=start_timestamp, end_timestamp=end_timestamp, ) def update_repo_settings( self, repo_id: str, repo_type: RepoTypeLike, **settings: Any, ) -> dict: """Update repo settings (Studio or Skill). Parameters ---------- repo_id : str Canonical ``owner/name`` identifier. repo_type : str or RepoType Either ``"studio"`` or ``"skill"``. **settings : Any Setting key/value pairs forwarded to the backend. Returns ------- dict Updated settings payload. Raises ------ NotSupportedError When ``repo_type`` is neither studio nor skill. Examples -------- >>> api.update_repo_settings( ... "alice/chat-demo", ... repo_type="studio", ... visibility="public", ... hardware="GPU-A10", ... ) """ rt = self._normalize_repo_type(repo_type) owner, name = self._parse_repo_id(repo_id) for old_key, new_key in _STUDIO_FIELD_RENAMES.items(): if old_key in settings: settings[new_key] = settings.pop(old_key) if rt is RepoType.STUDIO: return self.openapi.update_studio_settings(owner, name, settings) if rt is RepoType.SKILL: return self.openapi.update_skill_settings(owner, name, settings) raise NotSupportedError(f"update_repo_settings is not supported for repo_type={rt.value!r}.") # ================================================================== # Secrets # ================================================================== def list_secrets(self, repo_id: str, repo_type: RepoTypeLike = RepoType.STUDIO) -> list[dict]: """List secrets attached to a Studio. Examples -------- >>> api.list_secrets("alice/chat-demo") [{'key': 'OPENAI_API_KEY', 'updated_at': 1712345678}, ...] """ rt = self._normalize_repo_type(repo_type) if rt is not RepoType.STUDIO: raise NotSupportedError(f"Secret management is only supported for studio (got {rt.value!r}).") owner, name = self._parse_repo_id(repo_id) data = self.openapi.list_studio_secrets(owner, name) if isinstance(data, list): return data if isinstance(data, dict): for key in ("items", "secrets", "list"): if isinstance(data.get(key), list): return data[key] return [] def add_secret( self, repo_id: str, key: str, value: str, repo_type: RepoTypeLike = RepoType.STUDIO, ) -> dict: """Add a new secret to a Studio. Examples -------- >>> api.add_secret("alice/chat-demo", "OPENAI_API_KEY", "sk-...") """ rt = self._normalize_repo_type(repo_type) if rt is not RepoType.STUDIO: raise NotSupportedError("Only studio secrets are supported.") owner, name = self._parse_repo_id(repo_id) return self.openapi.add_studio_secret(owner, name, key, value) def update_secret( self, repo_id: str, key: str, value: str, repo_type: RepoTypeLike = RepoType.STUDIO, ) -> dict: """Update an existing secret value. Examples -------- >>> api.update_secret("alice/chat-demo", "OPENAI_API_KEY", "sk-new-...") """ rt = self._normalize_repo_type(repo_type) if rt is not RepoType.STUDIO: raise NotSupportedError("Only studio secrets are supported.") owner, name = self._parse_repo_id(repo_id) return self.openapi.update_studio_secret(owner, name, key, value) def delete_secret( self, repo_id: str, key: str, repo_type: RepoTypeLike = RepoType.STUDIO, ) -> dict: """Delete a secret from a Studio. Examples -------- >>> api.delete_secret("alice/chat-demo", "OPENAI_API_KEY") """ rt = self._normalize_repo_type(repo_type) if rt is not RepoType.STUDIO: raise NotSupportedError("Only studio secrets are supported.") owner, name = self._parse_repo_id(repo_id) return self.openapi.delete_studio_secret(owner, name, key) # ================================================================== # MCP convenience wrappers # ================================================================== def list_mcp_servers( self, *, search: str | None = None, page_number: int = 1, page_size: int = 10, **extra: Any, ) -> PagedResult[dict]: """List MCP servers via the OpenAPI surface. Parameters ---------- search : str, optional Free-text search query. page_number : int, optional 1-based page index. Default is 1. page_size : int, optional Items per page. Default is 10. **extra : Any Additional filter fields. ``None`` values are dropped. Returns ------- PagedResult[dict] Paginated MCP server listing. Examples -------- >>> page = api.list_mcp_servers(search="weather", page_size=5) >>> page.total_count 12 """ payload = self.openapi.list_mcp_servers( search=search, page_number=page_number, page_size=page_size, extra={k: v for k, v in extra.items() if v is not None} or None, ) items, total, page, size = self._extract_paged(payload) return PagedResult(items=list(items), total_count=total, page_number=page, page_size=size) def get_mcp_server( self, server_id: str, *, get_operational_url: bool | None = None, ) -> dict: """Fetch a single MCP server's metadata. Parameters ---------- server_id : str MCP server identifier. get_operational_url : bool, optional When ``True``, the response includes the live runtime URL. Returns ------- dict MCP server metadata. Examples -------- >>> info = api.get_mcp_server("alice/weather-mcp", get_operational_url=True) >>> info["operational_url"] 'https://...' """ return self.openapi.get_mcp_server(server_id, get_operational_url=get_operational_url) def deploy_mcp_server(self, server_id: str, *, payload: Mapping[str, Any] | None = None) -> dict: """Deploy or redeploy an MCP server. Examples -------- >>> api.deploy_mcp_server("alice/weather-mcp", payload={"region": "cn-hangzhou"}) """ return self.openapi.deploy_mcp_server(server_id, payload) def undeploy_mcp_server(self, server_id: str) -> dict: """Undeploy a running MCP server. Examples -------- >>> api.undeploy_mcp_server("alice/weather-mcp") """ return self.openapi.undeploy_mcp_server(server_id) # ================================================================== # Cache # ================================================================== def verify_cache( self, repo_id: str, repo_type: RepoTypeLike = "model", *, revision: str | None = None, cache_dir: str | Path | None = None, local_dir: str | Path | None = None, ) -> CacheVerification: """Verify local repository files against SHA-256 checksums from the Hub.""" if cache_dir is not None and local_dir is not None: raise InvalidParameter("cache_dir and local_dir are mutually exclusive") rt = self._normalize_repo_type(repo_type) _, resolved_revision = _resolve_verification_root( repo_id, str(rt), revision=revision, cache_dir=cache_dir, local_dir=local_dir, ) files = self.list_repo_files(repo_id, rt, revision=resolved_revision, recursive=True) expected = {file.path: file.sha256 for file in files if not file.is_dir and file.path} return _verify_cache( repo_id, str(rt), expected, revision=resolved_revision, cache_dir=Path(cache_dir) if cache_dir else None, local_dir=Path(local_dir) if local_dir else None, ) def scan_cache(self, cache_dir: str | Path | None = None) -> CacheInfo: """Inspect the local cache directory. Parameters ---------- cache_dir : str or Path, optional Override the default cache root. Returns ------- CacheInfo Aggregated cache stats (total bytes, repo entries, ...). Examples -------- >>> info = api.scan_cache() >>> info.size_on_disk 12345678 >>> [r.repo_id for r in info.repos][:3] ['alice/llama-7b', 'bob/imagenet', 'carol/whisper-base'] """ return _scan_cache(Path(cache_dir) if cache_dir else None) def clear_cache( self, *, cache_dir: str | Path | None = None, repo_type: RepoTypeLike | None = None, repo_id: str | None = None, ) -> int: """Remove cached data and return the number of bytes freed. Parameters ---------- cache_dir : str or Path, optional Override the default cache root. repo_type : str or RepoType, optional Restrict deletion to a single repository type. repo_id : str, optional Restrict deletion to a single repository id. Returns ------- int Number of bytes reclaimed from disk. Examples -------- Wipe the cache for a single model: >>> api.clear_cache(repo_type="model", repo_id="alice/llama-7b") 4823412 Wipe everything: >>> api.clear_cache() """ rt_value: str | None = None if repo_type is not None: rt_value = str(self._normalize_repo_type(repo_type)) return _clear_cache( cache_dir=Path(cache_dir) if cache_dir else None, repo_type=rt_value, repo_id=repo_id, )