Skip to content

repository_orm.model

Store the common business model of all entities.

Entity

Bases: BaseModel

Model of any object no defined by it's attributes whom instead has an identity.

Unlike value objects, they have identity equality. We can change their values, and they are still recognizably the same thing.

An entity with a negative id means that the id needs to be set by the repository.

The _defined_values are used to know which attributes were set by the user at the time of merging objects.

Source code in repository_orm/model.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class Entity(BaseModel):
    """Model of any object no defined by it's attributes whom instead has an identity.

    Unlike value objects, they have *identity equality*. We can change their values, and
    they are still recognizably the same thing.

    An entity with a negative id means that the id needs to be set by the repository.

    The _defined_values are used to know which attributes were set by the user at the
    time of merging objects.
    """

    id_: EntityID = -1
    _defined_values: Dict[str, Any] = PrivateAttr()
    _skip_on_merge: List[str] = []

    # ANN401: Any not allowed, but it's what we have.
    def __init__(self, **data: Any) -> None:  # noqa: ANN401
        """Initialize the defined values."""
        super().__init__(**data)
        self._defined_values = data

    def __lt__(self, other: "Entity") -> bool:
        """Assert if an object is smaller than us.

        Args:
            other: Entity to compare.
        """
        if isinstance(other.id_, int) and isinstance(self.id_, int):
            return self.id_ < other.id_
        return str(self.id_) < str(other.id_)

    def __gt__(self, other: "Entity") -> bool:
        """Assert if an object is greater than us.

        Args:
            other: Entity to compare.
        """
        if isinstance(other.id_, int) and isinstance(self.id_, int):
            return self.id_ > other.id_
        return str(self.id_) > str(other.id_)

    def __hash__(self) -> int:
        """Create an unique hash of the class object."""
        return hash(f"{self.model_name}-{self.id_}")

    # ANN401: Any not allowed, but it's what we have.
    def __setattr__(self, attribute: str, value: Any) -> None:  # noqa: ANN401
        """Store the set attribute into the _defined_values."""
        if attribute != "_defined_values":
            self._defined_values[attribute] = value
        super().__setattr__(attribute, value)

    @property
    def model_name(self) -> str:
        """Return the entity model name."""
        return self.schema()["title"]

    def merge(self, other: "Entity") -> "Entity":
        """Update the attributes with the ones manually set by the user of other.

        If the other object has default values not set by the user, they won't be
        propagated to `self`.

        Args:
            other: Entity to compare.
        """
        if not isinstance(other, type(self)):
            raise ValueError(
                "Can't merge objects of different models "
                f"({self.model_name} with {other.model_name})."
            )
        if self.id_ != other.id_:
            raise ValueError(f"Can't merge two {self.model_name}s with different ids")

        # Merge objects
        # W0212: access to an internal property, but it's managed by us so there is
        # no problem on it.
        for attribute, value in other._defined_values.items():  # noqa: W0212
            if attribute not in self._skip_on_merge:
                setattr(self, attribute, value)

        return self

    @property
    def defined_values(self) -> Dict[str, Any]:
        """Return the entity defined values."""
        return self._defined_values

    def clear_defined_values(self) -> None:
        """Remove all references to defined values.

        I tried to return self so that it can be used chained with repo.get(), but I get
        a mypy error `Incompatible return value type (got "Entity", expected "Entity")`
        """
        self._defined_values = {}

__gt__(other)

Assert if an object is greater than us.

Parameters:

Name Type Description Default
other Entity

Entity to compare.

 required
Source code in repository_orm/model.py 
46
47
48
49
50
51
52
53
54
def __gt__(self, other: "Entity") -> bool:
    """Assert if an object is greater than us.

    Args:
        other: Entity to compare.
    """
    if isinstance(other.id_, int) and isinstance(self.id_, int):
        return self.id_ > other.id_
    return str(self.id_) > str(other.id_)

__hash__()

Create an unique hash of the class object.

Source code in repository_orm/model.py 
56
57
58
def __hash__(self) -> int:
    """Create an unique hash of the class object."""
    return hash(f"{self.model_name}-{self.id_}")

__init__(**data)

Initialize the defined values.

Source code in repository_orm/model.py 
31
32
33
34
def __init__(self, **data: Any) -> None:  # noqa: ANN401
    """Initialize the defined values."""
    super().__init__(**data)
    self._defined_values = data

__lt__(other)

Assert if an object is smaller than us.

Parameters:

Name Type Description Default
other Entity

Entity to compare.

 required
Source code in repository_orm/model.py 
36
37
38
39
40
41
42
43
44
def __lt__(self, other: "Entity") -> bool:
    """Assert if an object is smaller than us.

    Args:
        other: Entity to compare.
    """
    if isinstance(other.id_, int) and isinstance(self.id_, int):
        return self.id_ < other.id_
    return str(self.id_) < str(other.id_)

__setattr__(attribute, value)

