Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
# Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import datetime | import datetime | ||||
from abc import ABCMeta, abstractmethod | from abc import ABCMeta, abstractmethod | ||||
from enum import Enum | from enum import Enum | ||||
from hashlib import sha256 | from hashlib import sha256 | ||||
from typing import Dict, List, Optional, TypeVar, Union | from typing import Dict, Optional, Tuple, TypeVar, Union | ||||
import attr | import attr | ||||
from attrs_strict import type_validator | from attrs_strict import type_validator | ||||
import dateutil.parser | import dateutil.parser | ||||
import iso8601 | import iso8601 | ||||
from .identifiers import ( | from .identifiers import ( | ||||
normalize_timestamp, | normalize_timestamp, | ||||
Show All 21 Lines | |||||
def dictify(value): | def dictify(value): | ||||
"Helper function used by BaseModel.to_dict()" | "Helper function used by BaseModel.to_dict()" | ||||
if isinstance(value, BaseModel): | if isinstance(value, BaseModel): | ||||
return value.to_dict() | return value.to_dict() | ||||
elif isinstance(value, Enum): | elif isinstance(value, Enum): | ||||
return value.value | return value.value | ||||
elif isinstance(value, dict): | elif isinstance(value, dict): | ||||
return {k: dictify(v) for k, v in value.items()} | return {k: dictify(v) for k, v in value.items()} | ||||
elif isinstance(value, list): | elif isinstance(value, tuple): | ||||
return [dictify(v) for v in value] | return tuple(dictify(v) for v in value) | ||||
else: | else: | ||||
return value | return value | ||||
ModelType = TypeVar("ModelType", bound="BaseModel") | ModelType = TypeVar("ModelType", bound="BaseModel") | ||||
class BaseModel: | class BaseModel: | ||||
▲ Show 20 Lines • Show All 356 Lines • ▼ Show 20 Lines | committer_date = attr.ib( | ||||
type=Optional[TimestampWithTimezone], validator=type_validator() | type=Optional[TimestampWithTimezone], validator=type_validator() | ||||
) | ) | ||||
type = attr.ib(type=RevisionType, validator=type_validator()) | type = attr.ib(type=RevisionType, validator=type_validator()) | ||||
directory = attr.ib(type=Sha1Git, validator=type_validator()) | directory = attr.ib(type=Sha1Git, validator=type_validator()) | ||||
synthetic = attr.ib(type=bool, validator=type_validator()) | synthetic = attr.ib(type=bool, validator=type_validator()) | ||||
metadata = attr.ib( | metadata = attr.ib( | ||||
type=Optional[Dict[str, object]], validator=type_validator(), default=None | type=Optional[Dict[str, object]], validator=type_validator(), default=None | ||||
) | ) | ||||
parents = attr.ib( | parents = attr.ib(type=Tuple[Sha1Git, ...], validator=type_validator(), default=()) | ||||
type=List[Sha1Git], validator=type_validator(), default=attr.Factory(list) | |||||
) | |||||
id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | ||||
@staticmethod | @staticmethod | ||||
def compute_hash(object_dict): | def compute_hash(object_dict): | ||||
return revision_identifier(object_dict) | return revision_identifier(object_dict) | ||||
@classmethod | @classmethod | ||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
d = d.copy() | d = d.copy() | ||||
date = d.pop("date") | date = d.pop("date") | ||||
if date: | if date: | ||||
date = TimestampWithTimezone.from_dict(date) | date = TimestampWithTimezone.from_dict(date) | ||||
committer_date = d.pop("committer_date") | committer_date = d.pop("committer_date") | ||||
if committer_date: | if committer_date: | ||||
committer_date = TimestampWithTimezone.from_dict(committer_date) | committer_date = TimestampWithTimezone.from_dict(committer_date) | ||||
return cls( | return cls( | ||||
author=Person.from_dict(d.pop("author")), | author=Person.from_dict(d.pop("author")), | ||||
committer=Person.from_dict(d.pop("committer")), | committer=Person.from_dict(d.pop("committer")), | ||||
date=date, | date=date, | ||||
committer_date=committer_date, | committer_date=committer_date, | ||||
type=RevisionType(d.pop("type")), | type=RevisionType(d.pop("type")), | ||||
parents=tuple(d.pop("parents")), # for BW compat | |||||
**d, | **d, | ||||
) | ) | ||||
def anonymize(self) -> "Revision": | def anonymize(self) -> "Revision": | ||||
"""Returns an anonymized version of the Revision object. | """Returns an anonymized version of the Revision object. | ||||
Anonymization consists in replacing the author and committer with an anonymized | Anonymization consists in replacing the author and committer with an anonymized | ||||
Person object. | Person object. | ||||
Show All 9 Lines | class DirectoryEntry(BaseModel): | ||||
type = attr.ib(type=str, validator=attr.validators.in_(["file", "dir", "rev"])) | type = attr.ib(type=str, validator=attr.validators.in_(["file", "dir", "rev"])) | ||||
target = attr.ib(type=Sha1Git, validator=type_validator()) | target = attr.ib(type=Sha1Git, validator=type_validator()) | ||||
perms = attr.ib(type=int, validator=type_validator()) | perms = attr.ib(type=int, validator=type_validator()) | ||||
"""Usually one of the values of `swh.model.from_disk.DentryPerms`.""" | """Usually one of the values of `swh.model.from_disk.DentryPerms`.""" | ||||
@attr.s(frozen=True) | @attr.s(frozen=True) | ||||
class Directory(BaseModel, HashableObject): | class Directory(BaseModel, HashableObject): | ||||
entries = attr.ib(type=List[DirectoryEntry], validator=type_validator()) | entries = attr.ib(type=Tuple[DirectoryEntry, ...], validator=type_validator()) | ||||
id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | ||||
@staticmethod | @staticmethod | ||||
def compute_hash(object_dict): | def compute_hash(object_dict): | ||||
return directory_identifier(object_dict) | return directory_identifier(object_dict) | ||||
@classmethod | @classmethod | ||||
def from_dict(cls, d): | def from_dict(cls, d): | ||||
d = d.copy() | d = d.copy() | ||||
return cls( | return cls( | ||||
entries=[DirectoryEntry.from_dict(entry) for entry in d.pop("entries")], **d | entries=tuple( | ||||
DirectoryEntry.from_dict(entry) for entry in d.pop("entries") | |||||
), | |||||
**d, | |||||
) | ) | ||||
@attr.s(frozen=True) | @attr.s(frozen=True) | ||||
class BaseContent(BaseModel): | class BaseContent(BaseModel): | ||||
status = attr.ib( | status = attr.ib( | ||||
type=str, validator=attr.validators.in_(["visible", "hidden", "absent"]) | type=str, validator=attr.validators.in_(["visible", "hidden", "absent"]) | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 162 Lines • Show Last 20 Lines |