Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
Show All 11 Lines | |||||
import attr | import attr | ||||
import dateutil.parser | import dateutil.parser | ||||
import iso8601 | import iso8601 | ||||
from .identifiers import ( | from .identifiers import ( | ||||
normalize_timestamp, directory_identifier, revision_identifier, | normalize_timestamp, directory_identifier, revision_identifier, | ||||
release_identifier, snapshot_identifier | release_identifier, snapshot_identifier | ||||
) | ) | ||||
from .hashutil import DEFAULT_ALGORITHMS, hash_to_bytes | from .hashutil import DEFAULT_ALGORITHMS, hash_to_bytes, MultiHash | ||||
class MissingData(Exception): | class MissingData(Exception): | ||||
"""Raised by `Content.with_data` when it has no way of fetching the | """Raised by `Content.with_data` when it has no way of fetching the | ||||
data (but not when fetching the data fails).""" | data (but not when fetching the data fails).""" | ||||
pass | pass | ||||
▲ Show 20 Lines • Show All 355 Lines • ▼ Show 20 Lines | def from_dict(cls, d): | ||||
**d) | **d) | ||||
@attr.s(frozen=True) | @attr.s(frozen=True) | ||||
class BaseContent(BaseModel): | class BaseContent(BaseModel): | ||||
status = attr.ib( | status = attr.ib( | ||||
type=str, | type=str, | ||||
validator=attr.validators.in_(['visible', 'hidden', 'absent'])) | validator=attr.validators.in_(['visible', 'hidden', 'absent'])) | ||||
@staticmethod | |||||
def _hash_data(data: bytes): | |||||
"""Hash some data, returning most of the fields of a content object""" | |||||
d = MultiHash.from_data(data).digest() | |||||
d['data'] = data | |||||
d['length'] = len(data) | |||||
vlorentz: Maybe prepend an underscore? There's not reason for it to be used by other modules. | |||||
return d | |||||
def to_dict(self): | def to_dict(self): | ||||
content = super().to_dict() | content = super().to_dict() | ||||
if content['ctime'] is None: | if content['ctime'] is None: | ||||
del content['ctime'] | del content['ctime'] | ||||
return content | return content | ||||
@classmethod | @classmethod | ||||
def from_dict(cls, d, use_subclass=True): | def from_dict(cls, d, use_subclass=True): | ||||
if use_subclass: | if use_subclass: | ||||
# Chooses a subclass to instantiate instead. | # Chooses a subclass to instantiate instead. | ||||
Done Inline Actionsreason should default to None, and the function error if it's not set. vlorentz: `reason` should default to None, and the function error if it's not set. | |||||
if d['status'] == 'absent': | if d['status'] == 'absent': | ||||
return SkippedContent.from_dict(d) | return SkippedContent.from_dict(d) | ||||
else: | else: | ||||
return Content.from_dict(d) | return Content.from_dict(d) | ||||
else: | else: | ||||
return super().from_dict(d) | return super().from_dict(d) | ||||
def get_hash(self, hash_name): | def get_hash(self, hash_name): | ||||
Show All 33 Lines | class Content(BaseContent): | ||||
def to_dict(self): | def to_dict(self): | ||||
content = super().to_dict() | content = super().to_dict() | ||||
if content['data'] is None: | if content['data'] is None: | ||||
del content['data'] | del content['data'] | ||||
return content | return content | ||||
@classmethod | @classmethod | ||||
def from_data(cls, data, status='visible') -> 'Content': | |||||
"""Generate a Content from a given `data` byte string. | |||||
This populates the Content with the hashes and length for the data | |||||
passed as argument, as well as the data itself. | |||||
""" | |||||
d = cls._hash_data(data) | |||||
d['status'] = status | |||||
return cls(**d) | |||||
@classmethod | |||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
return super().from_dict(d, use_subclass=False) | return super().from_dict(d, use_subclass=False) | ||||
def with_data(self) -> 'Content': | def with_data(self) -> 'Content': | ||||
"""Loads the `data` attribute; meaning that it is guaranteed not to | """Loads the `data` attribute; meaning that it is guaranteed not to | ||||
be None after this call. | be None after this call. | ||||
This call is almost a no-op, but subclasses may overload this method | This call is almost a no-op, but subclasses may overload this method | ||||
Show All 39 Lines | class SkippedContent(BaseContent): | ||||
def to_dict(self): | def to_dict(self): | ||||
content = super().to_dict() | content = super().to_dict() | ||||
if content['origin'] is None: | if content['origin'] is None: | ||||
del content['origin'] | del content['origin'] | ||||
return content | return content | ||||
@classmethod | @classmethod | ||||
def from_data(cls, data, reason: str) -> 'SkippedContent': | |||||
"""Generate a SkippedContent from a given `data` byte string. | |||||
This populates the SkippedContent with the hashes and length for the | |||||
data passed as argument. | |||||
You can use `attr.evolve` on such a generated content to nullify some | |||||
of its attributes, e.g. for tests. | |||||
""" | |||||
d = cls._hash_data(data) | |||||
del d['data'] | |||||
d['status'] = 'absent' | |||||
d['reason'] = reason | |||||
return cls(**d) | |||||
@classmethod | |||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
d2 = d | d2 = d | ||||
d = d.copy() | d = d.copy() | ||||
if d.pop('data', None) is not None: | if d.pop('data', None) is not None: | ||||
raise ValueError('SkippedContent has no "data" attribute %r' % d2) | raise ValueError('SkippedContent has no "data" attribute %r' % d2) | ||||
return super().from_dict(d, use_subclass=False) | return super().from_dict(d, use_subclass=False) |
Maybe prepend an underscore? There's not reason for it to be used by other modules.