Skip to content

relation_proxy

RelationProxy

Bases: Generic[T], list

Proxy of the Relation that is a list with special methods.

Source code in ormar/relations/relation_proxy.py
 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
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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class RelationProxy(Generic[T], list):
    """
    Proxy of the Relation that is a list with special methods.
    """

    def __init__(
        self,
        relation: "Relation",
        type_: "RelationType",
        to: Type["T"],
        field_name: str,
        data_: Any = None,
    ) -> None:
        super().__init__(data_ or ())
        self.relation: "Relation[T]" = relation
        self.type_: "RelationType" = type_
        self.field_name = field_name
        self._owner: "Model" = self.relation.manager.owner
        self.queryset_proxy: QuerysetProxy[T] = QuerysetProxy[T](
            relation=self.relation, to=to, type_=type_
        )
        self._related_field_name: Optional[str] = None

    @property
    def related_field_name(self) -> str:
        """
        On first access calculates the name of the related field, later stored in
        _related_field_name property.

        :return: name of the related field
        :rtype: str
        """
        if self._related_field_name:
            return self._related_field_name
        owner_field = self._owner.Meta.model_fields[self.field_name]
        self._related_field_name = owner_field.get_related_name()

        return self._related_field_name

    def __getitem__(self, item: Any) -> "T":  # type: ignore
        return super().__getitem__(item)

    def __getattribute__(self, item: str) -> Any:
        """
        Since some QuerySetProxy methods overwrite builtin list methods we
        catch calls to them and delegate it to QuerySetProxy instead.

        :param item: name of attribute
        :type item: str
        :return: value of attribute
        :rtype: Any
        """
        if item in ["count", "clear"]:
            self._initialize_queryset()
            return getattr(self.queryset_proxy, item)
        return super().__getattribute__(item)

    def __getattr__(self, item: str) -> Any:
        """
        Delegates calls for non existing attributes to QuerySetProxy.

        :param item: name of attribute/method
        :type item: str
        :return: method from QuerySetProxy if exists
        :rtype: method
        """
        self._initialize_queryset()
        return getattr(self.queryset_proxy, item)

    def _clear(self) -> None:
        super().clear()

    def _initialize_queryset(self) -> None:
        """
        Initializes the QuerySetProxy if not yet initialized.
        """
        if not self._check_if_queryset_is_initialized():
            self.queryset_proxy.queryset = self._set_queryset()

    def _check_if_queryset_is_initialized(self) -> bool:
        """
        Checks if the QuerySetProxy is already set and ready.
        :return: result of the check
        :rtype: bool
        """
        return (
            hasattr(self.queryset_proxy, "queryset")
            and self.queryset_proxy.queryset is not None
        )

    def _check_if_model_saved(self) -> None:
        """
        Verifies if the parent model of the relation has been already saved.
        Otherwise QuerySetProxy cannot filter by parent primary key.
        """
        pk_value = self._owner.pk
        if not pk_value:
            raise RelationshipInstanceError(
                "You cannot query relationships from unsaved model."
            )

    def _set_queryset(self) -> "QuerySet[T]":
        """
        Creates new QuerySet with relation model and pre filters it with currents
        parent model primary key, so all queries by definition are already related
        to the parent model only, without need for user to filter them.

        :return: initialized QuerySet
        :rtype: QuerySet
        """
        related_field_name = self.related_field_name
        pkname = self._owner.get_column_alias(self._owner.Meta.pkname)
        self._check_if_model_saved()
        kwargs = {f"{related_field_name}__{pkname}": self._owner.pk}
        queryset = (
            ormar.QuerySet(
                model_cls=self.relation.to, proxy_source_model=self._owner.__class__
            )
            .select_related(related_field_name)
            .filter(**kwargs)
        )
        return queryset

    async def remove(  # type: ignore
        self, item: "T", keep_reversed: bool = True
    ) -> None:
        """
        Removes the related from relation with parent.

        Through models are automatically deleted for m2m relations.

        For reverse FK relations keep_reversed flag marks if the reversed models
        should be kept or deleted from the database too (False means that models
        will be deleted, and not only removed from relation).

        :param item: child to remove from relation
        :type item: Model
        :param keep_reversed: flag if the reversed model should be kept or deleted too
        :type keep_reversed: bool
        """
        if item not in self:
            raise NoMatch(
                f"Object {self._owner.get_name()} has no "
                f"{item.get_name()} with given primary key!"
            )
        await self._owner.signals.pre_relation_remove.send(
            sender=self._owner.__class__,
            instance=self._owner,
            child=item,
            relation_name=self.field_name,
        )
        super().remove(item)
        relation_name = self.related_field_name
        relation = item._orm._get(relation_name)
        # if relation is None:  # pragma nocover
        #     raise ValueError(
        #         f"{self._owner.get_name()} does not have relation {relation_name}"
        #     )
        if relation:
            relation.remove(self._owner)
        self.relation.remove(item)
        if self.type_ == ormar.RelationType.MULTIPLE:
            await self.queryset_proxy.delete_through_instance(item)
        else:
            if keep_reversed:
                setattr(item, relation_name, None)
                await item.update()
            else:
                await item.delete()
        await self._owner.signals.post_relation_remove.send(
            sender=self._owner.__class__,
            instance=self._owner,
            child=item,
            relation_name=self.field_name,
        )

    async def add(self, item: "T", **kwargs: Any) -> None:
        """
        Adds child model to relation.

        For ManyToMany relations through instance is automatically created.

        :param kwargs: dict of additional keyword arguments for through instance
        :type kwargs: Any
        :param item: child to add to relation
        :type item: Model
        """
        relation_name = self.related_field_name
        await self._owner.signals.pre_relation_add.send(
            sender=self._owner.__class__,
            instance=self._owner,
            child=item,
            relation_name=self.field_name,
            passed_kwargs=kwargs,
        )
        self._check_if_model_saved()
        if self.type_ == ormar.RelationType.MULTIPLE:
            await self.queryset_proxy.create_through_instance(item, **kwargs)
            setattr(self._owner, self.field_name, item)
        else:
            setattr(item, relation_name, self._owner)
            await item.upsert()
        await self._owner.signals.post_relation_add.send(
            sender=self._owner.__class__,
            instance=self._owner,
            child=item,
            relation_name=self.field_name,
            passed_kwargs=kwargs,
        )

