Skip to content

relations

Package handles relations on models, returning related models on calls and exposing QuerySetProxy for m2m and reverse relations.

AliasManager

Keep all aliases of relations between different tables. One global instance is shared between all models.

Source code in ormar/relations/alias_manager.py
Python
 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
class AliasManager:
    """
    Keep all aliases of relations between different tables.
    One global instance is shared between all models.
    """

    def __init__(self) -> None:
        self._aliases_new: dict[str, str] = dict()
        self._reversed_aliases: dict[str, str] = dict()
        self._prefixed_tables: dict[str, NamedFromClause] = dict()

    def __contains__(self, item: str) -> bool:
        return self._aliases_new.__contains__(item)

    def __getitem__(self, key: str) -> Any:
        return self._aliases_new.__getitem__(key)

    @property
    def reversed_aliases(self) -> dict:
        """
        Returns swapped key-value pairs from aliases where alias is the key.

        :return: dictionary of prefix to relation
        :rtype: dict
        """
        if self._reversed_aliases:
            return self._reversed_aliases
        reversed_aliases = {v: k for k, v in self._aliases_new.items()}
        self._reversed_aliases = reversed_aliases
        return self._reversed_aliases

    @staticmethod
    def prefixed_columns(
        alias: str, table: sqlalchemy.Table, fields: Optional[list] = None
    ) -> list[Label[Any]]:
        """
        Creates a list of aliases sqlalchemy text clauses from
        string alias and sqlalchemy.Table.

        Optional list of fields to include can be passed to extract only those columns.
        List has to have sqlalchemy names of columns (ormar aliases) not the ormar ones.

        :param alias: alias of given table
        :type alias: str
        :param table: table from which fields should be aliased
        :type table: sqlalchemy.Table
        :param fields: fields to include
        :type fields: Optional[list[str]]
        :return: list of sqlalchemy text clauses with "column name as aliased name"
        :rtype: list[text]
        """
        alias = f"{alias}_" if alias else ""
        aliased_fields = [f"{alias}{x}" for x in fields] if fields else []
        all_columns = (
            table.columns
            if not fields
            else [
                col
                for col in table.columns
                if col.name in fields or col.name in aliased_fields
            ]
        )
        return [column.label(f"{alias}{column.name}") for column in all_columns]

    def prefixed_table_name(
        self, alias: str, table: sqlalchemy.Table
    ) -> NamedFromClause:
        """
        Creates text clause with table name with aliased name.

        :param alias: alias of given table
        :type alias: str
        :param table: table
        :type table: sqlalchemy.Table
        :return: sqlalchemy text clause as "table_name aliased_name"
        :rtype: sqlalchemy text clause
        """
        full_alias = f"{alias}_{table.name}"
        key = f"{full_alias}_{id(table)}"
        return self._prefixed_tables.setdefault(key, table.alias(full_alias))

    def add_relation_type(
        self,
        source_model: type["Model"],
        relation_name: str,
        reverse_name: Optional[str] = None,
    ) -> None:
        """
        Registers the relations defined in ormar models.
        Given the relation it registers also the reverse side of this relation.

        Used by both ForeignKey and ManyToMany relations.

        Each relation is registered as Model name and relation name.
        Each alias registered has to be unique.

        Aliases are used to construct joins to assure proper links between tables.
        That way you can link to the same target tables from multiple fields
        on one model as well as from multiple different models in one join.

        :param source_model: model with relation defined
        :type source_model: source Model
        :param relation_name: name of the relation to define
        :type relation_name: str
        :param reverse_name: name of related_name fo given relation for m2m relations
        :type reverse_name: Optional[str]
        :return: none
        :rtype: None
        """
        parent_key = f"{source_model.get_name()}_{relation_name}"
        if parent_key not in self._aliases_new:
            self.add_alias(parent_key)

        to_field = source_model.ormar_config.model_fields[relation_name]
        child_model = to_field.to
        child_key = f"{child_model.get_name()}_{reverse_name}"
        if child_key not in self._aliases_new:
            self.add_alias(child_key)

    def add_alias(self, alias_key: str) -> str:
        """
        Adds alias to the dictionary of aliases under given key.

        :param alias_key: key of relation to generate alias for
        :type alias_key: str
        :return: generated alias
        :rtype: str
        """
        alias = get_table_alias()
        self._aliases_new[alias_key] = alias
        return alias

    def resolve_relation_alias(
        self, from_model: Union[type["Model"], type["ModelRow"]], relation_name: str
    ) -> str:
        """
        Given model and relation name returns the alias for this relation.

        :param from_model: model with relation defined
        :type from_model: source Model
        :param relation_name: name of the relation field
        :type relation_name: str
        :return: alias of the relation
        :rtype: str
        """
        alias = self._aliases_new.get(f"{from_model.get_name()}_{relation_name}", "")
        return alias

    def resolve_relation_alias_after_complex(
        self,
        source_model: Union[type["Model"], type["ModelRow"]],
        relation_str: str,
        relation_field: "ForeignKeyField",
    ) -> str:
        """
        Given source model and relation string returns the alias for this complex
        relation if it exists, otherwise fallback to normal relation from a relation
        field definition.

        :param relation_field: field with direct relation definition
        :type relation_field: "ForeignKeyField"
        :param source_model: model with query starts
        :type source_model: source Model
        :param relation_str: string with relation joins defined
        :type relation_str: str
        :return: alias of the relation
        :rtype: str
        """
        alias = ""
        if relation_str and "__" in relation_str:
            alias = self.resolve_relation_alias(
                from_model=source_model, relation_name=relation_str
            )
        if not alias:
            alias = self.resolve_relation_alias(
                from_model=relation_field.get_source_model(),
                relation_name=relation_field.get_relation_name(),
            )
        return alias

