Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/loader.py
| Show First 20 Lines • Show All 113 Lines • ▼ Show 20 Lines | def get_default_version(self) -> str: | ||||
| """Retrieve the latest release version if any. | """Retrieve the latest release version if any. | ||||
| Returns: | Returns: | ||||
| Latest version | Latest version | ||||
| """ | """ | ||||
| return "" | return "" | ||||
| def last_snapshot(self) -> Optional[Snapshot]: | def last_snapshot(self, last_visit: Optional[Dict[str, Any]]) -> Optional[Snapshot]: | ||||
| """Retrieve the last snapshot | """Retrieve the last snapshot out of the last visit. | ||||
| """ | """ | ||||
| if not last_visit: | |||||
| return None | |||||
| snapshot = None | snapshot = None | ||||
| visit = self.storage.origin_visit_get_latest(self.url, require_snapshot=True) | visit_status = self.storage.origin_visit_status_get_latest( | ||||
| snapshot_id = None if not visit else visit.get("snapshot") | self.url, last_visit["visit"], require_snapshot=True, | ||||
| ) | |||||
| snapshot_id = None if not visit_status else visit_status.snapshot | |||||
| if snapshot_id is not None: | if snapshot_id is not None: | ||||
| snapshot_dict = snapshot_get_all_branches(self.storage, snapshot_id) | snapshot_dict = snapshot_get_all_branches(self.storage, snapshot_id) | ||||
| if snapshot_dict: | if snapshot_dict: | ||||
| snapshot = Snapshot.from_dict(snapshot_dict) | snapshot = Snapshot.from_dict(snapshot_dict) | ||||
| return snapshot | return snapshot | ||||
| def known_artifacts(self, snapshot: Optional[Snapshot]) -> Dict[Sha1Git, BaseModel]: | def known_artifacts(self, snapshot: Optional[Snapshot]) -> Dict[Sha1Git, BaseModel]: | ||||
| """Retrieve the known releases/artifact for the origin. | """Retrieve the known releases/artifact for the origin. | ||||
| ▲ Show 20 Lines • Show All 161 Lines • ▼ Show 20 Lines | def load(self) -> Dict: | ||||
| self.storage.origin_visit_status_add([visit_status]) | self.storage.origin_visit_status_add([visit_status]) | ||||
| result: Dict[str, Any] = { | result: Dict[str, Any] = { | ||||
| "status": status_load, | "status": status_load, | ||||
| } | } | ||||
| if snapshot_id: | if snapshot_id: | ||||
| result["snapshot_id"] = hash_to_hex(snapshot_id) | result["snapshot_id"] = hash_to_hex(snapshot_id) | ||||
| return result | return result | ||||
| # last visit if any | |||||
| last_visit = self.storage.origin_visit_get_latest(self.url) | |||||
| # Prepare origin and origin_visit | # Prepare origin and origin_visit | ||||
| origin = Origin(url=self.url) | origin = Origin(url=self.url) | ||||
| try: | try: | ||||
| self.storage.origin_add_one(origin) | self.storage.origin_add_one(origin) | ||||
| visit = self.storage.origin_visit_add( | visit = self.storage.origin_visit_add( | ||||
| [ | [ | ||||
| OriginVisit( | OriginVisit( | ||||
| origin=self.url, | origin=self.url, | ||||
| date=self.visit_date, | date=self.visit_date, | ||||
| type=self.visit_type, | type=self.visit_type, | ||||
| status="ongoing", | status="ongoing", | ||||
| snapshot=None, | snapshot=None, | ||||
| ) | ) | ||||
| ] | ] | ||||
| )[0] | )[0] | ||||
| except Exception as e: | except Exception as e: | ||||
| logger.exception("Failed to initialize origin_visit for %s", self.url) | logger.exception("Failed to initialize origin_visit for %s", self.url) | ||||
| sentry_sdk.capture_exception(e) | sentry_sdk.capture_exception(e) | ||||
| return {"status": "failed"} | return {"status": "failed"} | ||||
| try: | try: | ||||
| last_snapshot = self.last_snapshot() | last_snapshot = self.last_snapshot(last_visit) | ||||
| logger.debug("last snapshot: %s", last_snapshot) | logger.debug("last snapshot: %s", last_snapshot) | ||||
| known_artifacts = self.known_artifacts(last_snapshot) | known_artifacts = self.known_artifacts(last_snapshot) | ||||
| logger.debug("known artifacts: %s", known_artifacts) | logger.debug("known artifacts: %s", known_artifacts) | ||||
| except Exception as e: | except Exception as e: | ||||
| logger.exception("Failed to get previous state for %s", self.url) | logger.exception("Failed to get previous state for %s", self.url) | ||||
| sentry_sdk.capture_exception(e) | sentry_sdk.capture_exception(e) | ||||
| status_visit = "partial" | status_visit = "partial" | ||||
| status_load = "failed" | status_load = "failed" | ||||
| ▲ Show 20 Lines • Show All 173 Lines • Show Last 20 Lines | |||||