Store the set attribute into the _defined_values.

Source code in repository_orm/model.py 
61
62
63
64
65
def __setattr__(self, attribute: str, value: Any) -> None:  # noqa: ANN401
    """Store the set attribute into the _defined_values."""
    if attribute != "_defined_values":
        self._defined_values[attribute] = value
    super().__setattr__(attribute, value)

clear_defined_values()

Remove all references to defined values.

I tried to return self so that it can be used chained with repo.get(), but I get a mypy error Incompatible return value type (got "Entity", expected "Entity")

Source code in repository_orm/model.py 
103
104
105
106
107
108
109
def clear_defined_values(self) -> None:
    """Remove all references to defined values.

    I tried to return self so that it can be used chained with repo.get(), but I get
    a mypy error `Incompatible return value type (got "Entity", expected "Entity")`
    """
    self._defined_values = {}

defined_values() property

Return the entity defined values.

Source code in repository_orm/model.py 
 98
 99
100
101
@property
def defined_values(self) -> Dict[str, Any]:
    """Return the entity defined values."""
    return self._defined_values

merge(other)

Update the attributes with the ones manually set by the user of other.

If the other object has default values not set by the user, they won't be propagated to self.

Parameters:

Name Type Description Default
other Entity

Entity to compare.

 required
Source code in repository_orm/model.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def merge(self, other: "Entity") -> "Entity":
    """Update the attributes with the ones manually set by the user of other.

    If the other object has default values not set by the user, they won't be
    propagated to `self`.

    Args:
        other: Entity to compare.
    """
    if not isinstance(other, type(self)):
        raise ValueError(
            "Can't merge objects of different models "
            f"({self.model_name} with {other.model_name})."
        )
    if self.id_ != other.id_:
        raise ValueError(f"Can't merge two {self.model_name}s with different ids")

    # Merge objects
    # W0212: access to an internal property, but it's managed by us so there is
    # no problem on it.
    for attribute, value in other._defined_values.items():  # noqa: W0212
        if attribute not in self._skip_on_merge:
            setattr(self, attribute, value)

    return self

model_name() property

Return the entity model name.

Source code in repository_orm/model.py 
67
68
69
70
@property
def model_name(self) -> str:
    """Return the entity model name."""
    return self.schema()["title"]

File

Bases: Entity, Generic[AnyStr]

Model a computer file.

Source code in repository_orm/model.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class File(Entity, Generic[AnyStr]):
    """Model a computer file."""

    path: str
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    owner: Optional[str] = None
    group: Optional[str] = None
    permissions: Optional[str] = None

    # The use of a private attribute and the impossibility of loading the content
    # at object creation will be fixed on Pydantic 1.9.
    # We will be able to define the excluded attribute content in the Config of the
    # model.
    #
    # For more information on how to improve this code, read this:
    # https://lyz-code.github.io/blue-book/coding/python/pydantic/#define-fields-to-exclude-from-exporting-at-config-level # noqa:E501
    _content: Optional[AnyStr] = PrivateAttr(None)
    # If the content is of type bytes
    is_bytes: bool = False

    @property
    def basename(self) -> str:
        """Return the name of the file."""
        return os.path.basename(self.path)

    @property
    def dirname(self) -> str:
        """Return the name of the file."""
        return os.path.dirname(self.path)

    @property
    def extension(self) -> str:
        """Return the name of the file."""
        return self.basename.split(".")[-1]

    @property
    def content(self) -> AnyStr:
        """Return the content of the file.

        Returns:
            The content of the file.

        Raises:
            FileContentNotLoadedError: if the content is not yet loaded.
        """
        if self._content is None:
            raise FileContentNotLoadedError(
                "The content of the file has not been loaded yet."
            )
        return self._content

basename() property

Return the name of the file.

Source code in repository_orm/model.py 
137
138
139
140
@property
def basename(self) -> str:
    """Return the name of the file."""
    return os.path.basename(self.path)

content() property

Return the content of the file.

Returns:

Type Description
AnyStr

The content of the file.

Raises:

Type Description
FileContentNotLoadedError

if the content is not yet loaded.
Source code in repository_orm/model.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@property
def content(self) -> AnyStr:
    """Return the content of the file.

    Returns:
        The content of the file.

    Raises:
        FileContentNotLoadedError: if the content is not yet loaded.
    """
    if self._content is None:
        raise FileContentNotLoadedError(
            "The content of the file has not been loaded yet."
        )
    return self._content

dirname() property

Return the name of the file.

Source code in repository_orm/model.py 
142
143
144
145
@property
def dirname(self) -> str:
    """Return the name of the file."""
    return os.path.dirname(self.path)

extension() property

Return the name of the file.

Source code in repository_orm/model.py 
147
148
149
150
@property
def extension(self) -> str:
    """Return the name of the file."""
    return self.basename.split(".")[-1]
Last update: 2021-04-08