Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
Show First 20 Lines • Show All 548 Lines • ▼ Show 20 Lines | class BaseContent(BaseModel): | ||||
def _hash_data(data: bytes): | def _hash_data(data: bytes): | ||||
"""Hash some data, returning most of the fields of a content object""" | """Hash some data, returning most of the fields of a content object""" | ||||
d = MultiHash.from_data(data).digest() | d = MultiHash.from_data(data).digest() | ||||
d['data'] = data | d['data'] = data | ||||
d['length'] = len(data) | d['length'] = len(data) | ||||
return d | return d | ||||
def to_dict(self): | |||||
content = super().to_dict() | |||||
if content['ctime'] is None: | |||||
del content['ctime'] | |||||
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. | ||||
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) | ||||
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 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': | def from_data(cls, data, status='visible', ctime=None) -> 'Content': | ||||
"""Generate a Content from a given `data` byte string. | """Generate a Content from a given `data` byte string. | ||||
This populates the Content with the hashes and length for the data | This populates the Content with the hashes and length for the data | ||||
passed as argument, as well as the data itself. | passed as argument, as well as the data itself. | ||||
""" | """ | ||||
d = cls._hash_data(data) | d = cls._hash_data(data) | ||||
d['status'] = status | d['status'] = status | ||||
d['ctime'] = ctime | |||||
return cls(**d) | return cls(**d) | ||||
@classmethod | @classmethod | ||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
if isinstance(d.get('ctime'), str): | |||||
d = d.copy() | |||||
ardumont: Apparently missing the associated test case (thank the tooling for that one ;) | |||||
d['ctime'] = dateutil.parser.parse(d['ctime']) | |||||
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 | ||||
to lazy-load data (eg. from disk or objstorage).""" | to lazy-load data (eg. from disk or objstorage).""" | ||||
▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 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': | def from_data( | ||||
cls, | |||||
data: bytes, | |||||
reason: str, | |||||
ctime: Optional[datetime.datetime] = None) -> 'SkippedContent': | |||||
"""Generate a SkippedContent from a given `data` byte string. | """Generate a SkippedContent from a given `data` byte string. | ||||
This populates the SkippedContent with the hashes and length for the | This populates the SkippedContent with the hashes and length for the | ||||
data passed as argument. | data passed as argument. | ||||
You can use `attr.evolve` on such a generated content to nullify some | You can use `attr.evolve` on such a generated content to nullify some | ||||
of its attributes, e.g. for tests. | of its attributes, e.g. for tests. | ||||
""" | """ | ||||
d = cls._hash_data(data) | d = cls._hash_data(data) | ||||
del d['data'] | del d['data'] | ||||
d['status'] = 'absent' | d['status'] = 'absent' | ||||
d['reason'] = reason | d['reason'] = reason | ||||
d['ctime'] = ctime | |||||
return cls(**d) | return cls(**d) | ||||
@classmethod | @classmethod | ||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
d2 = d.copy() | d2 = d.copy() | ||||
if d2.pop('data', None) is not None: | if d2.pop('data', None) is not None: | ||||
raise ValueError('SkippedContent has no "data" attribute %r' % d) | raise ValueError('SkippedContent has no "data" attribute %r' % d) | ||||
return super().from_dict(d2, use_subclass=False) | return super().from_dict(d2, use_subclass=False) |
Apparently missing the associated test case (thank the tooling for that one ;)