Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/git/loader.py
Show First 20 Lines • Show All 128 Lines • ▼ Show 20 Lines | ): | ||||
self.origin_url = url | self.origin_url = url | ||||
self.base_url = base_url | self.base_url = base_url | ||||
self.ignore_history = ignore_history | self.ignore_history = ignore_history | ||||
self.repo_representation = repo_representation | self.repo_representation = repo_representation | ||||
# state initialized in fetch_data | # state initialized in fetch_data | ||||
self.remote_refs: Dict[bytes, bytes] = {} | self.remote_refs: Dict[bytes, bytes] = {} | ||||
self.symbolic_refs: Dict[bytes, bytes] = {} | self.symbolic_refs: Dict[bytes, bytes] = {} | ||||
self.last_visit = None | |||||
def fetch_pack_from_origin( | def fetch_pack_from_origin( | ||||
self, | self, | ||||
origin_url: str, | origin_url: str, | ||||
base_snapshot: Optional[Snapshot], | base_snapshot: Optional[Snapshot], | ||||
do_activity: Callable[[bytes], None], | do_activity: Callable[[bytes], None], | ||||
) -> FetchPackReturn: | ) -> FetchPackReturn: | ||||
"""Fetch a pack from the origin""" | """Fetch a pack from the origin""" | ||||
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines | ) -> Tuple[Dict[bytes, bytes], Dict[bytes, Set[bytes]]]: | ||||
id_to_type[id] = type | id_to_type[id] = type | ||||
type_to_ids[type].add(id) | type_to_ids[type].add(id) | ||||
return id_to_type, type_to_ids | return id_to_type, type_to_ids | ||||
def prepare_origin_visit(self, *args, **kwargs) -> None: | def prepare_origin_visit(self, *args, **kwargs) -> None: | ||||
self.visit_date = datetime.datetime.now(tz=datetime.timezone.utc) | self.visit_date = datetime.datetime.now(tz=datetime.timezone.utc) | ||||
self.origin = Origin(url=self.origin_url) | self.origin = Origin(url=self.origin_url) | ||||
self.last_visit = self.storage.origin_visit_get_latest(self.origin.url) | |||||
def get_full_snapshot(self, origin_url) -> Optional[Snapshot]: | def get_full_snapshot(self, origin_url) -> Optional[Snapshot]: | ||||
visit = self.storage.origin_visit_get_latest(origin_url, require_snapshot=True) | if self.last_visit is None: | ||||
if visit and visit["snapshot"]: | return None | ||||
snapshot = snapshot_get_all_branches(self.storage, visit["snapshot"]) | visit_id = self.last_visit["visit"] | ||||
assert visit_id is not None | |||||
visit_status = self.storage.origin_visit_status_get_latest( | |||||
self.origin_url, visit_id, require_snapshot=True | |||||
) | |||||
if visit_status and visit_status.snapshot: | |||||
snapshot = snapshot_get_all_branches(self.storage, visit_status.snapshot) | |||||
else: | else: | ||||
snapshot = None | snapshot = None | ||||
if snapshot is None: | if snapshot is None: | ||||
return None | return None | ||||
return Snapshot.from_dict(snapshot) | return Snapshot.from_dict(snapshot) | ||||
ardumont: hmmm, wondering if that should not go directly in swh.loader.core.loader then...
That way… | |||||
def prepare(self, *args, **kwargs) -> None: | def prepare(self, *args, **kwargs) -> None: | ||||
assert self.origin is not None | assert self.origin is not None | ||||
base_origin_url = origin_url = self.origin.url | base_origin_url = origin_url = self.origin.url | ||||
prev_snapshot: Optional[Snapshot] = None | prev_snapshot: Optional[Snapshot] = None | ||||
▲ Show 20 Lines • Show All 304 Lines • Show Last 20 Lines |
hmmm, wondering if that should not go directly in swh.loader.core.loader then...
That way, remaining dvcs loaders (svn, git, hg) have access to it directly...
get-last-snapshot or something