swh.loader.core.utils module#

swh.loader.core.utils.clean_dangling_folders(dirpath: str, pattern_check: str, log=None) None[source]#
Clean up potential dangling temporary working folder rooted at dirpath. Those

folders must match a dedicated pattern and not belonging to a live pid.

Parameters:
  • dirpath – Path to check for dangling files

  • pattern_check – A dedicated pattern to check on first level directory (e.g swh.loader.mercurial., swh.loader.svn.)

  • log (Logger) – Optional logger

exception swh.loader.core.utils.CloneTimeout[source]#

Bases: Exception

exception swh.loader.core.utils.CloneFailure[source]#

Bases: Exception

swh.loader.core.utils.clone_with_timeout(src: str, dest: str, clone_func: Callable[[], None], timeout: float) None[source]#

Clone a repository with timeout.

Parameters:
  • src – clone source

  • dest – clone destination

  • clone_func – callable that does the actual cloning

  • timeout – timeout in seconds

swh.loader.core.utils.parse_visit_date(visit_date: Optional[Union[datetime, str]]) Optional[datetime][source]#

Convert visit date from either None, a string or a datetime to either None or datetime.

swh.loader.core.utils.nix_hashes(filepath: Path, hash_names: Iterable[str]) MultiHash[source]#

Compute nix-store hashes on filepath.

Raises:

FileNotFoundError in case the nix-store command is not available on the system.

swh.loader.core.utils.compute_nar_hashes(filepath: Path, hash_names: List[str] = ['sha256'], is_tarball=True) Dict[str, str][source]#

Compute nar checksums dict out of a filepath (tarball or plain file).

If it’s a tarball, this uncompresses the tarball in a temporary directory to compute the nix hashes (and then cleans it up).

Parameters:
  • filepath – The tarball (if is_tarball is True) or a filepath

  • hash_names – The list of checksums to compute

  • is_tarball – Whether filepath represents a tarball or not

Returns:

The dict of checksums values whose keys are present in hash_names.