Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
# Copyright (C) 2018-2020 The Software Heritage developers | # Copyright (C) 2018-2020 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 copy import deepcopy | |||||
from enum import Enum | from enum import Enum | ||||
from hashlib import sha256 | from hashlib import sha256 | ||||
from typing import Dict, Optional, Tuple, TypeVar, Union | from typing import Dict, Iterable, Optional, Tuple, TypeVar, Union | ||||
from typing_extensions import Final | from typing_extensions import Final | ||||
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 ( | ||||
▲ Show 20 Lines • Show All 385 Lines • ▼ Show 20 Lines | |||||
class RevisionType(Enum): | class RevisionType(Enum): | ||||
GIT = "git" | GIT = "git" | ||||
TAR = "tar" | TAR = "tar" | ||||
DSC = "dsc" | DSC = "dsc" | ||||
SUBVERSION = "svn" | SUBVERSION = "svn" | ||||
MERCURIAL = "hg" | MERCURIAL = "hg" | ||||
def tuplify_extra_headers(value: Iterable) -> Tuple: | |||||
return tuple((k, v) for k, v in value) | |||||
@attr.s(frozen=True) | @attr.s(frozen=True) | ||||
class Revision(BaseModel, HashableObject): | class Revision(BaseModel, HashableObject): | ||||
object_type: Final = "revision" | object_type: Final = "revision" | ||||
message = attr.ib(type=Optional[bytes], validator=type_validator()) | message = attr.ib(type=Optional[bytes], validator=type_validator()) | ||||
author = attr.ib(type=Person, validator=type_validator()) | author = attr.ib(type=Person, validator=type_validator()) | ||||
committer = attr.ib(type=Person, validator=type_validator()) | committer = attr.ib(type=Person, validator=type_validator()) | ||||
date = attr.ib(type=Optional[TimestampWithTimezone], validator=type_validator()) | date = attr.ib(type=Optional[TimestampWithTimezone], validator=type_validator()) | ||||
committer_date = attr.ib( | 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(type=Tuple[Sha1Git, ...], validator=type_validator(), default=()) | parents = attr.ib(type=Tuple[Sha1Git, ...], validator=type_validator(), default=()) | ||||
id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | id = attr.ib(type=Sha1Git, validator=type_validator(), default=b"") | ||||
extra_headers = attr.ib( | |||||
type=Tuple[Tuple[bytes, bytes], ...], # but it makes mypy sad | |||||
validator=type_validator(), | |||||
converter=tuplify_extra_headers, # type: ignore | |||||
default=(), | |||||
) | |||||
def __attrs_post_init__(self): | |||||
super().__attrs_post_init__() | |||||
# ensure metadata is a deep copy of whatever was given, and if needed | |||||
# extract extra_headers from there | |||||
if self.metadata: | |||||
metadata = deepcopy(self.metadata) | |||||
if not self.extra_headers and "extra_headers" in metadata: | |||||
object.__setattr__( | |||||
self, | |||||
"extra_headers", | |||||
tuplify_extra_headers(metadata.pop("extra_headers")), | |||||
) | |||||
attr.validate(self) | |||||
object.__setattr__(self, "metadata", metadata) | |||||
@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() | ||||
▲ Show 20 Lines • Show All 233 Lines • Show Last 20 Lines |