Skip to content

relation

Relation

Bases: Generic[T]

Keeps related Models and handles adding/removing of the children.

Source code in ormar/relations/relation.py
 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
110
111
112
113
114
115
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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class Relation(Generic[T]):
    """
    Keeps related Models and handles adding/removing of the children.
    """

    def __init__(
        self,
        manager: "RelationsManager",
        type_: RelationType,
        field_name: str,
        to: Type["T"],
        through: Type["Model"] = None,
    ) -> None:
        """
        Initialize the Relation and keep the related models either as instances of
        passed Model, or as a RelationProxy which is basically a list of models with
        some special behavior, as it exposes QuerySetProxy and allows querying the
        related models already pre filtered by parent model.

        :param manager: reference to relation manager
        :type manager: RelationsManager
        :param type_: type of the relation
        :type type_: RelationType
        :param field_name: name of the relation field
        :type field_name: str
        :param to: model to which relation leads to
        :type to: Type[Model]
        :param through: model through which relation goes for m2m relations
        :type through: Type[Model]
        """
        self.manager = manager
        self._owner: "Model" = manager.owner
        self._type: RelationType = type_
        self._to_remove: Set = set()
        self.to: Type["T"] = to
        self._through = through
        self.field_name: str = field_name
        self.related_models: Optional[Union[RelationProxy, "Model"]] = (
            RelationProxy(relation=self, type_=type_, to=to, field_name=field_name)
            if type_ in (RelationType.REVERSE, RelationType.MULTIPLE)
            else None
        )

    def clear(self) -> None:
        if self._type in (RelationType.PRIMARY, RelationType.THROUGH):
            self.related_models = None
            self._owner.__dict__[self.field_name] = None
        elif self.related_models is not None:
            related_models = cast("RelationProxy", self.related_models)
            related_models._clear()
            self._owner.__dict__[self.field_name] = None

    @property
    def through(self) -> Type["Model"]:
        if not self._through:  # pragma: no cover
            raise RelationshipInstanceError("Relation does not have through model!")
        return self._through

    def _clean_related(self) -> None:
        """
        Removes dead weakrefs from RelationProxy.
        """
        cleaned_data = [
            x
            for i, x in enumerate(self.related_models)  # type: ignore
            if i not in self._to_remove
        ]
        self.related_models = RelationProxy(
            relation=self,
            type_=self._type,
            to=self.to,
            field_name=self.field_name,
            data_=cleaned_data,
        )
        relation_name = self.field_name
        self._owner.__dict__[relation_name] = cleaned_data
        self._to_remove = set()

    def _find_existing(
        self, child: Union["NewBaseModel", Type["NewBaseModel"]]
    ) -> Optional[int]:
        """
        Find child model in RelationProxy if exists.

        :param child: child model to find
        :type child: Model
        :return: index of child in RelationProxy
        :rtype: Optional[ind]
        """
        if not isinstance(self.related_models, RelationProxy):  # pragma nocover
            raise ValueError("Cannot find existing models in parent relation type")
        if self._to_remove:
            self._clean_related()
        for ind, relation_child in enumerate(self.related_models[:]):
            try:
                if relation_child == child:
                    return ind
            except ReferenceError:  # pragma no cover
                self._to_remove.add(ind)
        return None

    def add(self, child: "Model") -> None:
        """
        Adds child Model to relation, either sets child as related model or adds
        it to the list in RelationProxy depending on relation type.

        :param child: model to add to relation
        :type child: Model
        """
        relation_name = self.field_name
        if self._type in (RelationType.PRIMARY, RelationType.THROUGH):
            self.related_models = child
            self._owner.__dict__[relation_name] = child
        else:
            if self._find_existing(child) is None:
                self.related_models.append(child)  # type: ignore
                rel = self._owner.__dict__.get(relation_name, [])
                rel = rel or []
                if not isinstance(rel, list):
                    rel = [rel]
                rel.append(child)
                self._owner.__dict__[relation_name] = rel

    def remove(self, child: Union["NewBaseModel", Type["NewBaseModel"]]) -> None:
        """
        Removes child Model from relation, either sets None as related model or removes
        it from the list in RelationProxy depending on relation type.

        :param child: model to remove from relation
        :type child: Model
        """
        relation_name = self.field_name
        if self._type == RelationType.PRIMARY:
            if self.related_models == child:
                self.related_models = None
                del self._owner.__dict__[relation_name]
        else:
            position = self._find_existing(child)
            if position is not None:
                self.related_models.pop(position)  # type: ignore
                del self._owner.__dict__[relation_name][position]

    def get(self) -> Optional[Union[List["Model"], "Model"]]:
        """
        Return the related model or models from RelationProxy.

        :return: related model/models if set
        :rtype: Optional[Union[List[Model], Model]]
        """
        return self.related_models

    def __repr__(self) -> str:  # pragma no cover
        if self._to_remove:
            self._clean_related()
        return str(self.related_models)

