Changeset View
Changeset View
Standalone View
Standalone View
swh/model/from_disk.py
Show All 12 Lines | |||||
from typing import Any, Iterable, Iterator, List, Optional, Pattern, Tuple | from typing import Any, Iterable, Iterator, List, Optional, Pattern, Tuple | ||||
import attr | import attr | ||||
from attrs_strict import type_validator | from attrs_strict import type_validator | ||||
from typing_extensions import Final | from typing_extensions import Final | ||||
from . import model | from . import model | ||||
from .exceptions import InvalidDirectoryPath | from .exceptions import InvalidDirectoryPath | ||||
from .hashutil import MultiHash | from .hashutil import MultiHash, hash_to_bytes, hash_to_hex | ||||
from .identifiers import ( | from .identifiers import ( | ||||
CoreSWHID, | CoreSWHID, | ||||
ObjectType, | ObjectType, | ||||
directory_entry_sort_key, | directory_entry_sort_key, | ||||
directory_identifier, | directory_identifier, | ||||
) | ) | ||||
from .identifiers import identifier_to_bytes as id_to_bytes | |||||
from .identifiers import identifier_to_str as id_to_str | |||||
from .merkle import MerkleLeaf, MerkleNode | from .merkle import MerkleLeaf, MerkleNode | ||||
@attr.s(frozen=True, slots=True) | @attr.s(frozen=True, slots=True) | ||||
class DiskBackedContent(model.BaseContent): | class DiskBackedContent(model.BaseContent): | ||||
"""Content-like class, which allows lazy-loading data from the disk.""" | """Content-like class, which allows lazy-loading data from the disk.""" | ||||
object_type: Final = "content_file" | object_type: Final = "content_file" | ||||
▲ Show 20 Lines • Show All 175 Lines • ▼ Show 20 Lines | def from_file(cls, *, path, max_content_length=None): | ||||
return obj | return obj | ||||
def swhid(self) -> CoreSWHID: | def swhid(self) -> CoreSWHID: | ||||
"""Return node identifier as a SWHID | """Return node identifier as a SWHID | ||||
""" | """ | ||||
return CoreSWHID(object_type=ObjectType.CONTENT, object_id=self.hash) | return CoreSWHID(object_type=ObjectType.CONTENT, object_id=self.hash) | ||||
def __repr__(self): | def __repr__(self): | ||||
return "Content(id=%s)" % id_to_str(self.hash) | return "Content(id=%s)" % hash_to_hex(self.hash) | ||||
def compute_hash(self): | def compute_hash(self): | ||||
return self.data["sha1_git"] | return self.data["sha1_git"] | ||||
def to_model(self) -> model.BaseContent: | def to_model(self) -> model.BaseContent: | ||||
"""Builds a `model.BaseContent` object based on this leaf.""" | """Builds a `model.BaseContent` object based on this leaf.""" | ||||
data = self.get_data().copy() | data = self.get_data().copy() | ||||
data.pop("perms", None) | data.pop("perms", None) | ||||
▲ Show 20 Lines • Show All 263 Lines • ▼ Show 20 Lines | def entries(self): | ||||
return self.__entries | return self.__entries | ||||
def swhid(self) -> CoreSWHID: | def swhid(self) -> CoreSWHID: | ||||
"""Return node identifier as a SWHID | """Return node identifier as a SWHID | ||||
""" | """ | ||||
return CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=self.hash) | return CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=self.hash) | ||||
def compute_hash(self): | def compute_hash(self): | ||||
return id_to_bytes(directory_identifier({"entries": self.entries})) | return hash_to_bytes(directory_identifier({"entries": self.entries})) | ||||
def to_model(self) -> model.Directory: | def to_model(self) -> model.Directory: | ||||
"""Builds a `model.Directory` object based on this node; | """Builds a `model.Directory` object based on this node; | ||||
ignoring its children.""" | ignoring its children.""" | ||||
return model.Directory.from_dict(self.get_data()) | return model.Directory.from_dict(self.get_data()) | ||||
def __getitem__(self, key): | def __getitem__(self, key): | ||||
if not isinstance(key, bytes): | if not isinstance(key, bytes): | ||||
Show All 35 Lines | def __delitem__(self, key): | ||||
if b"/" not in key: | if b"/" not in key: | ||||
super().__delitem__(key) | super().__delitem__(key) | ||||
else: | else: | ||||
key1, key2 = key.rsplit(b"/", 1) | key1, key2 = key.rsplit(b"/", 1) | ||||
del self[key1][key2] | del self[key1][key2] | ||||
def __repr__(self): | def __repr__(self): | ||||
return "Directory(id=%s, entries=[%s])" % ( | return "Directory(id=%s, entries=[%s])" % ( | ||||
id_to_str(self.hash), | hash_to_hex(self.hash), | ||||
", ".join(str(entry) for entry in self), | ", ".join(str(entry) for entry in self), | ||||
) | ) |