reversed_aliases property

Returns swapped key-value pairs from aliases where alias is the key.

:return: dictionary of prefix to relation :rtype: dict

add_alias(alias_key)

Adds alias to the dictionary of aliases under given key.

:param alias_key: key of relation to generate alias for :type alias_key: str :return: generated alias :rtype: str

Source code in ormar/relations/alias_manager.py
Python
149
150
151
152
153
154
155
156
157
158
159
160
def add_alias(self, alias_key: str) -> str:
    """
    Adds alias to the dictionary of aliases under given key.

    :param alias_key: key of relation to generate alias for
    :type alias_key: str
    :return: generated alias
    :rtype: str
    """
    alias = get_table_alias()
    self._aliases_new[alias_key] = alias
    return alias

add_relation_type(source_model, relation_name, reverse_name=None)

Registers the relations defined in ormar models. Given the relation it registers also the reverse side of this relation.

Used by both ForeignKey and ManyToMany relations.

Each relation is registered as Model name and relation name. Each alias registered has to be unique.

Aliases are used to construct joins to assure proper links between tables. That way you can link to the same target tables from multiple fields on one model as well as from multiple different models in one join.

:param source_model: model with relation defined :type source_model: source Model :param relation_name: name of the relation to define :type relation_name: str :param reverse_name: name of related_name fo given relation for m2m relations :type reverse_name: Optional[str] :return: none :rtype: None

Source code in ormar/relations/alias_manager.py
Python
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
def add_relation_type(
    self,
    source_model: type["Model"],
    relation_name: str,
    reverse_name: Optional[str] = None,
) -> None:
    """
    Registers the relations defined in ormar models.
    Given the relation it registers also the reverse side of this relation.

    Used by both ForeignKey and ManyToMany relations.

    Each relation is registered as Model name and relation name.
    Each alias registered has to be unique.

    Aliases are used to construct joins to assure proper links between tables.
    That way you can link to the same target tables from multiple fields
    on one model as well as from multiple different models in one join.

    :param source_model: model with relation defined
    :type source_model: source Model
    :param relation_name: name of the relation to define
    :type relation_name: str
    :param reverse_name: name of related_name fo given relation for m2m relations
    :type reverse_name: Optional[str]
    :return: none
    :rtype: None
    """
    parent_key = f"{source_model.get_name()}_{relation_name}"
    if parent_key not in self._aliases_new:
        self.add_alias(parent_key)

    to_field = source_model.ormar_config.model_fields[relation_name]
    child_model = to_field.to
    child_key = f"{child_model.get_name()}_{reverse_name}"
    if child_key not in self._aliases_new:
        self.add_alias(child_key)

