"""Internal client for ``/api/v1/`` endpoints not covered by the OpenAPI spec. This module is **private** — external callers should interact through the public :class:`~modelscope_hub.api.HubApi` facade. Auth strategy ------------- The legacy API authenticates via a session cookie (``m_session_id``) rather than the ``Authorization: Bearer`` header used by the OpenAPI surface. On first use the client sets ``m_session_id = `` on the ``requests.Session`` cookie jar — no round-trip to ``/api/v1/login`` is needed for API calls. """ from __future__ import annotations import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from typing import IO, Any, BinaryIO from urllib.parse import quote_plus, urlparse import requests from requests.adapters import HTTPAdapter, Retry from .constants import ( API_CONNECT_TIMEOUT, API_MAX_RETRIES, API_TIMEOUT, LEGACY_API_PREFIX, REPO_FILES_TRUNCATION_LIMIT, REPO_TREE_MAX_REQUESTS, REPO_TREE_WALK_WORKERS, UPLOAD_BLOB_CONNECT_TIMEOUT_SECONDS, UPLOAD_BLOB_READ_TIMEOUT_SECONDS, UPLOAD_HTTP_RETRY_ALLOWED_METHODS, RepoType, ) from .errors import InvalidParameter, NetworkError, RequestTimeoutError, ServerError, raise_for_status from .utils.logger import get_logger logger = get_logger("legacy_api") # --------------------------------------------------------------------------- # Repo-type routing table: maps RepoType → URL path segment (plural) # --------------------------------------------------------------------------- _REPO_TYPE_SEGMENT: dict[str, str] = { RepoType.MODEL: "models", RepoType.DATASET: "datasets", RepoType.STUDIO: "studios", RepoType.SKILL: "skills", RepoType.MCP: "mcps", } # For /api/v1/repos/ style endpoints, all use "{type}s" pattern _REPOS_SEGMENT: dict[str, str] = { RepoType.MODEL: "models", RepoType.DATASET: "datasets", RepoType.STUDIO: "studios", RepoType.SKILL: "skills", RepoType.MCP: "mcps", } def _resolve_segment(repo_type: str) -> str: """Return the URL path segment for the given repo_type.""" return _REPO_TYPE_SEGMENT.get(repo_type, f"{repo_type}s") def _is_dataset(repo_type: str) -> bool: """Whether the repo_type addresses the dataset endpoints.""" return repo_type in (RepoType.DATASET, "dataset", "datasets") def _entry_path(entry: dict) -> str: """Return the repo-relative path of a file-tree entry.""" return entry.get("Path") or entry.get("path") or entry.get("Name") or "" def _is_dir_entry(entry: dict) -> bool: """Whether a file-tree entry is a directory (git tree object).""" return (entry.get("Type") or entry.get("type") or "blob") == "tree" class LegacyClient: """Internal client for /api/v1/ endpoints not covered by OpenAPI. All network IO goes through :meth:`_request`, which handles auth headers, retries, and structured error raising. """ def __init__( self, token: str | None, endpoint: str, timeout: int = API_TIMEOUT, max_retries: int = API_MAX_RETRIES, user_agent: str | None = None, ) -> None: self._token = token self._endpoint = endpoint.rstrip("/") self._timeout: int | tuple[int, int] = (API_CONNECT_TIMEOUT, timeout) self._session_authenticated = False self._session = requests.Session() if user_agent: self._session.headers["User-Agent"] = user_agent retry = Retry( total=max_retries, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=UPLOAD_HTTP_RETRY_ALLOWED_METHODS, ) adapter = HTTPAdapter(max_retries=retry) self._session.mount("https://", adapter) self._session.mount("http://", adapter) # ------------------------------------------------------------------ # Properties # ------------------------------------------------------------------ @property def endpoint(self) -> str: return self._endpoint @property def token(self) -> str | None: return self._token @token.setter def token(self, value: str | None) -> None: self._token = value self._session_authenticated = False self._session.cookies.clear() # ------------------------------------------------------------------ # Transport # ------------------------------------------------------------------ def _build_url(self, path: str) -> str: """Construct full URL from a path relative to the legacy API prefix.""" return f"{self._endpoint}{LEGACY_API_PREFIX}/{path.lstrip('/')}" def _headers(self, extra: dict[str, str] | None = None) -> dict[str, str]: headers: dict[str, str] = { "Content-Type": "application/json", "X-Request-ID": uuid.uuid4().hex, } if self._token: headers["Authorization"] = f"Bearer {self._token}" if extra: headers.update(extra) return headers def _ensure_session_auth(self) -> None: """Set the ``m_session_id`` cookie on the session for legacy API auth. Both cookie and Bearer header are sent: older endpoints rely on the ``m_session_id`` cookie, while newer endpoints accept the ``Authorization: Bearer`` header. Sending both is harmless and maximises compatibility. """ if self._session_authenticated or not self._token: return domain = urlparse(self._endpoint).hostname or "" self._session.cookies.set("m_session_id", self._token, domain=domain, path="/") self._session_authenticated = True logger.debug("Legacy session cookie set for domain=%s", domain) def _request( self, method: str, path: str, *, params: dict[str, Any] | None = None, json_body: Any | None = None, data: Any | None = None, files: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: int | None = None, stream: bool = False, ) -> requests.Response: """Unified request dispatcher with auth, retry, and error handling.""" self._ensure_session_auth() url = self._build_url(path) merged_headers = self._headers(headers) # Remove Content-Type for non-json payloads so requests can set # the correct boundary for multipart or the right encoding for data. if (data is not None or files is not None) and json_body is None: merged_headers.pop("Content-Type", None) logger.debug("%s %s", method, url) try: resp = self._session.request( method=method, url=url, params=params, json=json_body, data=data, files=files, headers=merged_headers, timeout=timeout or self._timeout, stream=stream, ) except requests.exceptions.RetryError as exc: raise ServerError(f"Max retries exceeded: {exc}") from exc except requests.ConnectionError as exc: raise NetworkError(f"Connection failed: {exc}") from exc except requests.Timeout as exc: raise RequestTimeoutError(f"Request timed out: {exc}") from exc logger.debug("%s %s -> %s", method, url, resp.status_code) if resp.status_code >= 400: logger.debug( "Request failed: %s %s params=%s status=%s body=%s", method, url, params, resp.status_code, resp.text[:500] if resp.text else "", ) raise_for_status(resp) return resp def _json_data(self, resp: requests.Response) -> Any: """Extract the 'Data' field from a standard legacy API JSON response.""" body = resp.json() return body.get("Data", body) # ------------------------------------------------------------------ # Auth # ------------------------------------------------------------------ def login(self, access_token: str) -> tuple[dict, requests.cookies.RequestsCookieJar]: """Authenticate via access token and return (user_data, cookies). POST /api/v1/login Performs a full login round-trip that returns a git token and server-issued session cookies. For routine API access, the ``m_session_id`` cookie set by :meth:`_ensure_session_auth` is sufficient — this method is only needed when the caller wants the git token or the full cookie set from the server. """ self._token = access_token self._session_authenticated = False self._session.cookies.clear() self._ensure_session_auth() resp = self._request("POST", "login", json_body={"AccessToken": access_token}) return self._json_data(resp), resp.cookies # ------------------------------------------------------------------ # Repo CRUD (model / dataset) # ------------------------------------------------------------------ def create_repo(self, repo_type: str, body: dict[str, Any]) -> dict: """POST /api/v1/{type}s — create a new repository. ``body`` is the fully-constructed request payload (PascalCase keys). The dataset endpoint requires multipart form data (matching the upstream server implementation), while model uses JSON. """ segment = _resolve_segment(repo_type) if repo_type in (RepoType.DATASET, "dataset"): files = {k: (None, str(v)) for k, v in body.items() if v is not None} resp = self._request("POST", segment, files=files) else: resp = self._request("POST", segment, json_body=body) return self._json_data(resp) def create_aigc_model(self, body: dict[str, Any]) -> dict: """POST /api/v1/models/aigc — create an AIGC model repository. AIGC repositories use a dedicated legacy endpoint and payload shape; routing them through :meth:`create_repo` would incorrectly target the ordinary ``/models`` endpoint. """ resp = self._request("POST", "models/aigc", json_body=body) return self._json_data(resp) def get_repo_info(self, repo_id: str, repo_type: str) -> dict: """GET /api/v1/{type}s/{repo_id} — fetch repository metadata. Uses cookie-based session auth which supports private repositories (unlike the OpenAPI Bearer-only endpoint). """ segment = _resolve_segment(repo_type) resp = self._request("GET", f"{segment}/{repo_id}") return self._json_data(resp) def list_datasets( self, *, owner: str | None = None, page_number: int = 1, page_size: int = 50, ) -> dict: """GET /api/v1/datasets — list datasets with full private visibility. The legacy endpoint returns ALL datasets for the authenticated user (including private ones that the OpenAPI search index may miss). """ params: dict[str, Any] = { "PageNumber": page_number, "PageSize": page_size, } if owner: params["owner"] = owner try: resp = self._request("GET", "datasets", params=params) except InvalidParameter as exc: if "fromIndex" in str(exc) and "toIndex" in str(exc): return { "items": [], "total_count": 0, "page_number": page_number, "page_size": page_size, } raise body = resp.json() return { "items": body.get("Data") or [], "total_count": body.get("TotalCount") or 0, "page_number": body.get("PageNumber") or page_number, "page_size": body.get("PageSize") or page_size, } def delete_repo(self, repo_id: str, repo_type: str) -> None: """Delete a repository. DELETE /api/v1/models/{owner}/{name} DELETE /api/v1/datasets/{dataset_id} """ segment = _resolve_segment(repo_type) self._request("DELETE", f"{segment}/{repo_id}") # ------------------------------------------------------------------ # File Tree # ------------------------------------------------------------------ def list_repo_files( self, repo_id: str, repo_type: str, revision: str = "master", recursive: bool = True, root: str | None = None, ) -> list[dict]: """List files in a repository. Models/studios/etc: GET /api/v1/{type}s/{repo_id}/repo/files Datasets: GET /api/v1/datasets/{repo_id}/repo/tree ``repo/files`` supports no pagination and silently truncates at :data:`REPO_FILES_TRUNCATION_LIMIT` entries, so a recursive listing that comes back exactly at the limit is re-enumerated directory by directory (see :meth:`_walk_repo_files`). Datasets take the ``repo/tree`` endpoint, which does paginate, so they are paged through instead. """ if _is_dataset(repo_type): if recursive: return self.list_dataset_files_paginated( repo_id=repo_id, revision=revision, root_path=root or "/", ) return self._list_files_page(repo_id, repo_type, revision, recursive=False, root=root) entries = self._list_files_page(repo_id, repo_type, revision, recursive=recursive, root=root) if len(entries) < REPO_FILES_TRUNCATION_LIMIT: return entries if not recursive: logger.warning( "Directory %r of %s holds at least %d entries and the server returns no more " "than that for a single listing; the result is incomplete.", root or "/", repo_id, REPO_FILES_TRUNCATION_LIMIT, ) return entries return self._walk_repo_files(repo_id, repo_type, revision, root=root, prefetched=entries) def _list_files_page( self, repo_id: str, repo_type: str, revision: str, *, recursive: bool, root: str | None = None, ) -> list[dict]: """Perform a single file-tree request and unwrap the entry list.""" segment = _resolve_segment(repo_type) params: dict[str, Any] = { "Revision": revision, "Recursive": str(recursive), } if root: params["Root"] = root suffix = "repo/tree" if _is_dataset(repo_type) else "repo/files" resp = self._request("GET", f"{segment}/{repo_id}/{suffix}", params=params) data = self._json_data(resp) if isinstance(data, list): return data # Sometimes wrapped: {"Data": {"Files": [...]}} if isinstance(data, dict): return data.get("Files") or data.get("files") or [] return [] def _walk_repo_files( self, repo_id: str, repo_type: str, revision: str, *, root: str | None = None, prefetched: list[dict] | None = None, ) -> list[dict]: """Enumerate a file tree the server truncated, one directory at a time. Since ``repo/files`` caps every response, the only way to see past the cap is to scope requests with ``Root``. Each subtree is first requested whole; only the subtrees that come back at the cap are listed shallowly and walked per child directory. The request count therefore scales with the number of oversized directories, not with the directory total. The walk proceeds one tree level at a time and issues the listings of a level concurrently, since a deep repo needs hundreds of round trips. ``prefetched`` is the already-truncated listing of ``root``, reused so the caller's request is not repeated. Entries are de-duplicated by path. Directories with more direct children than the cap cannot be enumerated at all — those are reported through a warning and the returned list is then knowingly incomplete. """ limit = REPO_FILES_TRUNCATION_LIMIT collected: dict[str, dict] = {} oversized: list[str] = [] requests_made = 0 progress_mark = 200 budget_spent = False logger.info( "Repo %s: file listing came back at the server cap of %d entries; walking the " "tree per directory to recover the full list ...", repo_id, limit, ) def absorb(entries: list[dict]) -> None: for entry in entries: path = _entry_path(entry) if path: collected.setdefault(path, entry) def fetch_level(roots: list[str | None], *, recursive: bool) -> dict[str | None, list[dict]]: """List several directories at once, clamped to the request budget.""" nonlocal requests_made, budget_spent, progress_mark if not roots: return {} remaining = REPO_TREE_MAX_REQUESTS - requests_made if remaining <= 0: budget_spent = True return {} if len(roots) > remaining: roots = roots[:remaining] budget_spent = True requests_made += len(roots) def one(subroot: str | None) -> list[dict]: return self._list_files_page( repo_id, repo_type, revision, recursive=recursive, root=subroot, ) if len(roots) == 1: listings = {roots[0]: one(roots[0])} else: workers = min(REPO_TREE_WALK_WORKERS, len(roots)) with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(one, subroot): subroot for subroot in roots} listings = {futures[future]: future.result() for future in as_completed(futures)} if requests_made >= progress_mark: logger.info( "Repo %s: %d listings done, %d entries collected so far ...", repo_id, requests_made, len(collected), ) progress_mark = requests_made + 200 return listings frontier: list[str | None] = [root] known: dict[str | None, list[dict]] = {} if prefetched is None else {root: prefetched} while frontier: known.update(fetch_level([r for r in frontier if r not in known], recursive=True)) truncated: list[str | None] = [] for subroot in frontier: subtree = known.pop(subroot, None) if subtree is None: continue # budget ran out before this directory was reached # A truncated subtree is still a valid prefix, so keep its entries # and split the directory up to reach the rest. absorb(subtree) if len(subtree) >= limit: truncated.append(subroot) if not truncated or budget_spent: break shallow_listings = fetch_level(truncated, recursive=False) next_frontier: list[str | None] = [] for subroot in truncated: shallow = shallow_listings.get(subroot) if shallow is None: continue absorb(shallow) child_dirs = [path for path in map(_entry_path, filter(_is_dir_entry, shallow)) if path] if len(shallow) >= limit or not child_dirs: # This level alone exceeds the cap, or it has no sub-directories # to split by: part of it stays invisible whatever we do. Still # descend into the children we did see — that recovers strictly # more than giving up here. oversized.append(subroot or "/") next_frontier.extend(child_dirs) if budget_spent: break frontier = list(dict.fromkeys(next_frontier)) if budget_spent: logger.warning( "Repo %s: stopped walking the file tree after %d listings " "(MODELSCOPE_REPO_TREE_MAX_REQUESTS=%d); the file list is incomplete.", repo_id, requests_made, REPO_TREE_MAX_REQUESTS, ) if oversized: shown = ", ".join(oversized[:5]) + (" ..." if len(oversized) > 5 else "") logger.warning( "Repo %s: %d director%s hold %d or more direct entries, which the server " "cannot enumerate in full; the file list is incomplete: %s", repo_id, len(oversized), "y" if len(oversized) == 1 else "ies", limit, shown, ) logger.info( "Repo %s: collected %d entries from %d directory listings.", repo_id, len(collected), requests_made, ) return list(collected.values()) def list_dataset_files_paginated( self, repo_id: str, revision: str = "master", page_size: int = 200, root_path: str = "/", ) -> list[dict]: """List files in a dataset repo using pagination. Datasets can have millions of files, so this method pages through ``GET /api/v1/datasets/{repo_id}/repo/tree`` with ``PageNumber``/``PageSize`` params. """ all_files: list[dict] = [] page_number = 1 while True: params: dict[str, Any] = { "Revision": revision, "Recursive": "True", "PageNumber": page_number, "PageSize": page_size, } if root_path and root_path != "/": params["Root"] = root_path resp = self._request( "GET", f"datasets/{repo_id}/repo/tree", params=params, ) data = self._json_data(resp) if isinstance(data, list): files = data elif isinstance(data, dict): files = data.get("Files") or data.get("files") or [] else: files = [] all_files.extend(files) if len(files) < page_size: break page_number += 1 return all_files # ------------------------------------------------------------------ # Revisions # ------------------------------------------------------------------ def list_revisions(self, repo_id: str, repo_type: str) -> list[dict]: """List branches and tags of a repository. GET /api/v1/{type}s/{repo_id}/revisions """ segment = _resolve_segment(repo_type) resp = self._request("GET", f"{segment}/{repo_id}/revisions") data = self._json_data(resp) if isinstance(data, dict): revision_map = data.get("RevisionMap") or {} tags = revision_map.get("Tags") or [] branches = revision_map.get("Branches") or [] return tags + branches return data if isinstance(data, list) else [] def list_revisions_detail( self, repo_id: str, repo_type: str, *, end_time: int | None = None, ) -> tuple[list[dict], list[dict]]: """List branches and tags as separate lists. GET /api/v1/{type}s/{repo_id}/revisions[?EndTime=...] Returns ``(branches, tags)`` where each element is a list of dicts with at least ``Revision`` and ``CreatedAt`` keys. """ segment = _resolve_segment(repo_type) params: dict[str, Any] = {} if end_time is not None: params["EndTime"] = end_time resp = self._request( "GET", f"{segment}/{repo_id}/revisions", params=params or None, ) data = self._json_data(resp) if isinstance(data, dict): revision_map = data.get("RevisionMap") or {} return ( revision_map.get("Branches") or [], revision_map.get("Tags") or [], ) return [], [] def create_aigc_model_tag(self, body: dict[str, Any]) -> dict: """POST /api/v1/models/aigc/repo/tag — create an AIGC model version.""" resp = self._request("POST", "models/aigc/repo/tag", json_body=body) return self._json_data(resp) def create_tag( self, repo_id: str, repo_type: str, tag: str, revision: str, ) -> dict: """Create a tag on a repository. POST /api/v1/{type}s/{repo_id}/repo/tag """ segment = _resolve_segment(repo_type) body = {"TagName": tag, "Ref": revision} resp = self._request("POST", f"{segment}/{repo_id}/repo/tag", json_body=body) return self._json_data(resp) # ------------------------------------------------------------------ # File deletion # ------------------------------------------------------------------ def delete_file( self, repo_id: str, repo_type: str, file_path: str, revision: str = "master", ) -> dict: """Delete a single file from the repository. DELETE /api/v1/{type}s/{owner}/{name}/file?FilePath=...&Revision=... (for models) DELETE /api/v1/datasets/{owner}/{name}/repo?FilePath=... (for datasets) """ segment = _resolve_segment(repo_type) if repo_type == RepoType.DATASET: resp = self._request( "DELETE", f"{segment}/{repo_id}/repo", params={"FilePath": file_path}, ) else: resp = self._request( "DELETE", f"{segment}/{repo_id}/file", params={"FilePath": file_path, "Revision": revision}, ) return self._json_data(resp) # ------------------------------------------------------------------ # Git Commits # ------------------------------------------------------------------ def create_commit( self, repo_id: str, repo_type: str, operations: list[dict], commit_message: str, revision: str = "master", ) -> dict: """Create a Git commit with file operations. POST /api/v1/repos/{type}s/{repo_id}/commit/{revision} Each operation should be a dict with keys: - action: "create" | "update" | "delete" - path: path in repo - type: "normal" | "lfs" - size: file size in bytes - sha256: hex digest (empty string for normal files) - content: base64 content (for normal files) - encoding: "base64" (for normal files) """ segment = _resolve_segment(repo_type) payload = { "commit_message": commit_message, "actions": operations, } resp = self._request( "POST", f"repos/{segment}/{repo_id}/commit/{revision}", json_body=payload, ) return self._json_data(resp) # ------------------------------------------------------------------ # Blob Upload (LFS) # ------------------------------------------------------------------ def validate_blobs( self, repo_id: str, repo_type: str, objects: list[dict[str, Any]], ) -> dict[str, str | None]: """Check which blobs need uploading via the LFS batch API. POST /api/v1/repos/{type}s/{repo_id}/info/lfs/objects/batch Returns mapping: sha256 → upload_url (if needs upload) or None (exists). """ segment = _resolve_segment(repo_type) payload = {"operation": "upload", "objects": objects} resp = self._request( "POST", f"repos/{segment}/{repo_id}/info/lfs/objects/batch", json_body=payload, ) data = self._json_data(resp) result: dict[str, str | None] = {} resp_objects = data.get("objects", []) if isinstance(data, dict) else [] for obj in resp_objects: oid = obj.get("oid", "") actions = obj.get("actions", {}) upload_action = actions.get("upload", {}) href = upload_action.get("href") result[oid] = href # None means already exists return result def upload_blob( self, upload_url: str, data: str | bytes | BinaryIO | IO[bytes] | Any, size: int, *, headers: dict[str, str] | None = None, timeout: int | tuple[int, int] | None = None, ) -> dict: """Upload a blob to the presigned URL returned by :meth:`validate_blobs`. PUT {upload_url} Sends both ``Authorization: Bearer`` and ``Cookie: m_session_id`` headers to authenticate against the LFS domain (which may differ from the main API domain). Returns the parsed JSON response body on success. """ upload_headers: dict[str, str] = { "Content-Length": str(size), "X-Request-ID": uuid.uuid4().hex, } if self._token: upload_headers["Authorization"] = f"Bearer {self._token}" upload_headers["Cookie"] = f"m_session_id={self._token}" if headers: upload_headers.update(headers) try: resp = self._session.put( upload_url, data=data, headers=upload_headers, timeout=timeout or (UPLOAD_BLOB_CONNECT_TIMEOUT_SECONDS, UPLOAD_BLOB_READ_TIMEOUT_SECONDS), ) except requests.ConnectionError as exc: raise NetworkError(f"Blob upload connection failed: {exc}") from exc except requests.Timeout as exc: raise RequestTimeoutError(f"Blob upload timed out: {exc}") from exc raise_for_status(resp) # Presigned URLs (cloud storage) may return empty bodies on success. try: body = resp.json() except (ValueError, RuntimeError): return {} if isinstance(body, dict) and body.get("Code") not in (200, "200", None): from .errors import APIError raise APIError( body.get("Message") or body.get("message") or f"Blob upload failed (Code={body.get('Code')})", status_code=resp.status_code, response_body=body, url=upload_url, method="PUT", ) return body # ------------------------------------------------------------------ # Raw Download URL # ------------------------------------------------------------------ def get_download_url( self, repo_id: str, repo_type: str, file_path: str, revision: str = "master", ) -> str: """Construct the file download URL (no request is made). URL pattern: {endpoint}/api/v1/{type}s/{repo_id}/repo?Revision={rev}&FilePath={path} """ segment = _resolve_segment(repo_type) return ( f"{self._endpoint}{LEGACY_API_PREFIX}/{segment}/{repo_id}/repo" f"?Revision={quote_plus(revision)}&FilePath={quote_plus(file_path)}" ) # ------------------------------------------------------------------ # Collections # ------------------------------------------------------------------ def get_collection( self, collection_id: str, *, element_type: str = "skill", page_number: int = 1, page_size: int = 50, ) -> dict: """Get collection details and its elements. GET /api/v1/collections?Fid=...&ElementType=... """ params = { "Fid": collection_id, "ElementType": element_type, "PageNumber": page_number, "PageSize": page_size, } resp = self._request("GET", "collections", params=params) return self._json_data(resp) # ------------------------------------------------------------------ # Archive Download (skill repos) # ------------------------------------------------------------------ def download_archive( self, repo_id: str, repo_type: str, revision: str = "master", headers: dict[str, str] | None = None, ) -> requests.Response: """Download the entire repo as a zip archive. GET /api/v1/{type}s/{repo_id}/archive/zip/{revision} Skills (and potentially other repo types) do not support per-file download via ``/repo?FilePath=...``. This method streams the archive endpoint instead. """ segment = _resolve_segment(repo_type) return self._request( "GET", f"{segment}/{repo_id}/archive/zip/{revision}", headers=headers, stream=True, ) # ------------------------------------------------------------------ # Raw Download URL # ------------------------------------------------------------------ def download_stream( self, repo_id: str, repo_type: str, file_path: str, revision: str = "master", headers: dict[str, str] | None = None, ) -> requests.Response: """Start a streaming download of a file. Returns a Response with stream=True for chunked reading. """ segment = _resolve_segment(repo_type) params = {"Revision": revision, "FilePath": file_path} return self._request( "GET", f"{segment}/{repo_id}/repo", params=params, headers=headers, stream=True, )