Module facetorch.downloader

Authenticated, manifest-aware model artifact downloaders.

Classes

class DownloaderGDrive (file_id: str,
path_local: str,
*,
sha256: Optional[str] = None,
size_bytes: Optional[int] = None,
expected_format: Optional[str] = None,
revision: str = 'local-gdrive-object-v1',
offline: Optional[bool] = None,
allow_legacy_models: bool = False,
verify_on_use: bool = True,
device: Any = 'cpu')

Expand source code

class DownloaderGDrive(_VerifiedDownloader):
    """Verified Google Drive downloader for explicitly described artifacts."""

    def __init__(
        self,
        file_id: str,
        path_local: str,
        *,
        sha256: Optional[str] = None,
        size_bytes: Optional[int] = None,
        expected_format: Optional[str] = None,
        revision: str = "local-gdrive-object-v1",
        offline: Optional[bool] = None,
        allow_legacy_models: bool = False,
        verify_on_use: bool = True,
        device: Any = "cpu",
    ) -> None:
        super().__init__(
            file_id,
            path_local,
            offline=offline,
            allow_legacy_models=allow_legacy_models,
            verify_on_use=verify_on_use,
        )
        self.sha256 = sha256
        self.size_bytes = size_bytes
        self.expected_format = expected_format
        self.revision = revision
        self.device = device

    def _descriptor(self) -> ArtifactDescriptor:
        filename = Path(self.path_local).name
        return _direct_descriptor(
            source="gdrive",
            repo_id=self.file_id,
            revision=self.revision,
            filename=filename,
            path_local=self.path_local,
            sha256=self.sha256,
            size_bytes=self.size_bytes,
            expected_format=self.expected_format,
            device=self.device,
        )

    def run(self, force_download: bool = False) -> str:
        target = Path(self.path_local).expanduser()
        _ensure_directory(target.parent)
        descriptor = self._descriptor()
        if descriptor.format == "torchscript" and not self.allow_legacy_models:
            raise ModelCompatibilityError(
                "Google Drive TorchScript models require allow_legacy_models=True."
            )
        with _DirectoryLock(target.parent / ".facetorch-download.lock"):
            if not force_download:
                existing = self._verified_existing(descriptor, target)
                if existing is not None:
                    return existing
            if self.offline:
                raise OfflineCacheError(
                    f"Offline mode requires a verified cached artifact at {target}."
                )
            with tempfile.TemporaryDirectory(
                prefix=".facetorch-download-", dir=target.parent
            ) as temporary_dir:
                temporary_path = Path(temporary_dir) / descriptor.filename
                url = (
                    "https://drive.google.com/uc?&id="
                    f"{self.file_id}&confirm=t"
                )
                downloaded = gdown.download(
                    url, output=os.fspath(temporary_path), quiet=False
                )
                candidate = Path(downloaded) if downloaded else temporary_path
                if not candidate.is_file():
                    raise ArtifactIntegrityError(
                        "Google Drive download did not produce an artifact for "
                        f"file_id {self.file_id!r}."
                    )
                _atomic_promote(candidate, target, descriptor)
        return self._activate(descriptor, target)

Verified Google Drive downloader for explicitly described artifacts.

Base class for downloaders.

All downloaders should subclass it. All subclass should overwrite:

• Methods:run, supporting to run the download functionality.

Args

-----=

file_id : str

ID of the hosted file (e.g. Google Drive File ID).

path_local : str

The file is downloaded to this local path.

Ancestors

• facetorch.downloader._VerifiedDownloader
• BaseDownloader

Inherited members

• BaseDownloader:
  • run

class DownloaderHuggingFace (file_id: str,
path_local: str,
repo_id: Optional[str] = None,
filename: Optional[str] = None,
export_filenames_by_torch_minor: Optional[Dict[str, str]] = None,
fallback_filenames: Optional[List[str]] = None,
enable_default_torch_export_routing: bool = False,
*,
manifest_id: Optional[str] = None,
revision: Optional[str] = None,
sha256: Optional[str] = None,
size_bytes: Optional[int] = None,
expected_format: Optional[str] = None,
offline: Optional[bool] = None,
allow_legacy_models: bool = False,
verify_on_use: bool = True,
device: Any = 'cpu',
manifest: Optional[ArtifactManifest] = None,
torch_version: Optional[str] = None)

