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 | |||||