def METHOD_NAME(
    repo: "Repo", targets: "TargetType", unpartial: bool = False, **kwargs
) -> Set["HashInfo"]:
    """Save (download) imports from their original source location.
    Imports which are already cached will not be downloaded.
    Returns:
        Objects which were downloaded from source location.
    """
    from dvc.stage.exceptions import DataSourceChanged
    from dvc_data.index import md5, save
    from dvc_data.index.checkout import apply, compare
    from dvc_objects.fs.callbacks import Callback
    downloaded: Set["HashInfo"] = set()
    legacy_unfetched, unfetched, changed = unfetched_view(
        repo.index, targets, unpartial=unpartial, **kwargs
    )
    for dep in changed:
        logger.warning(str(DataSourceChanged(f"{dep.stage} ({dep})")))
    for view, cache in [
        (legacy_unfetched, repo.cache.legacy),
        (unfetched, repo.cache.local),
    ]:
        data_view = view.data["repo"]
        if len(data_view):
            if not cache.fs.exists(cache.path):
                os.makedirs(cache.path)
            with TemporaryDirectory(dir=cache.path) as tmpdir:
                with Callback.as_tqdm_callback(
                    desc="Downloading imports from source",
                    unit="files",
                ) as cb:
                    diff = compare(None, data_view)
                    apply(diff, tmpdir, cache.fs, callback=cb, storage="remote")
                md5(data_view, name=cache.hash_name)
                save(data_view, odb=cache, hardlink=True)
            downloaded.update(
                entry.hash_info
                for _, entry in data_view.iteritems()
                if entry.meta is not None
                and not entry.meta.isdir
                and entry.hash_info is not None
            )
    if unpartial:
        unpartial_imports(partial_view(repo.index, targets, **kwargs))
    return downloaded