Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/mercurial/from_disk.py
| # Copyright (C) 2020-2021 The Software Heritage developers | # Copyright (C) 2020-2021 The Software Heritage developers | ||||||||||||||||||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||||||||||||||||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||||||||||||||||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||||||||||||||||||
| from collections import deque | from collections import deque | ||||||||||||||||||||
| from datetime import datetime | from datetime import datetime | ||||||||||||||||||||
| import os | import os | ||||||||||||||||||||
| from shutil import rmtree | from shutil import rmtree | ||||||||||||||||||||
| from tempfile import mkdtemp | from tempfile import mkdtemp | ||||||||||||||||||||
| from typing import Deque, Dict, Optional, Tuple, TypeVar, Union | from typing import Deque, Dict, List, Optional, Tuple, TypeVar, Union | ||||||||||||||||||||
| from swh.loader.core.loader import BaseLoader | from swh.loader.core.loader import BaseLoader | ||||||||||||||||||||
| from swh.loader.core.utils import clean_dangling_folders | from swh.loader.core.utils import clean_dangling_folders | ||||||||||||||||||||
| from swh.loader.mercurial.utils import parse_visit_date | from swh.loader.mercurial.utils import parse_visit_date | ||||||||||||||||||||
| from swh.model.from_disk import Content, DentryPerms, Directory | from swh.model.from_disk import Content, DentryPerms, Directory | ||||||||||||||||||||
| from swh.model.hashutil import MultiHash, hash_to_bytehex | from swh.model.hashutil import MultiHash, hash_to_bytehex, hash_to_bytes | ||||||||||||||||||||
| from swh.model.model import ( | from swh.model.model import ( | ||||||||||||||||||||
| ObjectType, | ObjectType, | ||||||||||||||||||||
| Origin, | Origin, | ||||||||||||||||||||
| Person, | Person, | ||||||||||||||||||||
| Release, | Release, | ||||||||||||||||||||
| Revision, | Revision, | ||||||||||||||||||||
| RevisionType, | RevisionType, | ||||||||||||||||||||
| Sha1Git, | Sha1Git, | ||||||||||||||||||||
| Snapshot, | Snapshot, | ||||||||||||||||||||
| SnapshotBranch, | SnapshotBranch, | ||||||||||||||||||||
| TargetType, | TargetType, | ||||||||||||||||||||
| TimestampWithTimezone, | TimestampWithTimezone, | ||||||||||||||||||||
| ) | ) | ||||||||||||||||||||
| from swh.model.model import Content as ModelContent | from swh.model.model import Content as ModelContent | ||||||||||||||||||||
| from swh.storage.algos.snapshot import snapshot_get_latest | |||||||||||||||||||||
| from swh.storage.interface import StorageInterface | from swh.storage.interface import StorageInterface | ||||||||||||||||||||
| from . import hgutil | from . import hgutil | ||||||||||||||||||||
| from .archive_extract import tmp_extract | from .archive_extract import tmp_extract | ||||||||||||||||||||
| from .hgutil import HgNodeId | from .hgutil import HgFilteredSet, HgNodeId, HgSpanSet | ||||||||||||||||||||
| FLAG_PERMS = { | FLAG_PERMS = { | ||||||||||||||||||||
| b"l": DentryPerms.symlink, | b"l": DentryPerms.symlink, | ||||||||||||||||||||
| b"x": DentryPerms.executable_content, | b"x": DentryPerms.executable_content, | ||||||||||||||||||||
| b"": DentryPerms.content, | b"": DentryPerms.content, | ||||||||||||||||||||
| } # type: Dict[bytes, DentryPerms] | } # type: Dict[bytes, DentryPerms] | ||||||||||||||||||||
| TEMPORARY_DIR_PREFIX_PATTERN = "swh.loader.mercurial.from_disk" | TEMPORARY_DIR_PREFIX_PATTERN = "swh.loader.mercurial.from_disk" | ||||||||||||||||||||
| ▲ Show 20 Lines • Show All 96 Lines • ▼ Show 20 Lines | ): | ||||||||||||||||||||
| # it is used for differential tree update by store_directories | # it is used for differential tree update by store_directories | ||||||||||||||||||||
| self._last_root = HgDirectory() | self._last_root = HgDirectory() | ||||||||||||||||||||
| # Cache the content hash across revisions to avoid recalculation. | # Cache the content hash across revisions to avoid recalculation. | ||||||||||||||||||||
| self._content_hash_cache: hgutil.LRUCacheDict = hgutil.LRUCacheDict( | self._content_hash_cache: hgutil.LRUCacheDict = hgutil.LRUCacheDict( | ||||||||||||||||||||
| content_cache_size, | content_cache_size, | ||||||||||||||||||||
| ) | ) | ||||||||||||||||||||
| # hg node id of the latest snapshot branch heads | |||||||||||||||||||||
| # used to find what are the new revisions since last snapshot | |||||||||||||||||||||
| self._latest_heads: List[HgNodeId] = [] | |||||||||||||||||||||
| self._load_status = "eventful" | |||||||||||||||||||||
| def pre_cleanup(self) -> None: | def pre_cleanup(self) -> None: | ||||||||||||||||||||
| """As a first step, will try and check for dangling data to cleanup. | """As a first step, will try and check for dangling data to cleanup. | ||||||||||||||||||||
| This should do its best to avoid raising issues. | This should do its best to avoid raising issues. | ||||||||||||||||||||
| """ | """ | ||||||||||||||||||||
| clean_dangling_folders( | clean_dangling_folders( | ||||||||||||||||||||
| self._temp_directory, | self._temp_directory, | ||||||||||||||||||||
| pattern_check=TEMPORARY_DIR_PREFIX_PATTERN, | pattern_check=TEMPORARY_DIR_PREFIX_PATTERN, | ||||||||||||||||||||
| Show All 14 Lines | def prepare_origin_visit(self) -> None: | ||||||||||||||||||||
| """ | """ | ||||||||||||||||||||
| self.origin = Origin(url=self.origin_url) | self.origin = Origin(url=self.origin_url) | ||||||||||||||||||||
| def prepare(self) -> None: | def prepare(self) -> None: | ||||||||||||||||||||
| """Second step executed by the loader to prepare some state needed by | """Second step executed by the loader to prepare some state needed by | ||||||||||||||||||||
| the loader. | the loader. | ||||||||||||||||||||
| """ | """ | ||||||||||||||||||||
| # Set here to allow multiple calls to load on the same loader instance | |||||||||||||||||||||
| self._latest_heads = [] | |||||||||||||||||||||
| latest_snapshot = snapshot_get_latest(self.storage, self.origin_url) | |||||||||||||||||||||
| if latest_snapshot: | |||||||||||||||||||||
| snapshot_branches = [ | |||||||||||||||||||||
| branch.target | |||||||||||||||||||||
| for branch in latest_snapshot.branches.values() | |||||||||||||||||||||
| if branch.target_type != TargetType.ALIAS | |||||||||||||||||||||
| ] | |||||||||||||||||||||
| self._latest_heads = [ | |||||||||||||||||||||
| hash_to_bytes(revision.metadata["node"]) | |||||||||||||||||||||
| for revision in self.storage.revision_get(snapshot_branches) | |||||||||||||||||||||
| if revision and revision.metadata | |||||||||||||||||||||
| ] | |||||||||||||||||||||
| def fetch_data(self) -> bool: | def fetch_data(self) -> bool: | ||||||||||||||||||||
| """Fetch the data from the source the loader is currently loading | """Fetch the data from the source the loader is currently loading | ||||||||||||||||||||
ardumont: can't we make that one for comprehension instead of 2?
```
self._latest_heads =… | |||||||||||||||||||||
Done Inline ActionsMuch better, thanks. I saw the patch was accepted and just rebased it, I should have paid more attention. :) Alphare: Much better, thanks. I saw the patch was accepted and just rebased it, I should have paid more… | |||||||||||||||||||||
Not Done Inline Actions
same proposal with the diff format... (neat trick) the phabricator suggestion edit. ardumont: same proposal with the diff format...
(neat trick) the phabricator suggestion edit. | |||||||||||||||||||||
| 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. | ||||||||||||||||||||
| """ | """ | ||||||||||||||||||||
| if not self.directory: # no local repository | if not self.directory: # no local repository | ||||||||||||||||||||
| self._repo_directory = mkdtemp( | self._repo_directory = mkdtemp( | ||||||||||||||||||||
| prefix=TEMPORARY_DIR_PREFIX_PATTERN, | prefix=TEMPORARY_DIR_PREFIX_PATTERN, | ||||||||||||||||||||
| suffix=f"-{os.getpid()}", | suffix=f"-{os.getpid()}", | ||||||||||||||||||||
| dir=self._temp_directory, | dir=self._temp_directory, | ||||||||||||||||||||
| ) | ) | ||||||||||||||||||||
| self.log.debug( | self.log.debug( | ||||||||||||||||||||
| f"Cloning {self.origin_url} to {self.directory} " | f"Cloning {self.origin_url} to {self.directory} " | ||||||||||||||||||||
| f"with timeout {self._clone_timeout} seconds" | f"with timeout {self._clone_timeout} seconds" | ||||||||||||||||||||
| ) | ) | ||||||||||||||||||||
| hgutil.clone(self.origin_url, self._repo_directory, self._clone_timeout) | hgutil.clone(self.origin_url, self._repo_directory, self._clone_timeout) | ||||||||||||||||||||
| else: # existing local repository | else: # existing local repository | ||||||||||||||||||||
| # Allow to load on disk repository without cloning | # Allow to load on disk repository without cloning | ||||||||||||||||||||
| # for testing purpose. | # for testing purpose. | ||||||||||||||||||||
| self._repo_directory = self.directory | self._repo_directory = self.directory | ||||||||||||||||||||
| self._repo = hgutil.repository(self._repo_directory) | self._repo = hgutil.repository(self._repo_directory) | ||||||||||||||||||||
| return False | return False | ||||||||||||||||||||
Done Inline Actionsshould add type annotation douardda: should add type annotation | |||||||||||||||||||||
| def get_hg_revs_to_load(self) -> Union[HgFilteredSet, HgSpanSet]: | |||||||||||||||||||||
| """Return the hg revision numbers to load.""" | |||||||||||||||||||||
Not Done Inline Actions
ardumont: | |||||||||||||||||||||
| assert self._repo is not None | |||||||||||||||||||||
| repo: hgutil.Repository = self._repo | |||||||||||||||||||||
| if self._latest_heads: | |||||||||||||||||||||
| existing_heads = [] # heads that still exist in the repository | |||||||||||||||||||||
| for hg_nodeid in self._latest_heads: | |||||||||||||||||||||
| try: | |||||||||||||||||||||
| rev = repo[hg_nodeid].rev() | |||||||||||||||||||||
Not Done Inline Actions
ardumont: | |||||||||||||||||||||
| existing_heads.append(rev) | |||||||||||||||||||||
| except KeyError: # the node does not exist anymore | |||||||||||||||||||||
| pass | |||||||||||||||||||||
| # select revisions that are not ancestors of heads | |||||||||||||||||||||
| # and not the heads themselves | |||||||||||||||||||||
| new_revs = repo.revs("not ::(%ld)", existing_heads) | |||||||||||||||||||||
| # for now, reload all revisions if there are new commits | |||||||||||||||||||||
| # otherwise the loader will crash on missing parents | |||||||||||||||||||||
| # incremental loading will come in next commits | |||||||||||||||||||||
| if new_revs: | |||||||||||||||||||||
| return repo.revs("all()") | |||||||||||||||||||||
| else: | |||||||||||||||||||||
| return new_revs | |||||||||||||||||||||
| else: | |||||||||||||||||||||
| return repo.revs("all()") | |||||||||||||||||||||
| def store_data(self): | def store_data(self): | ||||||||||||||||||||
| """Store fetched data in the database.""" | """Store fetched data in the database.""" | ||||||||||||||||||||
| for rev in self._repo: | revs = self.get_hg_revs_to_load() | ||||||||||||||||||||
Done Inline Actionsnit: why not if not revs: ? douardda: nit: why not `if not revs:` ? | |||||||||||||||||||||
Done Inline ActionsWas not sure __bool__ / __nonzero__ was implemented by the class acezar: Was not sure `__bool__` / ` __nonzero__` was implemented by the class | |||||||||||||||||||||
| if not revs: | |||||||||||||||||||||
| self._load_status = "uneventful" | |||||||||||||||||||||
| return | |||||||||||||||||||||
| for rev in revs: | |||||||||||||||||||||
| self.store_revision(self._repo[rev]) | self.store_revision(self._repo[rev]) | ||||||||||||||||||||
| branch_by_hg_nodeid: Dict[HgNodeId, bytes] = { | branch_by_hg_nodeid: Dict[HgNodeId, bytes] = { | ||||||||||||||||||||
| hg_nodeid: name for name, hg_nodeid in hgutil.branches(self._repo).items() | hg_nodeid: name for name, hg_nodeid in hgutil.branches(self._repo).items() | ||||||||||||||||||||
| } | } | ||||||||||||||||||||
| tags_by_name: Dict[bytes, HgNodeId] = self._repo.tags() | tags_by_name: Dict[bytes, HgNodeId] = self._repo.tags() | ||||||||||||||||||||
| tags_by_hg_nodeid: Dict[HgNodeId, bytes] = { | tags_by_hg_nodeid: Dict[HgNodeId, bytes] = { | ||||||||||||||||||||
| hg_nodeid: name for name, hg_nodeid in tags_by_name.items() | hg_nodeid: name for name, hg_nodeid in tags_by_name.items() | ||||||||||||||||||||
| Show All 26 Lines | def store_data(self): | ||||||||||||||||||||
| ) | ) | ||||||||||||||||||||
| snapshot = Snapshot(branches=snapshot_branches) | snapshot = Snapshot(branches=snapshot_branches) | ||||||||||||||||||||
| self.storage.snapshot_add([snapshot]) | self.storage.snapshot_add([snapshot]) | ||||||||||||||||||||
| self.flush() | self.flush() | ||||||||||||||||||||
| self.loaded_snapshot_id = snapshot.id | self.loaded_snapshot_id = snapshot.id | ||||||||||||||||||||
| def load_status(self) -> Dict[str, str]: | |||||||||||||||||||||
| """Detailed loading status. | |||||||||||||||||||||
| Defaults to logging an eventful load. | |||||||||||||||||||||
| Returns: a dictionary that is eventually passed back as the task's | |||||||||||||||||||||
| result to the scheduler, allowing tuning of the task recurrence | |||||||||||||||||||||
| mechanism. | |||||||||||||||||||||
| """ | |||||||||||||||||||||
| return { | |||||||||||||||||||||
| "status": self._load_status, | |||||||||||||||||||||
| } | |||||||||||||||||||||
| def get_revision_id_from_hg_nodeid(self, hg_nodeid: HgNodeId) -> Sha1Git: | def get_revision_id_from_hg_nodeid(self, hg_nodeid: HgNodeId) -> Sha1Git: | ||||||||||||||||||||
| """Return the swhid of a revision given its hg nodeid. | """Return the swhid of a revision given its hg nodeid. | ||||||||||||||||||||
| Args: | Args: | ||||||||||||||||||||
| hg_nodeid: the hg nodeid of the revision. | hg_nodeid: the hg nodeid of the revision. | ||||||||||||||||||||
| Returns: | Returns: | ||||||||||||||||||||
| the swhid of the revision. | the swhid of the revision. | ||||||||||||||||||||
| ▲ Show 20 Lines • Show All 263 Lines • Show Last 20 Lines | |||||||||||||||||||||
can't we make that one for comprehension instead of 2?
self._latest_heads = [ hash_to_bytes(revision.metadata["node"]) for revision in self.storage.revision_get(snapshot_branches) if revision and revision.metadata ]