Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/npm/loader.py
Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Lines | def _known_versions(self, last_snapshot): | ||||
for revision in known_revisions: | for revision in known_revisions: | ||||
if not revision: | if not revision: | ||||
continue | continue | ||||
if 'package_source' in revision['metadata']: | if 'package_source' in revision['metadata']: | ||||
package = revision['metadata']['package_source'] | package = revision['metadata']['package_source'] | ||||
ret[(package['version'], package['sha1'])] = revision['id'] | ret[(package['version'], package['sha1'])] = revision['id'] | ||||
return ret | return ret | ||||
def _last_snapshot(self): | def last_snapshot(self): | ||||
""" | """ | ||||
Retrieve the last snapshot of the npm package if any. | Retrieve the last snapshot of the npm package if any. | ||||
""" | """ | ||||
snapshot = self.storage.snapshot_get_latest(self.origin_id) | snapshot = self.storage.snapshot_get_latest(self.origin_id) | ||||
if snapshot and snapshot.pop('next_branch', None): | if snapshot and snapshot.pop('next_branch', None): | ||||
snapshot = snapshot_get_all_branches(self.storage, snapshot['id']) | snapshot = snapshot_get_all_branches(self.storage, snapshot['id']) | ||||
return snapshot | return snapshot | ||||
Show All 10 Lines | def prepare(self, package_name, package_url, package_metadata_url): | ||||
self.package_name = package_name | self.package_name = package_name | ||||
self.origin_url = package_url | self.origin_url = package_url | ||||
self.package_contents = [] | self.package_contents = [] | ||||
self.package_directories = [] | self.package_directories = [] | ||||
self.package_revisions = [] | self.package_revisions = [] | ||||
self.package_load_status = 'uneventful' | self.package_load_status = 'uneventful' | ||||
self.package_visit_status = 'full' | self.package_visit_status = 'full' | ||||
last_snapshot = self._last_snapshot() | last_snapshot = self.last_snapshot() | ||||
self.known_versions = self._known_versions(last_snapshot) | self.known_versions = self._known_versions(last_snapshot) | ||||
self.new_versions = \ | self.new_versions = \ | ||||
self.npm_client.prepare_package_versions(self.known_versions) | self.npm_client.prepare_package_versions(self.known_versions) | ||||
def fetch_data(self): | def fetch_data(self): | ||||
""" | """ | ||||
Called once per package release version to process. | Called once per package release version to process. | ||||
Show All 15 Lines | def fetch_data(self): | ||||
data = next(self.new_versions) | data = next(self.new_versions) | ||||
self.package_load_status = 'eventful' | self.package_load_status = 'eventful' | ||||
except StopIteration: | except StopIteration: | ||||
self.done = True | self.done = True | ||||
return False | return False | ||||
package_metadata, author, package_source_data, dir_path = data | package_metadata, author, package_source_data, dir_path = data | ||||
# package release tarball was corrupted | |||||
if package_metadata is None: | |||||
return not self.done | |||||
dir_path = dir_path.encode('utf-8') | dir_path = dir_path.encode('utf-8') | ||||
directory = Directory.from_disk(path=dir_path, data=True) | directory = Directory.from_disk(path=dir_path, data=True) | ||||
objects = directory.collect() | objects = directory.collect() | ||||
self.package_contents = objects['content'].values() | self.package_contents = objects['content'].values() | ||||
self.package_directories = objects['directory'].values() | self.package_directories = objects['directory'].values() | ||||
date = date_parser.parse(package_source_data['date']) | date = date_parser.parse(package_source_data['date']) | ||||
▲ Show 20 Lines • Show All 100 Lines • Show Last 20 Lines |