__init__(manager, type_, field_name, to, through=None)

Initialize the Relation and keep the related models either as instances of passed Model, or as a RelationProxy which is basically a list of models with some special behavior, as it exposes QuerySetProxy and allows querying the related models already pre filtered by parent model.

Parameters:

Name Type Description Default
manager RelationsManager

reference to relation manager

required
type_ RelationType

type of the relation

required
field_name str

name of the relation field

required
to Type[T]

model to which relation leads to

required
through Type[Model]

model through which relation goes for m2m relations

None
Source code in ormar/relations/relation.py
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
def __init__(
    self,
    manager: "RelationsManager",
    type_: RelationType,
    field_name: str,
    to: Type["T"],
    through: Type["Model"] = None,
) -> None:
    """
    Initialize the Relation and keep the related models either as instances of
    passed Model, or as a RelationProxy which is basically a list of models with
    some special behavior, as it exposes QuerySetProxy and allows querying the
    related models already pre filtered by parent model.

    :param manager: reference to relation manager
    :type manager: RelationsManager
    :param type_: type of the relation
    :type type_: RelationType
    :param field_name: name of the relation field
    :type field_name: str
    :param to: model to which relation leads to
    :type to: Type[Model]
    :param through: model through which relation goes for m2m relations
    :type through: Type[Model]
    """
    self.manager = manager
    self._owner: "Model" = manager.owner
    self._type: RelationType = type_
    self._to_remove: Set = set()
    self.to: Type["T"] = to
    self._through = through
    self.field_name: str = field_name
    self.related_models: Optional[Union[RelationProxy, "Model"]] = (
        RelationProxy(relation=self, type_=type_, to=to, field_name=field_name)
        if type_ in (RelationType.REVERSE, RelationType.MULTIPLE)
        else None
    )

add(child)

Adds child Model to relation, either sets child as related model or adds it to the list in RelationProxy depending on relation type.

Parameters:

Name Type Description Default
child Model

model to add to relation

required
Source code in ormar/relations/relation.py
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def add(self, child: "Model") -> None:
    """
    Adds child Model to relation, either sets child as related model or adds
    it to the list in RelationProxy depending on relation type.

    :param child: model to add to relation
    :type child: Model
    """
    relation_name = self.field_name
    if self._type in (RelationType.PRIMARY, RelationType.THROUGH):
        self.related_models = child
        self._owner.__dict__[relation_name] = child
    else:
        if self._find_existing(child) is None:
            self.related_models.append(child)  # type: ignore
            rel = self._owner.__dict__.get(relation_name, [])
            rel = rel or []
            if not isinstance(rel, list):
                rel = [rel]
            rel.append(child)
            self._owner.__dict__[relation_name] = rel

get()

Return the related model or models from RelationProxy.

Returns:

Type Description
Optional[Union[List[Model], Model]]

related model/models if set

Source code in ormar/relations/relation.py
182
183
184
185
186
187
188
189
def get(self) -> Optional[Union[List["Model"], "Model"]]:
    """
    Return the related model or models from RelationProxy.

    :return: related model/models if set
    :rtype: Optional[Union[List[Model], Model]]
    """
    return self.related_models

remove(child)

Removes child Model from relation, either sets None as related model or removes it from the list in RelationProxy depending on relation type.

Parameters:

Name Type Description Default
child Union[NewBaseModel, Type[NewBaseModel]]

model to remove from relation

required
Source code in ormar/relations/relation.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def remove(self, child: Union["NewBaseModel", Type["NewBaseModel"]]) -> None:
    """
    Removes child Model from relation, either sets None as related model or removes
    it from the list in RelationProxy depending on relation type.

    :param child: model to remove from relation
    :type child: Model
    """
    relation_name = self.field_name
    if self._type == RelationType.PRIMARY:
        if self.related_models == child:
            self.related_models = None
            del self._owner.__dict__[relation_name]
    else:
        position = self._find_existing(child)
        if position is not None:
            self.related_models.pop(position)  # type: ignore
            del self._owner.__dict__[relation_name][position]

RelationType

Bases: Enum

Different types of relations supported by ormar:

  • ForeignKey = PRIMARY
  • reverse ForeignKey = REVERSE
  • ManyToMany = MULTIPLE
Source code in ormar/relations/relation.py
25
26
27
28
29
30
31
32
33
34
35
36
37
class RelationType(Enum):
    """
    Different types of relations supported by ormar:

    *  ForeignKey = PRIMARY
    *  reverse ForeignKey = REVERSE
    *  ManyToMany = MULTIPLE
    """

    PRIMARY = 1
    REVERSE = 2
    MULTIPLE = 3
    THROUGH = 4