prefixed_columns(alias, table, fields=None) staticmethod

Creates a list of aliases sqlalchemy text clauses from string alias and sqlalchemy.Table.

Optional list of fields to include can be passed to extract only those columns. List has to have sqlalchemy names of columns (ormar aliases) not the ormar ones.

:param alias: alias of given table :type alias: str :param table: table from which fields should be aliased :type table: sqlalchemy.Table :param fields: fields to include :type fields: Optional[list[str]] :return: list of sqlalchemy text clauses with "column name as aliased name" :rtype: list[text]

Source code in ormar/relations/alias_manager.py
Python
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
@staticmethod
def prefixed_columns(
    alias: str, table: sqlalchemy.Table, fields: Optional[list] = None
) -> list[Label[Any]]:
    """
    Creates a list of aliases sqlalchemy text clauses from
    string alias and sqlalchemy.Table.

    Optional list of fields to include can be passed to extract only those columns.
    List has to have sqlalchemy names of columns (ormar aliases) not the ormar ones.

    :param alias: alias of given table
    :type alias: str
    :param table: table from which fields should be aliased
    :type table: sqlalchemy.Table
    :param fields: fields to include
    :type fields: Optional[list[str]]
    :return: list of sqlalchemy text clauses with "column name as aliased name"
    :rtype: list[text]
    """
    alias = f"{alias}_" if alias else ""
    aliased_fields = [f"{alias}{x}" for x in fields] if fields else []
    all_columns = (
        table.columns
        if not fields
        else [
            col
            for col in table.columns
            if col.name in fields or col.name in aliased_fields
        ]
    )
    return [column.label(f"{alias}{column.name}") for column in all_columns]

prefixed_table_name(alias, table)

Creates text clause with table name with aliased name.

:param alias: alias of given table :type alias: str :param table: table :type table: sqlalchemy.Table :return: sqlalchemy text clause as "table_name aliased_name" :rtype: sqlalchemy text clause

Source code in ormar/relations/alias_manager.py
Python
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def prefixed_table_name(
    self, alias: str, table: sqlalchemy.Table
) -> NamedFromClause:
    """
    Creates text clause with table name with aliased name.

    :param alias: alias of given table
    :type alias: str
    :param table: table
    :type table: sqlalchemy.Table
    :return: sqlalchemy text clause as "table_name aliased_name"
    :rtype: sqlalchemy text clause
    """
    full_alias = f"{alias}_{table.name}"
    key = f"{full_alias}_{id(table)}"
    return self._prefixed_tables.setdefault(key, table.alias(full_alias))

resolve_relation_alias(from_model, relation_name)

Given model and relation name returns the alias for this relation.

:param from_model: model with relation defined :type from_model: source Model :param relation_name: name of the relation field :type relation_name: str :return: alias of the relation :rtype: str

Source code in ormar/relations/alias_manager.py
Python
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def resolve_relation_alias(
    self, from_model: Union[type["Model"], type["ModelRow"]], relation_name: str
) -> str:
    """
    Given model and relation name returns the alias for this relation.

    :param from_model: model with relation defined
    :type from_model: source Model
    :param relation_name: name of the relation field
    :type relation_name: str
    :return: alias of the relation
    :rtype: str
    """
    alias = self._aliases_new.get(f"{from_model.get_name()}_{relation_name}", "")
    return alias

resolve_relation_alias_after_complex(source_model, relation_str, relation_field)

Given source model and relation string returns the alias for this complex relation if it exists, otherwise fallback to normal relation from a relation field definition.

:param relation_field: field with direct relation definition :type relation_field: "ForeignKeyField" :param source_model: model with query starts :type source_model: source Model :param relation_str: string with relation joins defined :type relation_str: str :return: alias of the relation :rtype: str

