Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
Show All 12 Lines | |||||
import dateutil.parser | import dateutil.parser | ||||
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 | ||||
class MissingData(Exception): | |||||
"""Raised by `Content.with_data` when it has no way of fetching the | |||||
data (but not when fetching the data fails).""" | |||||
pass | |||||
SHA1_SIZE = 20 | SHA1_SIZE = 20 | ||||
# TODO: Limit this to 20 bytes | # TODO: Limit this to 20 bytes | ||||
Sha1Git = bytes | Sha1Git = bytes | ||||
class BaseModel: | class BaseModel: | ||||
"""Base class for SWH model classes. | """Base class for SWH model classes. | ||||
▲ Show 20 Lines • Show All 328 Lines • ▼ Show 20 Lines | def from_dict(cls, d): | ||||
return cls( | return cls( | ||||
entries=[DirectoryEntry.from_dict(entry) | entries=[DirectoryEntry.from_dict(entry) | ||||
for entry in d.pop('entries')], | for entry in d.pop('entries')], | ||||
**d) | **d) | ||||
@attr.s(frozen=True) | @attr.s(frozen=True) | ||||
class BaseContent(BaseModel): | class BaseContent(BaseModel): | ||||
status = attr.ib( | |||||
type=str, | |||||
validator=attr.validators.in_(['visible', 'hidden', 'absent'])) | |||||
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): | ||||
Show All 24 Lines | class Content(BaseContent): | ||||
blake2s256 = attr.ib(type=bytes) | blake2s256 = attr.ib(type=bytes) | ||||
length = attr.ib(type=int) | length = attr.ib(type=int) | ||||
status = attr.ib( | status = attr.ib( | ||||
type=str, | type=str, | ||||
default='visible', | default='visible', | ||||
validator=attr.validators.in_(['visible', 'hidden'])) | validator=attr.validators.in_(['visible', 'hidden'])) | ||||
data = attr.ib(type=Optional[bytes], | |||||
default=None) | data = attr.ib(type=Optional[bytes], default=None) | ||||
ctime = attr.ib(type=Optional[datetime.datetime], | ctime = attr.ib(type=Optional[datetime.datetime], | ||||
default=None) | default=None) | ||||
@length.validator | @length.validator | ||||
def check_length(self, attribute, value): | def check_length(self, attribute, value): | ||||
"""Checks the length is positive.""" | """Checks the length is positive.""" | ||||
if value < 0: | if value < 0: | ||||
raise ValueError('Length must be positive.') | raise ValueError('Length must be positive.') | ||||
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_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': | |||||
"""Loads the `data` attribute; meaning that it is guaranteed not to | |||||
be None after this call. | |||||
This call is almost a no-op, but subclasses may overload this method | |||||
to lazy-load data (eg. from disk or objstorage).""" | |||||
if self.data is None: | |||||
raise MissingData('Content data is None.') | |||||
return self | |||||
@attr.s(frozen=True) | @attr.s(frozen=True) | ||||
class SkippedContent(BaseContent): | class SkippedContent(BaseContent): | ||||
sha1 = attr.ib(type=Optional[bytes]) | sha1 = attr.ib(type=Optional[bytes]) | ||||
sha1_git = attr.ib(type=Optional[Sha1Git]) | sha1_git = attr.ib(type=Optional[Sha1Git]) | ||||
sha256 = attr.ib(type=Optional[bytes]) | sha256 = attr.ib(type=Optional[bytes]) | ||||
blake2s256 = attr.ib(type=Optional[bytes]) | blake2s256 = attr.ib(type=Optional[bytes]) | ||||
length = attr.ib(type=int) | length = attr.ib(type=Optional[int]) | ||||
status = attr.ib( | status = attr.ib( | ||||
type=str, | type=str, | ||||
validator=attr.validators.in_(['absent'])) | validator=attr.validators.in_(['absent'])) | ||||
reason = attr.ib(type=Optional[str], | reason = attr.ib(type=Optional[str], | ||||
default=None) | default=None) | ||||
origin = attr.ib(type=Optional[Origin], | origin = attr.ib(type=Optional[Origin], | ||||
Show All 31 Lines |