Expand source code

class DownloaderHuggingFace(_VerifiedDownloader):
    """Resolve and authenticate one immutable Hugging Face model artifact."""

    def __init__(
        self,
        file_id: str,
        path_local: str,
        repo_id: Optional[str] = None,
        filename: Optional[str] = None,
        export_filenames_by_torch_minor: Optional[Dict[str, str]] = None,
        fallback_filenames: Optional[List[str]] = None,
        enable_default_torch_export_routing: bool = False,
        *,
        manifest_id: Optional[str] = None,
        revision: Optional[str] = None,
        sha256: Optional[str] = None,
        size_bytes: Optional[int] = None,
        expected_format: Optional[str] = None,
        offline: Optional[bool] = None,
        allow_legacy_models: bool = False,
        verify_on_use: bool = True,
        device: Any = "cpu",
        manifest: Optional[ArtifactManifest] = None,
        torch_version: Optional[str] = None,
    ) -> None:
        super().__init__(
            file_id,
            path_local,
            offline=offline,
            allow_legacy_models=allow_legacy_models,
            verify_on_use=verify_on_use,
        )
        self.repo_id = repo_id if repo_id else file_id
        self.filename = filename if filename else Path(path_local).name
        self.manifest_id = manifest_id
        self.revision = revision
        self.sha256 = sha256
        self.size_bytes = size_bytes
        self.expected_format = expected_format
        self.device = device
        self.manifest = manifest or get_model_manifest()
        self.torch_version = torch_version
        # Retained as inert attributes for source-configuration compatibility.
        self.export_filenames_by_torch_minor = export_filenames_by_torch_minor or {}
        self.fallback_filenames = fallback_filenames or []
        self.enable_default_torch_export_routing = enable_default_torch_export_routing
        self._candidate_index = -1
        self._resolved_candidates: tuple[ArtifactDescriptor, ...] = ()
        self._active_filename: Optional[str] = None
        self._last_candidates: List[str] = []

    def _runtime_version(self) -> str:
        if self.torch_version is not None:
            return self.torch_version
        import torch

        return str(torch.__version__)

    @property
    def _incompatibility_path(self) -> Path:
        return Path(self.path_local).expanduser().parent / ".incompatible.json"

    def _incompatibility_key(self) -> str:
        return incompatibility_key(
            self.manifest.manifest_revision,
            self._runtime_version(),
            self.device,
        )

    def _read_incompatible(self) -> set[str]:
        path = self._incompatibility_path
        try:
            return read_incompatible_artifact_ids(
                path, self._incompatibility_key()
            )
        except ArtifactIntegrityError:
            _quarantine(path, "invalid incompatibility sidecar")
            return set()

    def mark_incompatible(self) -> None:
        """Persist a runtime/schema rejection without executing the artifact again."""
        if self.active_descriptor is None or self.manifest_id is None:
            return
        path = self._incompatibility_path
        _ensure_directory(path.parent)
        with _DirectoryLock(path.parent / ".facetorch-sidecar.lock"):
            raw: Mapping[str, Any] = {}
            if path.is_file():
                try:
                    loaded = json.loads(path.read_text(encoding="utf-8"))
                    if isinstance(loaded, dict):
                        raw = loaded
                except (OSError, json.JSONDecodeError):
                    _quarantine(path, "invalid incompatibility sidecar")
            updated = dict(raw)
            values = set(updated.get(self._incompatibility_key(), []))
            values.add(self.active_descriptor.artifact_id)
            updated[self._incompatibility_key()] = sorted(values)
            temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
            temporary.write_text(
                json.dumps(updated, indent=2, sort_keys=True) + "\n",
                encoding="utf-8",
            )
            os.replace(temporary, path)

    def _resolve_candidates(self) -> tuple[ArtifactDescriptor, ...]:
        if self.manifest_id is None:
            descriptor = _direct_descriptor(
                source="huggingface",
                repo_id=self.repo_id,
                revision=self.revision,
                filename=self.filename,
                path_local=self.path_local,
                sha256=self.sha256,
                size_bytes=self.size_bytes,
                expected_format=self.expected_format,
                device=self.device,
            )
            if descriptor.format == "torchscript" and not self.allow_legacy_models:
                raise ModelCompatibilityError(
                    "Direct Hugging Face TorchScript models require "
                    "allow_legacy_models=True."
                )
            candidates = (descriptor,)
        else:
            candidates = self.manifest.candidates(
                self.manifest_id,
                torch_version=self._runtime_version(),
                device=self.device,
                allow_legacy_models=self.allow_legacy_models,
                incompatible_artifact_ids=self._read_incompatible(),
            )
            if any(item.repo_id != self.repo_id for item in candidates):
                raise ConfigurationError(
                    f"Configured repo_id for {self.manifest_id!r} does not match "
                    "the packaged immutable manifest."
                )
            if self.revision is not None and any(
                item.revision != self.revision for item in candidates
            ):
                raise ConfigurationError(
                    f"Configured revision for {self.manifest_id!r} does not match "
                    "the packaged immutable manifest."
                )
        self._resolved_candidates = candidates
        self._last_candidates = [item.filename for item in candidates]
        return candidates

    def _target_for(self, descriptor: ArtifactDescriptor) -> Path:
        return descriptor.cache_path(self.path_local)

    def resolve_cached_path(self) -> Optional[str]:
        """Activate an existing manifest target without mutating its cache.

        This fast path is intentionally available only when callers explicitly
        disable verification on use.  Downloads and verified cache reuse still go
        through :meth:`run`, which serializes verification and mutation with the
        directory lock.
        """
        if self.verify_on_use is not False:
            return None
        candidates = self._resolve_candidates()
        descriptor = candidates[0]
        target = self._target_for(descriptor)
        if not target.is_file():
            return None
        self._candidate_index = 0
        self._active_filename = descriptor.filename
        return self._activate(descriptor, target)

    def _download_descriptor(
        self, descriptor: ArtifactDescriptor, *, force_download: bool = False
    ) -> str:
        target = self._target_for(descriptor)
        _ensure_directory(target.parent)
        with _DirectoryLock(target.parent / ".facetorch-download.lock"):
            if not force_download:
                existing = self._verified_existing(descriptor, target)
                if existing is not None:
                    self._active_filename = descriptor.filename
                    return existing
            if self.offline:
                raise OfflineCacheError(
                    f"Offline mode requires verified artifact "
                    f"{descriptor.artifact_id!r} at {target}."
                )
            with tempfile.TemporaryDirectory(
                prefix=".facetorch-download-", dir=target.parent
            ) as temporary_dir:
                downloaded_path = hf_hub_download(
                    repo_id=descriptor.repo_id,
                    filename=descriptor.filename,
                    revision=descriptor.revision,
                    local_dir=temporary_dir,
                    force_download=force_download,
                )
                candidate = Path(downloaded_path)
                _atomic_promote(candidate, target, descriptor)
        self._active_filename = descriptor.filename
        return self._activate(descriptor, target)

    def _download_one_candidate(
        self, filename: str, force_download: bool = False
    ) -> str:
        """Download one authenticated candidate; retained for targeted callers."""
        candidates = self._resolved_candidates or self._resolve_candidates()
        try:
            descriptor = next(item for item in candidates if item.filename == filename)
        except StopIteration as exc:
            raise ConfigurationError(
                f"Filename {filename!r} is not an eligible authenticated candidate."
            ) from exc
        return self._download_descriptor(descriptor, force_download=force_download)

    def _build_candidate_filenames(self) -> List[str]:
        """Return only manifest-eligible candidates; never synthesize filenames."""
        return [item.filename for item in self._resolve_candidates()]

    def run(self, force_download: bool = False) -> str:
        candidates = self._resolve_candidates()
        self._candidate_index = 0
        return self._download_descriptor(
            candidates[0], force_download=force_download
        )

    def try_next(self, force_download: bool = False) -> bool:
        """Select one next manifest candidate after a persisted load rejection."""
        candidates = self._resolve_candidates()
        if self.active_descriptor is None:
            next_index = 0
        else:
            try:
                current = next(
                    index
                    for index, item in enumerate(candidates)
                    if item.artifact_id == self.active_descriptor.artifact_id
                )
                next_index = current + 1
            except StopIteration:
                next_index = 0
        if next_index >= len(candidates):
            return False
        self._candidate_index = next_index
        self._download_descriptor(
            candidates[next_index], force_download=force_download
        )
        return True

