Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/cookers/git_bare.py
Show First 20 Lines • Show All 70 Lines • ▼ Show 20 Lines | def assert_never(value: NoReturn, msg) -> NoReturn: | ||||
See https://mypy.readthedocs.io/en/latest/literal_types.html#exhaustive-checks | See https://mypy.readthedocs.io/en/latest/literal_types.html#exhaustive-checks | ||||
for details. | for details. | ||||
""" | """ | ||||
assert False, msg | assert False, msg | ||||
class GitBareCooker(BaseVaultCooker): | class GitBareCooker(BaseVaultCooker): | ||||
BUNDLE_TYPE = "git_bare" | |||||
SUPPORTED_OBJECT_TYPES = { | |||||
identifiers.ObjectType[obj_type.name] for obj_type in RootObjectType | |||||
} | |||||
use_fsck = True | use_fsck = True | ||||
obj_type: RootObjectType | obj_type: RootObjectType | ||||
def __init__(self, *args, **kwargs): | def __init__(self, *args, **kwargs): | ||||
super().__init__(*args, **kwargs) | super().__init__(*args, **kwargs) | ||||
self.obj_type = RootObjectType(self.bundle_type.split("_")[0]) | self.obj_type = RootObjectType[self.swhid.object_type.name] | ||||
def cache_type_key(self) -> str: | |||||
return self.bundle_type | |||||
def check_exists(self) -> bool: | def check_exists(self) -> bool: | ||||
if self.obj_type is RootObjectType.REVISION: | if self.obj_type is RootObjectType.REVISION: | ||||
return not list(self.storage.revision_missing([self.obj_id])) | return not list(self.storage.revision_missing([self.obj_id])) | ||||
elif self.obj_type is RootObjectType.DIRECTORY: | elif self.obj_type is RootObjectType.DIRECTORY: | ||||
return not list(self.storage.directory_missing([self.obj_id])) | return not list(self.storage.directory_missing([self.obj_id])) | ||||
elif self.obj_type is RootObjectType.SNAPSHOT: | elif self.obj_type is RootObjectType.SNAPSHOT: | ||||
return not list(self.storage.snapshot_missing([self.obj_id])) | return not list(self.storage.snapshot_missing([self.obj_id])) | ||||
else: | else: | ||||
assert_never(self.obj_type, f"Unexpected root object type: {self.obj_type}") | assert_never(self.obj_type, f"Unexpected root object type: {self.obj_type}") | ||||
def obj_swhid(self) -> identifiers.CoreSWHID: | |||||
return identifiers.CoreSWHID( | |||||
object_type=identifiers.ObjectType[self.obj_type.name], | |||||
object_id=self.obj_id, | |||||
) | |||||
def _push(self, stack: List[Sha1Git], obj_ids: Iterable[Sha1Git]) -> None: | def _push(self, stack: List[Sha1Git], obj_ids: Iterable[Sha1Git]) -> None: | ||||
assert not isinstance(obj_ids, bytes) | assert not isinstance(obj_ids, bytes) | ||||
revision_ids = [id_ for id_ in obj_ids if id_ not in self._seen] | revision_ids = [id_ for id_ in obj_ids if id_ not in self._seen] | ||||
self._seen.update(revision_ids) | self._seen.update(revision_ids) | ||||
stack.extend(revision_ids) | stack.extend(revision_ids) | ||||
def _pop(self, stack: List[Sha1Git], n: int) -> List[Sha1Git]: | def _pop(self, stack: List[Sha1Git], n: int) -> List[Sha1Git]: | ||||
obj_ids = stack[-n:] | obj_ids = stack[-n:] | ||||
▲ Show 20 Lines • Show All 131 Lines • ▼ Show 20 Lines | def write_refs(self, snapshot=None): | ||||
for (ref_name, ref_target) in refs.items(): | for (ref_name, ref_target) in refs.items(): | ||||
path = os.path.join(self.gitdir.encode(), ref_name) | path = os.path.join(self.gitdir.encode(), ref_name) | ||||
os.makedirs(os.path.dirname(path), exist_ok=True) | os.makedirs(os.path.dirname(path), exist_ok=True) | ||||
with open(path, "wb") as fd: | with open(path, "wb") as fd: | ||||
fd.write(ref_target) | fd.write(ref_target) | ||||
def write_archive(self): | def write_archive(self): | ||||
with tarfile.TarFile(mode="w", fileobj=self.fileobj) as tf: | with tarfile.TarFile(mode="w", fileobj=self.fileobj) as tf: | ||||
tf.add(self.gitdir, arcname=f"{self.obj_swhid()}.git", recursive=True) | tf.add(self.gitdir, arcname=f"{self.swhid}.git", recursive=True) | ||||
def _obj_path(self, obj_id: Sha1Git): | def _obj_path(self, obj_id: Sha1Git): | ||||
return os.path.join(self.gitdir, self._obj_relative_path(obj_id)) | return os.path.join(self.gitdir, self._obj_relative_path(obj_id)) | ||||
def _obj_relative_path(self, obj_id: Sha1Git): | def _obj_relative_path(self, obj_id: Sha1Git): | ||||
obj_id_hex = hash_to_hex(obj_id) | obj_id_hex = hash_to_hex(obj_id) | ||||
directory = obj_id_hex[0:2] | directory = obj_id_hex[0:2] | ||||
filename = obj_id_hex[2:] | filename = obj_id_hex[2:] | ||||
▲ Show 20 Lines • Show All 329 Lines • Show Last 20 Lines |