Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/core/loader.py
| Show First 20 Lines • Show All 230 Lines • ▼ Show 20 Lines | def fetch_data(self) -> bool: | |||||||||||||||||||||||||||||||||||||||||||||||||
| Returns: | Returns: | |||||||||||||||||||||||||||||||||||||||||||||||||
| a value that is interpreted as a boolean. If True, fetch_data needs | a value that is interpreted as a boolean. If True, fetch_data needs | |||||||||||||||||||||||||||||||||||||||||||||||||
| to be called again to complete loading. | to be called again to complete loading. | |||||||||||||||||||||||||||||||||||||||||||||||||
| """ | """ | |||||||||||||||||||||||||||||||||||||||||||||||||
| raise NotImplementedError | raise NotImplementedError | |||||||||||||||||||||||||||||||||||||||||||||||||
| def store_data(self): | def store_data(self, create_partial_visit: bool = False): | |||||||||||||||||||||||||||||||||||||||||||||||||
| """Store fetched data in the database. | """Store fetched data in the database. Use create_partial_visit boolean to decide | |||||||||||||||||||||||||||||||||||||||||||||||||
| whether to create partial snapshot or not. | ||||||||||||||||||||||||||||||||||||||||||||||||||
| Should call the :func:`maybe_load_xyz` methods, which handle the | ||||||||||||||||||||||||||||||||||||||||||||||||||
| bundles sent to storage, rather than send directly. | ||||||||||||||||||||||||||||||||||||||||||||||||||
| """ | """ | |||||||||||||||||||||||||||||||||||||||||||||||||
| raise NotImplementedError | raise NotImplementedError | |||||||||||||||||||||||||||||||||||||||||||||||||
| def store_metadata(self) -> None: | def store_metadata(self) -> None: | |||||||||||||||||||||||||||||||||||||||||||||||||
| """Store fetched metadata in the database. | """Store fetched metadata in the database. | |||||||||||||||||||||||||||||||||||||||||||||||||
| For more information, see implementation in :class:`DepositLoader`. | For more information, see implementation in :class:`DepositLoader`. | |||||||||||||||||||||||||||||||||||||||||||||||||
| """ | """ | |||||||||||||||||||||||||||||||||||||||||||||||||
| ▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Lines | def load(self) -> Dict[str, str]: | |||||||||||||||||||||||||||||||||||||||||||||||||
| "Load origin '%s' with type '%s'", self.origin.url, self.visit.type | "Load origin '%s' with type '%s'", self.origin.url, self.visit.type | |||||||||||||||||||||||||||||||||||||||||||||||||
| ) | ) | |||||||||||||||||||||||||||||||||||||||||||||||||
| try: | try: | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.prepare() | self.prepare() | |||||||||||||||||||||||||||||||||||||||||||||||||
| while True: | while True: | |||||||||||||||||||||||||||||||||||||||||||||||||
| more_data_to_fetch = self.fetch_data() | more_data_to_fetch = self.fetch_data() | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.store_data() | # if more data to fetch, let's also create a partial visit | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.store_data(create_partial_visit=more_data_to_fetch) | ||||||||||||||||||||||||||||||||||||||||||||||||||
vlorentz: This should use a keyword argument, for readability (make it clear what the argument means) | ||||||||||||||||||||||||||||||||||||||||||||||||||
| if not more_data_to_fetch: | if not more_data_to_fetch: | |||||||||||||||||||||||||||||||||||||||||||||||||
| break | break | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.store_metadata() | self.store_metadata() | |||||||||||||||||||||||||||||||||||||||||||||||||
| visit_status = OriginVisitStatus( | visit_status = OriginVisitStatus( | |||||||||||||||||||||||||||||||||||||||||||||||||
| origin=self.origin.url, | origin=self.origin.url, | |||||||||||||||||||||||||||||||||||||||||||||||||
| visit=self.visit.visit, | visit=self.visit.visit, | |||||||||||||||||||||||||||||||||||||||||||||||||
| type=self.visit_type, | type=self.visit_type, | |||||||||||||||||||||||||||||||||||||||||||||||||
| date=now(), | date=now(), | |||||||||||||||||||||||||||||||||||||||||||||||||
| status=self.visit_status(), | status=self.visit_status(), | |||||||||||||||||||||||||||||||||||||||||||||||||
| snapshot=self.loaded_snapshot_id, | snapshot=self.loaded_snapshot_id, | |||||||||||||||||||||||||||||||||||||||||||||||||
| ) | ) | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.origin_visit_status_add([visit_status]) | self.storage.origin_visit_status_add([visit_status]) | |||||||||||||||||||||||||||||||||||||||||||||||||
Done Inline Actions
The current code uses slightly ambiguous "create_partial_snapshot" parameter to store_data, which makes it create an OVS in some cases; and create the OVS here in other cases. Instead, you can create the OVS here, simply by moving it inside the loop. This keeps store_data as it was before the diff (doesn't need to concern itself with creating objects) and avoid duplicating OVS creation. Just because Phabricator mangles the diff, this is what the new code should look like: while True:
t1 = time.monotonic()
more_data_to_fetch = self.fetch_data()
t2 = time.monotonic()
total_time_fetch_data += t2 - t1
more_data_to_fetch = self.process_data() and more_data_to_fetch
t3 = time.monotonic()
total_time_process_data += t3 - t2
status = "partial" if more_data_to_fetch else self.visit_status()
self.store_data()
visit_status = OriginVisitStatus(
origin=self.origin.url,
visit=self.visit.visit,
type=self.visit_type,
date=now(),
status=status,
snapshot=self.loaded_snapshot_id,
)
self.storage.origin_visit_status_add([visit_status])
t4 = time.monotonic()
total_time_store_data += t4 - t3
if not more_data_to_fetch:
break
self.statsd_timing("fetch_data", total_time_fetch_data * 1000.0)
self.statsd_timing("process_data", total_time_process_data * 1000.0)
self.statsd_timing("store_data", total_time_store_data * 1000.0)vlorentz: The current code uses slightly ambiguous "create_partial_snapshot" parameter to store_data… | ||||||||||||||||||||||||||||||||||||||||||||||||||
Done Inline Actionsoops, you should move the status = "partial" if more_data_to_fetch else self.visit_status() line after the call to store_data in both my snippets vlorentz: oops, you should move the `status = "partial" if more_data_to_fetch else self.visit_status()`… | ||||||||||||||||||||||||||||||||||||||||||||||||||
| self.post_load() | self.post_load() | |||||||||||||||||||||||||||||||||||||||||||||||||
| except Exception as e: | except Exception as e: | |||||||||||||||||||||||||||||||||||||||||||||||||
| if isinstance(e, NotFound): | if isinstance(e, NotFound): | |||||||||||||||||||||||||||||||||||||||||||||||||
| status = "not_found" | status = "not_found" | |||||||||||||||||||||||||||||||||||||||||||||||||
| task_status = "uneventful" | task_status = "uneventful" | |||||||||||||||||||||||||||||||||||||||||||||||||
| else: | else: | |||||||||||||||||||||||||||||||||||||||||||||||||
| status = "partial" if self.loaded_snapshot_id else "failed" | status = "partial" if self.loaded_snapshot_id else "failed" | |||||||||||||||||||||||||||||||||||||||||||||||||
| task_status = "failed" | task_status = "failed" | |||||||||||||||||||||||||||||||||||||||||||||||||
| ▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines | class DVCSLoader(BaseLoader): | |||||||||||||||||||||||||||||||||||||||||||||||||
| def get_snapshot(self) -> Snapshot: | def get_snapshot(self) -> Snapshot: | |||||||||||||||||||||||||||||||||||||||||||||||||
| """Get the snapshot that needs to be loaded""" | """Get the snapshot that needs to be loaded""" | |||||||||||||||||||||||||||||||||||||||||||||||||
| raise NotImplementedError | raise NotImplementedError | |||||||||||||||||||||||||||||||||||||||||||||||||
| def eventful(self) -> bool: | def eventful(self) -> bool: | |||||||||||||||||||||||||||||||||||||||||||||||||
| """Whether the load was eventful""" | """Whether the load was eventful""" | |||||||||||||||||||||||||||||||||||||||||||||||||
| raise NotImplementedError | raise NotImplementedError | |||||||||||||||||||||||||||||||||||||||||||||||||
| def store_data(self) -> None: | def store_data(self, create_partial_visit: bool = False) -> None: | |||||||||||||||||||||||||||||||||||||||||||||||||
| assert self.origin | assert self.origin | |||||||||||||||||||||||||||||||||||||||||||||||||
| if self.save_data_path: | if self.save_data_path: | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.save_data() | self.save_data() | |||||||||||||||||||||||||||||||||||||||||||||||||
| if self.has_contents(): | if self.has_contents(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| for obj in self.get_contents(): | for obj in self.get_contents(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| if isinstance(obj, Content): | if isinstance(obj, Content): | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.content_add([obj]) | self.storage.content_add([obj]) | |||||||||||||||||||||||||||||||||||||||||||||||||
| elif isinstance(obj, SkippedContent): | elif isinstance(obj, SkippedContent): | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.skipped_content_add([obj]) | self.storage.skipped_content_add([obj]) | |||||||||||||||||||||||||||||||||||||||||||||||||
| else: | else: | |||||||||||||||||||||||||||||||||||||||||||||||||
| raise TypeError(f"Unexpected content type: {obj}") | raise TypeError(f"Unexpected content type: {obj}") | |||||||||||||||||||||||||||||||||||||||||||||||||
| if self.has_directories(): | if self.has_directories(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| for directory in self.get_directories(): | for directory in self.get_directories(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.directory_add([directory]) | self.storage.directory_add([directory]) | |||||||||||||||||||||||||||||||||||||||||||||||||
| if self.has_revisions(): | if self.has_revisions(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| for revision in self.get_revisions(): | for revision in self.get_revisions(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.revision_add([revision]) | self.storage.revision_add([revision]) | |||||||||||||||||||||||||||||||||||||||||||||||||
| if self.has_releases(): | if self.has_releases(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| for release in self.get_releases(): | for release in self.get_releases(): | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.release_add([release]) | self.storage.release_add([release]) | |||||||||||||||||||||||||||||||||||||||||||||||||
| snapshot = self.get_snapshot() | snapshot = self.get_snapshot() | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.snapshot_add([snapshot]) | self.storage.snapshot_add([snapshot]) | |||||||||||||||||||||||||||||||||||||||||||||||||
| # More work to do, we make a partial visit targeting the snapshot though That | ||||||||||||||||||||||||||||||||||||||||||||||||||
| # should ease further visit if we somehow can't make it through the ingestion. | ||||||||||||||||||||||||||||||||||||||||||||||||||
| if create_partial_visit: | ||||||||||||||||||||||||||||||||||||||||||||||||||
| assert isinstance(self.visit.visit, int) | ||||||||||||||||||||||||||||||||||||||||||||||||||
vlorentzUnsubmitted Done Inline Actions
maybe? vlorentz: maybe? | ||||||||||||||||||||||||||||||||||||||||||||||||||
ardumontAuthorUnsubmitted Done Inline ActionsI recall that's what i initially did and mypy was not happy about it. ardumont: I recall that's what i initially did and mypy was not happy about it. | ||||||||||||||||||||||||||||||||||||||||||||||||||
| visit_status = OriginVisitStatus( | ||||||||||||||||||||||||||||||||||||||||||||||||||
| origin=self.origin.url, | ||||||||||||||||||||||||||||||||||||||||||||||||||
| visit=self.visit.visit, | ||||||||||||||||||||||||||||||||||||||||||||||||||
| type=self.visit_type, | ||||||||||||||||||||||||||||||||||||||||||||||||||
| date=now(), | ||||||||||||||||||||||||||||||||||||||||||||||||||
| status="partial", | ||||||||||||||||||||||||||||||||||||||||||||||||||
| snapshot=snapshot.id, | ||||||||||||||||||||||||||||||||||||||||||||||||||
| ) | ||||||||||||||||||||||||||||||||||||||||||||||||||
| self.storage.origin_visit_status_add([visit_status]) | ||||||||||||||||||||||||||||||||||||||||||||||||||
| self.flush() | self.flush() | |||||||||||||||||||||||||||||||||||||||||||||||||
| self.loaded_snapshot_id = snapshot.id | self.loaded_snapshot_id = snapshot.id | |||||||||||||||||||||||||||||||||||||||||||||||||
Done Inline Actionsshould be in the if. We don't really want the flush to happen each time the loop occurs. ardumont: should be in the if. We don't really want the flush to happen each time the loop occurs.
Plus… | ||||||||||||||||||||||||||||||||||||||||||||||||||
Done Inline Actionsthat comment was for a previous implem, it's now irrelevant. ardumont: that comment was for a previous implem, it's now irrelevant. | ||||||||||||||||||||||||||||||||||||||||||||||||||
This should use a keyword argument, for readability (make it clear what the argument means)