Source code in ormar/relations/alias_manager.py
Python
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
def resolve_relation_alias_after_complex(
    self,
    source_model: Union[type["Model"], type["ModelRow"]],
    relation_str: str,
    relation_field: "ForeignKeyField",
) -> str:
    """
    Given source model and relation string returns the alias for this complex
    relation if it exists, otherwise fallback to normal relation from a relation
    field definition.

    :param relation_field: field with direct relation definition
    :type relation_field: "ForeignKeyField"
    :param source_model: model with query starts
    :type source_model: source Model
    :param relation_str: string with relation joins defined
    :type relation_str: str
    :return: alias of the relation
    :rtype: str
    """
    alias = ""
    if relation_str and "__" in relation_str:
        alias = self.resolve_relation_alias(
            from_model=source_model, relation_name=relation_str
        )
    if not alias:
        alias = self.resolve_relation_alias(
            from_model=relation_field.get_source_model(),
            relation_name=relation_field.get_relation_name(),
        )
    return alias

Relation

Bases: Generic[T]

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

Source code in ormar/relations/relation.py
Python
 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 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: Optional[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
        # ``RelationProxy`` is built lazily on the first reverse/m2m use
        # (``add`` / ``get`` / ``_clean_related``) to avoid the per-model
        # allocation when the relation is never read.
        self.related_models: Optional[Union[RelationProxy, "Model"]] = None

    def _ensure_proxy(self) -> RelationProxy:
        """
        Materialize and cache the ``RelationProxy`` for reverse/m2m relations
        on first use. Safe to call multiple times — subsequent calls return
        the cached proxy.

        :return: the relation's ``RelationProxy``
        :rtype: RelationProxy
        """
        proxy = self.related_models
        if not isinstance(proxy, RelationProxy):
            proxy = RelationProxy(
                relation=self,
                type_=self._type,
                to=self.to,
                field_name=self.field_name,
            )
            self.related_models = proxy
        return proxy

    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
        ]
        proxy = RelationProxy(
            relation=self,
            type_=self._type,
            to=self.to,
            field_name=self.field_name,
            data_=cleaned_data,
        )
        self.related_models = proxy
        self._owner.__dict__[self.field_name] = proxy
        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 child not in self.related_models:
            return None
        else:
            # We need to clear the weakrefs that don't point to anything anymore
            # There's an assumption here that if some of the related models
            # went out of scope, then they all did, so we can just check the first one
            try:
                self.related_models[0].__repr__.__self__
                return self.related_models.index(child)
            except ReferenceError:
                missing = self.related_models._get_list_of_missing_weakrefs()
                self._to_remove.update(missing)
            return self.related_models.index(child)

    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.

        For reverse / many-to-many relations the ``RelationProxy`` itself is
        stored under ``_owner.__dict__[relation_name]``, so a single O(1)
        membership check on the proxy's hash cache covers both the relation
        bookkeeping and the pydantic-visible ``__dict__`` slot — no parallel
        list, no second linear scan.

        :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
            return
        proxy = self._ensure_proxy()
        # ``_find_existing`` is the membership check *plus* a dead-weakref
        # probe — when a hash collision lands on a stale entry we need that
        # probe to populate ``_to_remove`` so the next ``get()`` triggers
        # ``_clean_related``.
        if self._find_existing(child) is None:
            proxy.append(child)
            self._owner.__dict__[relation_name] = proxy

    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

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

        For reverse / many-to-many relations the ``RelationProxy`` is
        materialized on first read so callers always see a list-like
        return value (even when no children have been registered yet).

        :return: related model/models if set
        :rtype: Optional[Union[list[Model], Model]]
        """
        if self._to_remove:
            self._clean_related()
        if self.related_models is None and self._type in (
            RelationType.REVERSE,
            RelationType.MULTIPLE,
        ):
            return self._ensure_proxy()
        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.

: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]

Source code in ormar/relations/relation.py
Python
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
def __init__(
    self,
    manager: "RelationsManager",
    type_: RelationType,
    field_name: str,
    to: type["T"],
    through: Optional[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
    # ``RelationProxy`` is built lazily on the first reverse/m2m use
    # (``add`` / ``get`` / ``_clean_related``) to avoid the per-model
    # allocation when the relation is never read.
    self.related_models: Optional[Union[RelationProxy, "Model"]] = 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.

For reverse / many-to-many relations the RelationProxy itself is stored under _owner.__dict__[relation_name], so a single O(1) membership check on the proxy's hash cache covers both the relation bookkeeping and the pydantic-visible __dict__ slot — no parallel list, no second linear scan.

:param child: model to add to relation :type child: Model

Source code in ormar/relations/relation.py
Python
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
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.

    For reverse / many-to-many relations the ``RelationProxy`` itself is
    stored under ``_owner.__dict__[relation_name]``, so a single O(1)
    membership check on the proxy's hash cache covers both the relation
    bookkeeping and the pydantic-visible ``__dict__`` slot — no parallel
    list, no second linear scan.

    :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
        return
    proxy = self._ensure_proxy()
    # ``_find_existing`` is the membership check *plus* a dead-weakref
    # probe — when a hash collision lands on a stale entry we need that
    # probe to populate ``_to_remove`` so the next ``get()`` triggers
    # ``_clean_related``.
    if self._find_existing(child) is None:
        proxy.append(child)
        self._owner.__dict__[relation_name] = proxy

get()

Return the related model or models from RelationProxy.

For reverse / many-to-many relations the RelationProxy is materialized on first read so callers always see a list-like return value (even when no children have been registered yet).

:return: related model/models if set :rtype: Optional[Union[list[Model], Model]]

Source code in ormar/relations/relation.py
Python
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def get(self) -> Optional[Union[list["Model"], "Model"]]:
    """
    Return the related model or models from RelationProxy.

    For reverse / many-to-many relations the ``RelationProxy`` is
    materialized on first read so callers always see a list-like
    return value (even when no children have been registered yet).

    :return: related model/models if set
    :rtype: Optional[Union[list[Model], Model]]
    """
    if self._to_remove:
        self._clean_related()
    if self.related_models is None and self._type in (
        RelationType.REVERSE,
        RelationType.MULTIPLE,
    ):
        return self._ensure_proxy()
    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.

:param child: model to remove from relation :type child: Model

Source code in ormar/relations/relation.py
Python
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
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

RelationType

Bases: Enum

Different types of relations supported by ormar:

  • ForeignKey = PRIMARY
  • reverse ForeignKey = REVERSE
  • ManyToMany = MULTIPLE
Source code in ormar/relations/relation.py
Python
15
16
17
18
19
20
21
22
23
24
25
26
27
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

RelationsManager

Manages relations on a Model, each Model has it's own instance.

Source code in ormar/relations/relation_manager.py
Python
 12
 13
 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
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
class RelationsManager:
    """
    Manages relations on a Model, each Model has it's own instance.
    """

    def __init__(
        self,
        related_fields: Optional[list["ForeignKeyField"]] = None,
        owner: Optional["Model"] = None,
    ) -> None:
        self.owner = proxy(owner)
        self._related_fields = related_fields or []
        # ``_field_map`` lets ``_get`` build a ``Relation`` lazily by name.
        # Holding only the field reference (not a constructed Relation) is
        # what skips the per-FK Relation/RelationProxy/QuerysetProxy
        # allocation tree on every ``Model.__init__``.
        self._field_map: dict[str, "ForeignKeyField"] = {
            field.name: field for field in self._related_fields
        }
        self._related_names = list(self._field_map)
        self._relations: dict[str, Relation] = dict()

    def __contains__(self, item: str) -> bool:
        """
        Checks if relation with given name is already registered.

        :param item: name of attribute
        :type item: str
        :return: result of the check
        :rtype: bool
        """
        return item in self._related_names

    def clear(self) -> None:
        for relation in self._relations.values():
            relation.clear()

    def get(self, name: str) -> Optional[Union["Model", Sequence["Model"]]]:
        """
        Returns the related model/models if relation is set.
        Actual call is delegated to Relation instance registered under relation name.

        :param name: name of the relation
        :type name: str
        :return: related model or list of related models if set
        :rtype: Optional[Union[Model, list[Model]]
        """
        relation = self._get(name)
        if relation is not None:
            return relation.get()
        return None  # pragma nocover

    @staticmethod
    def add(parent: "Model", child: "Model", field: "ForeignKeyField") -> None:
        """
        Adds relation on both sides -> meaning on both child and parent models.
        One side of the relation is always weakref proxy to avoid circular refs.

        Based on the side from which relation is added and relation name actual names
        of parent and child relations are established. The related models are registered
        on both ends.

        :param parent: parent model on which relation should be registered
        :type parent: Model
        :param child: child model to register
        :type child: Model
        :param field: field with relation definition
        :type field: ForeignKeyField
        """
        parent, child, child_name, to_name = get_relations_sides_and_names(
            field, parent, child
        )

        # print('adding parent', parent.get_name(), child.get_name(), child_name)
        parent_relation = parent._orm._get(child_name)
        if parent_relation:
            parent_relation.add(child)  # type: ignore

        # print('adding child', child.get_name(), parent.get_name(), to_name)
        child_relation = child._orm._get(to_name)
        if child_relation:
            child_relation.add(parent)

    def remove(
        self, name: str, child: Union["NewBaseModel", type["NewBaseModel"]]
    ) -> None:
        """
        Removes given child from relation with given name.
        Since you can have many relations between two models you need to pass a name
        of relation from which you want to remove the child.

        :param name: name of the relation
        :type name: str
        :param child: child to remove from relation
        :type child: Union[Model, type[Model]]
        """
        relation = self._get(name)
        if relation:
            relation.remove(child)

    @staticmethod
    def remove_parent(
        item: Union["NewBaseModel", type["NewBaseModel"]], parent: "Model", name: str
    ) -> None:
        """
        Removes given parent from relation with given name.
        Since you can have many relations between two models you need to pass a name
        of relation from which you want to remove the parent.

        :param item: model with parent registered
        :type item: Union[Model, type[Model]]
        :param parent: parent Model
        :type parent: Model
        :param name: name of the relation
        :type name: str
        """
        relation_name = item.ormar_config.model_fields[name].get_related_name()
        item._orm.remove(name, parent)
        parent._orm.remove(relation_name, item)

    def _get(self, name: str) -> Optional[Relation]:
        """
        Return the ``Relation`` for ``name``, building it on first access.

        Relations are constructed lazily so that ``Model.__init__`` does
        not allocate a ``Relation`` (and, transitively, ``RelationProxy`` /
        ``QuerysetProxy``) for every declared FK on every instance — most
        of which are never read on row-materialization paths.

        :param name: name of the relation
        :type name: str
        :return: existing or freshly constructed Relation, or None if the
            name does not correspond to a declared relation
        :rtype: ormar.relations.relation.Relation
        """
        relation = self._relations.get(name)
        if relation is not None:
            return relation
        field = self._field_map.get(name)
        if field is None:
            return None
        relation = Relation(
            manager=self,
            type_=self._get_relation_type(field),
            field_name=field.name,
            to=field.to,
            through=getattr(field, "through", None),
        )
        self._relations[name] = relation
        return relation

    def _get_relation_type(self, field: "BaseField") -> RelationType:
        """
        Returns type of the relation declared on a field.

        :param field: field with relation declaration
        :type field: BaseField
        :return: type of the relation defined on field
        :rtype: RelationType
        """
        if field.is_multi:
            return RelationType.MULTIPLE
        if field.is_through:
            return RelationType.THROUGH
        return RelationType.PRIMARY if not field.virtual else RelationType.REVERSE

__contains__(item)

Checks if relation with given name is already registered.

:param item: name of attribute :type item: str :return: result of the check :rtype: bool

Source code in ormar/relations/relation_manager.py
Python
34
35
36
37
38
39
40
41
42
43
def __contains__(self, item: str) -> bool:
    """
    Checks if relation with given name is already registered.

    :param item: name of attribute
    :type item: str
    :return: result of the check
    :rtype: bool
    """
    return item in self._related_names

add(parent, child, field) staticmethod

Adds relation on both sides -> meaning on both child and parent models. One side of the relation is always weakref proxy to avoid circular refs.

Based on the side from which relation is added and relation name actual names of parent and child relations are established. The related models are registered on both ends.

:param parent: parent model on which relation should be registered :type parent: Model :param child: child model to register :type child: Model :param field: field with relation definition :type field: ForeignKeyField

Source code in ormar/relations/relation_manager.py
Python
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
@staticmethod
def add(parent: "Model", child: "Model", field: "ForeignKeyField") -> None:
    """
    Adds relation on both sides -> meaning on both child and parent models.
    One side of the relation is always weakref proxy to avoid circular refs.

    Based on the side from which relation is added and relation name actual names
    of parent and child relations are established. The related models are registered
    on both ends.

    :param parent: parent model on which relation should be registered
    :type parent: Model
    :param child: child model to register
    :type child: Model
    :param field: field with relation definition
    :type field: ForeignKeyField
    """
    parent, child, child_name, to_name = get_relations_sides_and_names(
        field, parent, child
    )

    # print('adding parent', parent.get_name(), child.get_name(), child_name)
    parent_relation = parent._orm._get(child_name)
    if parent_relation:
        parent_relation.add(child)  # type: ignore

    # print('adding child', child.get_name(), parent.get_name(), to_name)
    child_relation = child._orm._get(to_name)
    if child_relation:
        child_relation.add(parent)

get(name)

Returns the related model/models if relation is set. Actual call is delegated to Relation instance registered under relation name.

:param name: name of the relation :type name: str :return: related model or list of related models if set :rtype: Optional[Union[Model, list[Model]]

Source code in ormar/relations/relation_manager.py
Python
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get(self, name: str) -> Optional[Union["Model", Sequence["Model"]]]:
    """
    Returns the related model/models if relation is set.
    Actual call is delegated to Relation instance registered under relation name.

    :param name: name of the relation
    :type name: str
    :return: related model or list of related models if set
    :rtype: Optional[Union[Model, list[Model]]
    """
    relation = self._get(name)
    if relation is not None:
        return relation.get()
    return None  # pragma nocover

remove(name, child)

Removes given child from relation with given name. Since you can have many relations between two models you need to pass a name of relation from which you want to remove the child.

:param name: name of the relation :type name: str :param child: child to remove from relation :type child: Union[Model, type[Model]]

Source code in ormar/relations/relation_manager.py
Python
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def remove(
    self, name: str, child: Union["NewBaseModel", type["NewBaseModel"]]
) -> None:
    """
    Removes given child from relation with given name.
    Since you can have many relations between two models you need to pass a name
    of relation from which you want to remove the child.

    :param name: name of the relation
    :type name: str
    :param child: child to remove from relation
    :type child: Union[Model, type[Model]]
    """
    relation = self._get(name)
    if relation:
        relation.remove(child)

remove_parent(item, parent, name) staticmethod

Removes given parent from relation with given name. Since you can have many relations between two models you need to pass a name of relation from which you want to remove the parent.

:param item: model with parent registered :type item: Union[Model, type[Model]] :param parent: parent Model :type parent: Model :param name: name of the relation :type name: str

Source code in ormar/relations/relation_manager.py
Python
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@staticmethod
def remove_parent(
    item: Union["NewBaseModel", type["NewBaseModel"]], parent: "Model", name: str
) -> None:
    """
    Removes given parent from relation with given name.
    Since you can have many relations between two models you need to pass a name
    of relation from which you want to remove the parent.

    :param item: model with parent registered
    :type item: Union[Model, type[Model]]
    :param parent: parent Model
    :type parent: Model
    :param name: name of the relation
    :type name: str
    """
    relation_name = item.ormar_config.model_fields[name].get_related_name()
    item._orm.remove(name, parent)
    parent._orm.remove(relation_name, item)

get_relations_sides_and_names(to_field, parent, child)

Determines the names of child and parent relations names, as well as changes one of the sides of the relation into weakref.proxy to model.

:param to_field: field with relation definition :type to_field: ForeignKeyField :param parent: parent model :type parent: Model :param child: child model :type child: Model :return: parent, child, child_name, to_name :rtype: tuple["Model", "Model", str, str]

Source code in ormar/relations/utils.py
Python
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def get_relations_sides_and_names(
    to_field: ForeignKeyField, parent: "Model", child: "Model"
) -> tuple["Model", "Model", str, str]:
    """
    Determines the names of child and parent relations names, as well as
    changes one of the sides of the relation into weakref.proxy to model.

    :param to_field: field with relation definition
    :type to_field: ForeignKeyField
    :param parent: parent model
    :type parent: Model
    :param child: child model
    :type child: Model
    :return: parent, child, child_name, to_name
    :rtype: tuple["Model", "Model", str, str]
    """
    to_name = to_field.name
    child_name = to_field.get_related_name()
    if to_field.virtual:
        child_name, to_name = to_name, child_name
        child, parent = parent, proxy(child)
    else:
        child = proxy(child)
    return parent, child, child_name, to_name