Resolve and authenticate one immutable Hugging Face model artifact.

Base class for downloaders.

All downloaders should subclass it. All subclass should overwrite:

• Methods:run, supporting to run the download functionality.

Args

-----=

file_id : str

ID of the hosted file (e.g. Google Drive File ID).

path_local : str

The file is downloaded to this local path.

Ancestors

• facetorch.downloader._VerifiedDownloader
• BaseDownloader

Methods

def mark_incompatible(self) ‑> None

Expand source code

def mark_incompatible(self) -> None:
    """Persist a runtime/schema rejection without executing the artifact again."""
    if self.active_descriptor is None or self.manifest_id is None:
        return
    path = self._incompatibility_path
    _ensure_directory(path.parent)
    with _DirectoryLock(path.parent / ".facetorch-sidecar.lock"):
        raw: Mapping[str, Any] = {}
        if path.is_file():
            try:
                loaded = json.loads(path.read_text(encoding="utf-8"))
                if isinstance(loaded, dict):
                    raw = loaded
            except (OSError, json.JSONDecodeError):
                _quarantine(path, "invalid incompatibility sidecar")
        updated = dict(raw)
        values = set(updated.get(self._incompatibility_key(), []))
        values.add(self.active_descriptor.artifact_id)
        updated[self._incompatibility_key()] = sorted(values)
        temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
        temporary.write_text(
            json.dumps(updated, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
        os.replace(temporary, path)

Persist a runtime/schema rejection without executing the artifact again.

def resolve_cached_path(self) ‑> str | None

Expand source code

def resolve_cached_path(self) -> Optional[str]:
    """Activate an existing manifest target without mutating its cache.

    This fast path is intentionally available only when callers explicitly
    disable verification on use.  Downloads and verified cache reuse still go
    through :meth:`run`, which serializes verification and mutation with the
    directory lock.
    """
    if self.verify_on_use is not False:
        return None
    candidates = self._resolve_candidates()
    descriptor = candidates[0]
    target = self._target_for(descriptor)
    if not target.is_file():
        return None
    self._candidate_index = 0
    self._active_filename = descriptor.filename
    return self._activate(descriptor, target)

Activate an existing manifest target without mutating its cache.

This fast path is intentionally available only when callers explicitly disable verification on use. Downloads and verified cache reuse still go through :meth:run, which serializes verification and mutation with the directory lock.

def try_next(self, force_download: bool = False) ‑> bool

Expand source code

def try_next(self, force_download: bool = False) -> bool:
    """Select one next manifest candidate after a persisted load rejection."""
    candidates = self._resolve_candidates()
    if self.active_descriptor is None:
        next_index = 0
    else:
        try:
            current = next(
                index
                for index, item in enumerate(candidates)
                if item.artifact_id == self.active_descriptor.artifact_id
            )
            next_index = current + 1
        except StopIteration:
            next_index = 0
    if next_index >= len(candidates):
        return False
    self._candidate_index = next_index
    self._download_descriptor(
        candidates[next_index], force_download=force_download
    )
    return True

Select one next manifest candidate after a persisted load rejection.

Inherited members

• BaseDownloader:
  • run