__getattr__(item)

Delegates calls for non existing attributes to QuerySetProxy.

Parameters:

Name Type Description Default
item str

name of attribute/method

required

Returns:

Type Description
method

method from QuerySetProxy if exists

Source code in ormar/relations/relation_proxy.py
73
74
75
76
77
78
79
80
81
82
83
def __getattr__(self, item: str) -> Any:
    """
    Delegates calls for non existing attributes to QuerySetProxy.

    :param item: name of attribute/method
    :type item: str
    :return: method from QuerySetProxy if exists
    :rtype: method
    """
    self._initialize_queryset()
    return getattr(self.queryset_proxy, item)

__getattribute__(item)

Since some QuerySetProxy methods overwrite builtin list methods we catch calls to them and delegate it to QuerySetProxy instead.

Parameters:

Name Type Description Default
item str

name of attribute

required

Returns:

Type Description
Any

value of attribute

Source code in ormar/relations/relation_proxy.py
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def __getattribute__(self, item: str) -> Any:
    """
    Since some QuerySetProxy methods overwrite builtin list methods we
    catch calls to them and delegate it to QuerySetProxy instead.

    :param item: name of attribute
    :type item: str
    :return: value of attribute
    :rtype: Any
    """
    if item in ["count", "clear"]:
        self._initialize_queryset()
        return getattr(self.queryset_proxy, item)
    return super().__getattribute__(item)

add(item, **kwargs) async

