swh.loader.package.deposit.loader module

swh.loader.package.deposit.loader.now() → datetime

class swh.loader.package.deposit.loader.DepositPackageInfo(url: str, version: str, filename: str, author_date: datetime, commit_date: datetime, client: str, id: int, collection: str, author: Person, committer: Person, release_notes: Optional[str], *, directory_extrinsic_metadata: List[RawExtrinsicMetadataCore] = [], checksums: Dict[str, str] = {})

Bases: BasePackageInfo

Method generated by attrs for class DepositPackageInfo.

author_date

dateCreated if any, deposit completed_date otherwise

Type:

codemeta

commit_date

datePublished if any, deposit completed_date otherwise

Type:

codemeta

id

Internal ID of the deposit in the deposit DB

collection

The collection in the deposit; see SWORD specification.

classmethod from_metadata(metadata: Dict[str, Any], url: str, filename: str, version: str) → DepositPackageInfo

extid() → None

Returns a unique intrinsic identifier of this package info, or None if this package info is not ‘deduplicatable’ (meaning that we will always load it, instead of checking the ExtID storage to see if we already did)

class swh.loader.package.deposit.loader.DepositLoader(storage: StorageInterface, url: str, deposit_id: str, deposit_client: ApiClient, default_filename: str = 'archive.tar', **kwargs: Any)

Bases: PackageLoader[DepositPackageInfo]

Load a deposited artifact into swh archive.

Constructor

Parameters:

• url – Origin url to associate the artifacts/metadata to

• deposit_id – Deposit identity

• deposit_client – Deposit api client

visit_type: str = 'deposit'

classmethod from_configfile(**kwargs: Any)

Instantiate a loader from the configuration loaded from the SWH_CONFIG_FILENAME envvar, with potential extra keyword arguments if their value is not None.

Parameters:

kwargs – kwargs passed to the loader instantiation

get_versions() → Sequence[str]

Return the list of all published package versions.

Raises:

class – swh.loader.exception.NotFound error when failing to read the published package versions.

Returns:

Sequence of published versions

get_metadata_authority() → MetadataAuthority

For package loaders that get extrinsic metadata, returns the authority the metadata are coming from.

get_metadata_fetcher() → MetadataFetcher

Returns a MetadataFetcher instance representing this package loader; which is used to for adding provenance information to extracted extrinsic metadata, if any.

get_package_info(version: str) → Iterator[Tuple[str, DepositPackageInfo]]

Given a release version of a package, retrieve the associated

package information for such version.

Parameters:

version – Package version

Returns:

(branch name, package metadata)

download_package(p_info: DepositPackageInfo, tmpdir: str) → List[Tuple[str, Mapping]]

Override to allow use of the dedicated deposit client

build_release(p_info: DepositPackageInfo, uncompressed_path: str, directory: bytes) → Optional[Release]

Build the release from the archive metadata (extrinsic artifact metadata) and the intrinsic metadata.

Parameters:

• p_info – Package information

• uncompressed_path – Artifact uncompressed path on disk

get_extrinsic_origin_metadata() → List[RawExtrinsicMetadataCore]

Returns metadata items, used by build_extrinsic_origin_metadata.

metadata()

Returns metadata from the deposit server

load() → Dict

Load for a specific origin the associated contents.

1. Get the list of versions in an origin.

2. Get the snapshot from the previous run of the loader, and filter out versions that were already loaded, if their extids match

Then, for each remaining version in the origin

3. Fetch the files for one package version By default, this can be implemented as a simple HTTP request. Loaders with more specific requirements can override this, e.g.: the PyPI loader checks the integrity of the downloaded files; the Debian loader has to download and check several files for one package version.

4. Extract the downloaded files. By default, this would be a universal archive/tarball extraction.

   Loaders for specific formats can override this method (for instance, the Debian loader uses dpkg-source -x).

5. Convert the extracted directory to a set of Software Heritage objects Using swh.model.from_disk.

6. Extract the metadata from the unpacked directories This would only be applicable for “smart” loaders like npm (parsing the package.json), PyPI (parsing the PKG-INFO file) or Debian (parsing debian/changelog and debian/control).

   On “minimal-metadata” sources such as the GNU archive, the lister should provide the minimal set of metadata needed to populate the revision/release objects (authors, dates) as an argument to the task.

7. Generate the revision/release objects for the given version. From the data generated at steps 3 and 4.

end for each

8. Generate and load the snapshot for the visit

Using the revisions/releases collected at step 7., and the branch information from step 2., generate a snapshot and load it into the Software Heritage archive

finalize_visit(status_visit: str, errors: Optional[List[str]] = None, **kwargs) → Dict[str, Any]

Finalize the visit:

• flush eventual unflushed data to storage

• update origin visit’s status

• return the task’s status

origin: Origin

loaded_snapshot_id: Optional[bytes]

parent_origins: Optional[List[Origin]]

If the given origin is a “forge fork” (ie. created with the “Fork” button of GitHub-like forges), build_extrinsic_origin_metadata() sets this to a list of origins it was forked from; closest parent first.

swh.loader.package.deposit.loader.parse_author(author) → Person

See prior fixme

class swh.loader.package.deposit.loader.ApiClient(url, auth: Optional[Mapping[str, str]])

Bases: object

Private Deposit Api client

do(method: str, url: str, *args, **kwargs)

Internal method to deal with requests, possibly with basic http

authentication.

Parameters:

method (str) – supported http methods as in get/post/put

Returns:

The request’s execution output

archive_get(deposit_id: Union[int, str], tmpdir: str, filename: str) → Tuple[str, Dict]

Retrieve deposit’s archive artifact locally

metadata_url(deposit_id: Union[int, str]) → str

metadata_get(deposit_id: Union[int, str]) → Dict[str, Any]

Retrieve deposit’s metadata artifact as json

status_update(deposit_id: Union[int, str], status: str, errors: Optional[List[str]] = None, release_id: Optional[str] = None, directory_id: Optional[str] = None, snapshot_id: Optional[str] = None, origin_url: Optional[str] = None)

Update deposit’s information including status, and persistent identifiers result of the loading.