Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
Show First 20 Lines • Show All 198 Lines • ▼ Show 20 Lines | def check(self) -> None: | ||||
"""Performs internal consistency checks, and raises an error if one fails.""" | """Performs internal consistency checks, and raises an error if one fails.""" | ||||
attr.validate(self) | attr.validate(self) | ||||
def _compute_hash_from_manifest(manifest: bytes) -> Sha1Git: | def _compute_hash_from_manifest(manifest: bytes) -> Sha1Git: | ||||
return hashlib.new("sha1", manifest).digest() | return hashlib.new("sha1", manifest).digest() | ||||
class HashableObject(metaclass=ABCMeta): | class BaseHashableModel(BaseModel, metaclass=ABCMeta): | ||||
"""Mixin to automatically compute object identifier hash when | """Mixin to automatically compute object identifier hash when | ||||
the associated model is instantiated.""" | the associated model is instantiated.""" | ||||
__slots__ = () | __slots__ = () | ||||
id: Sha1Git | id: Sha1Git | ||||
def compute_hash(self) -> bytes: | def compute_hash(self) -> bytes: | ||||
Show All 13 Lines | def __attrs_post_init__(self): | ||||
if not self.id: | if not self.id: | ||||
obj_id = self.compute_hash() | obj_id = self.compute_hash() | ||||
object.__setattr__(self, "id", obj_id) | object.__setattr__(self, "id", obj_id) | ||||
def unique_key(self) -> KeyType: | def unique_key(self) -> KeyType: | ||||
return self.id | return self.id | ||||
def check(self) -> None: | def check(self) -> None: | ||||
super().check() # type: ignore | super().check() | ||||
if self.id != self.compute_hash(): | if self.id != self.compute_hash(): | ||||
raise ValueError("'id' does not match recomputed hash.") | raise ValueError("'id' does not match recomputed hash.") | ||||
class HashableObjectWithManifest(HashableObject): | class HashableObjectWithManifest(BaseHashableModel): | ||||
"""Derived class of HashableObject, for objects that may need to store | """Derived class of BaseHashableModel, for objects that may need to store | ||||
verbatim git objects as ``raw_manifest`` to preserve original hashes.""" | verbatim git objects as ``raw_manifest`` to preserve original hashes.""" | ||||
__slots__ = () | __slots__ = () | ||||
raw_manifest: Optional[bytes] = None | raw_manifest: Optional[bytes] = None | ||||
"""Stores the original content of git objects when they cannot be faithfully | """Stores the original content of git objects when they cannot be faithfully | ||||
represented using only the other attributes. | represented using only the other attributes. | ||||
▲ Show 20 Lines • Show All 327 Lines • ▼ Show 20 Lines | def offset_minutes(self): | ||||
... Timestamp(seconds=1642765364, microseconds=0), offset_bytes=b"+0530" | ... Timestamp(seconds=1642765364, microseconds=0), offset_bytes=b"+0530" | ||||
... ).offset_minutes() | ... ).offset_minutes() | ||||
330 | 330 | ||||
""" | """ | ||||
return self._parse_offset_bytes(self.offset_bytes) | return self._parse_offset_bytes(self.offset_bytes) | ||||
@attr.s(frozen=True, slots=True) | @attr.s(frozen=True, slots=True) | ||||
class Origin(HashableObject, BaseModel): | class Origin(BaseHashableModel): | ||||
"""Represents a software source: a VCS and an URL.""" | """Represents a software source: a VCS and an URL.""" | ||||
object_type: Final = "origin" | object_type: Final = "origin" | ||||
url = attr.ib(type=str, validator=type_validator()) | url = attr.ib(type=str, validator=type_validator()) | ||||
id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | ||||
▲ Show 20 Lines • Show All 126 Lines • ▼ Show 20 Lines | def check_target(self, attribute, value): | ||||
raise ValueError("Wrong length for bytes identifier: %d" % len(value)) | raise ValueError("Wrong length for bytes identifier: %d" % len(value)) | ||||
@classmethod | @classmethod | ||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
return cls(target=d["target"], target_type=TargetType(d["target_type"])) | return cls(target=d["target"], target_type=TargetType(d["target_type"])) | ||||
@attr.s(frozen=True, slots=True) | @attr.s(frozen=True, slots=True) | ||||
class Snapshot(HashableObject, BaseModel): | class Snapshot(BaseHashableModel): | ||||
"""Represents the full state of an origin at a given point in time.""" | """Represents the full state of an origin at a given point in time.""" | ||||
object_type: Final = "snapshot" | object_type: Final = "snapshot" | ||||
branches = attr.ib( | branches = attr.ib( | ||||
type=ImmutableDict[bytes, Optional[SnapshotBranch]], | type=ImmutableDict[bytes, Optional[SnapshotBranch]], | ||||
validator=type_validator(), | validator=type_validator(), | ||||
converter=freeze_optional_dict, | converter=freeze_optional_dict, | ||||
▲ Show 20 Lines • Show All 621 Lines • ▼ Show 20 Lines | def normalize_discovery_date(value: Any) -> datetime.datetime: | ||||
if value.tzinfo is None: | if value.tzinfo is None: | ||||
raise ValueError("discovery_date must be a timezone-aware datetime.") | raise ValueError("discovery_date must be a timezone-aware datetime.") | ||||
# Normalize timezone to utc, and truncate microseconds to 0 | # Normalize timezone to utc, and truncate microseconds to 0 | ||||
return value.astimezone(datetime.timezone.utc).replace(microsecond=0) | return value.astimezone(datetime.timezone.utc).replace(microsecond=0) | ||||
@attr.s(frozen=True, slots=True) | @attr.s(frozen=True, slots=True) | ||||
class RawExtrinsicMetadata(HashableObject, BaseModel): | class RawExtrinsicMetadata(BaseHashableModel): | ||||
object_type: Final = "raw_extrinsic_metadata" | object_type: Final = "raw_extrinsic_metadata" | ||||
# target object | # target object | ||||
target = attr.ib(type=ExtendedSWHID, validator=type_validator()) | target = attr.ib(type=ExtendedSWHID, validator=type_validator()) | ||||
# source | # source | ||||
discovery_date = attr.ib(type=datetime.datetime, converter=normalize_discovery_date) | discovery_date = attr.ib(type=datetime.datetime, converter=normalize_discovery_date) | ||||
authority = attr.ib(type=MetadataAuthority, validator=type_validator()) | authority = attr.ib(type=MetadataAuthority, validator=type_validator()) | ||||
▲ Show 20 Lines • Show All 200 Lines • ▼ Show 20 Lines | def swhid(self) -> ExtendedSWHID: | ||||
"""Returns a SWHID representing this RawExtrinsicMetadata object.""" | """Returns a SWHID representing this RawExtrinsicMetadata object.""" | ||||
return ExtendedSWHID( | return ExtendedSWHID( | ||||
object_type=SwhidExtendedObjectType.RAW_EXTRINSIC_METADATA, | object_type=SwhidExtendedObjectType.RAW_EXTRINSIC_METADATA, | ||||
object_id=self.id, | object_id=self.id, | ||||
) | ) | ||||
@attr.s(frozen=True, slots=True) | @attr.s(frozen=True, slots=True) | ||||
class ExtID(HashableObject, BaseModel): | class ExtID(BaseHashableModel): | ||||
object_type: Final = "extid" | object_type: Final = "extid" | ||||
extid_type = attr.ib(type=str, validator=type_validator()) | extid_type = attr.ib(type=str, validator=type_validator()) | ||||
extid = attr.ib(type=bytes, validator=type_validator()) | extid = attr.ib(type=bytes, validator=type_validator()) | ||||
target = attr.ib(type=CoreSWHID, validator=type_validator()) | target = attr.ib(type=CoreSWHID, validator=type_validator()) | ||||
extid_version = attr.ib(type=int, validator=type_validator(), default=0) | extid_version = attr.ib(type=int, validator=type_validator(), default=0) | ||||
id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"", repr=hash_repr) | id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"", repr=hash_repr) | ||||
Show All 40 Lines |