Adds child model to relation.

For ManyToMany relations through instance is automatically created.

Parameters:

Name Type Description Default
kwargs Any

dict of additional keyword arguments for through instance

required
item 'T'

child to add to relation

required
Source code in ormar/relations/relation_proxy.py
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def add(self, item: "T", **kwargs: Any) -> None:
    """
    Adds child model to relation.

    For ManyToMany relations through instance is automatically created.

    :param kwargs: dict of additional keyword arguments for through instance
    :type kwargs: Any
    :param item: child to add to relation
    :type item: Model
    """
    relation_name = self.related_field_name
    await self._owner.signals.pre_relation_add.send(
        sender=self._owner.__class__,
        instance=self._owner,
        child=item,
        relation_name=self.field_name,
        passed_kwargs=kwargs,
    )
    self._check_if_model_saved()
    if self.type_ == ormar.RelationType.MULTIPLE:
        await self.queryset_proxy.create_through_instance(item, **kwargs)
        setattr(self._owner, self.field_name, item)
    else:
        setattr(item, relation_name, self._owner)
        await item.upsert()
    await self._owner.signals.post_relation_add.send(
        sender=self._owner.__class__,
        instance=self._owner,
        child=item,
        relation_name=self.field_name,
        passed_kwargs=kwargs,
    )

related_field_name() property

On first access calculates the name of the related field, later stored in _related_field_name property.

Returns:

Type Description
str

name of the related field

Source code in ormar/relations/relation_proxy.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@property
def related_field_name(self) -> str:
    """
    On first access calculates the name of the related field, later stored in
    _related_field_name property.

    :return: name of the related field
    :rtype: str
    """
    if self._related_field_name:
        return self._related_field_name
    owner_field = self._owner.Meta.model_fields[self.field_name]
    self._related_field_name = owner_field.get_related_name()

    return self._related_field_name

remove(item, keep_reversed=True) async

Removes the related from relation with parent.

Through models are automatically deleted for m2m relations.

For reverse FK relations keep_reversed flag marks if the reversed models should be kept or deleted from the database too (False means that models will be deleted, and not only removed from relation).

Parameters:

Name Type Description Default
item 'T'

child to remove from relation

required
keep_reversed bool

flag if the reversed model should be kept or deleted too

True
Source code in ormar/relations/relation_proxy.py
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
async def remove(  # type: ignore
    self, item: "T", keep_reversed: bool = True
) -> None:
    """
    Removes the related from relation with parent.

    Through models are automatically deleted for m2m relations.

    For reverse FK relations keep_reversed flag marks if the reversed models
    should be kept or deleted from the database too (False means that models
    will be deleted, and not only removed from relation).

    :param item: child to remove from relation
    :type item: Model
    :param keep_reversed: flag if the reversed model should be kept or deleted too
    :type keep_reversed: bool
    """
    if item not in self:
        raise NoMatch(
            f"Object {self._owner.get_name()} has no "
            f"{item.get_name()} with given primary key!"
        )
    await self._owner.signals.pre_relation_remove.send(
        sender=self._owner.__class__,
        instance=self._owner,
        child=item,
        relation_name=self.field_name,
    )
    super().remove(item)
    relation_name = self.related_field_name
    relation = item._orm._get(relation_name)
    # if relation is None:  # pragma nocover
    #     raise ValueError(
    #         f"{self._owner.get_name()} does not have relation {relation_name}"
    #     )
    if relation:
        relation.remove(self._owner)
    self.relation.remove(item)
    if self.type_ == ormar.RelationType.MULTIPLE:
        await self.queryset_proxy.delete_through_instance(item)
    else:
        if keep_reversed:
            setattr(item, relation_name, None)
            await item.update()
        else:
            await item.delete()
    await self._owner.signals.post_relation_remove.send(
        sender=self._owner.__class__,
        instance=self._owner,
        child=item,
        relation_name=self.field_name,
    )