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 
 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
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, text] = 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: List = None
    ) -> List[text]:
        """
        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) -> text:
        """
        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: 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.Meta.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: Dict property

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

Returns:

Type Description
Dict

dictionary of prefix to relation

add_alias(alias_key)

Adds alias to the dictionary of aliases under given key.

Parameters:

Name Type Description Default
alias_key str

key of relation to generate alias for

 required

Returns:

Type Description
str

generated alias
Source code in ormar\relations\alias_manager.py 
143
144
145
146
147
148
149
150
151
152
153
154
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.

Parameters:

Name Type Description Default
source_model Type[Model]

model with relation defined

 required
relation_name str

name of the relation to define

 required
reverse_name str

name of related_name fo given relation for m2m relations

 None

Returns:

Type Description
None

none
Source code in ormar\relations\alias_manager.py 
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
def add_relation_type(
    self, source_model: Type["Model"], relation_name: str, reverse_name: 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.Meta.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.

Parameters:

Name Type Description Default
alias str

alias of given table

 required
table sqlalchemy.Table

table from which fields should be aliased

 required
fields List

fields to include

 None

Returns:

Type Description
List[text]

list of sqlalchemy text clauses with "column name as aliased name"
Source code in ormar\relations\alias_manager.py 
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
@staticmethod
def prefixed_columns(
    alias: str, table: sqlalchemy.Table, fields: List = None
) -> List[text]:
    """
    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.

Parameters:

Name Type Description Default
alias str

alias of given table

 required
table sqlalchemy.Table

table

 required

Returns:

Type Description
sqlalchemy text clause

sqlalchemy text clause as "table_name aliased_name"
Source code in ormar\relations\alias_manager.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def prefixed_table_name(self, alias: str, table: sqlalchemy.Table) -> text:
    """
    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.

Parameters:

Name Type Description Default
from_model Union[Type[Model], Type[ModelRow]]

model with relation defined

 required
relation_name str

name of the relation field

 required

Returns:

Type Description
str

alias of the relation
Source code in ormar\relations\alias_manager.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
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.

Parameters:

Name Type Description Default
relation_field ForeignKeyField

field with direct relation definition

 required
source_model Union[Type[Model], Type[ModelRow]]

model with query starts

 required
relation_str str

string with relation joins defined

 required

Returns:

Type Description
str

alias of the relation
Source code in ormar\relations\alias_manager.py 
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
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 
 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
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 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.

        :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]]
        """
        if self._to_remove:
            self._clean_related()
        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 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
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 
187
188
189
190
191
192
193
194
195
196
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]]
    """
    if self._to_remove:
        self._clean_related()
    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 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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

RelationsManager

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

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

    def __init__(
        self,
        related_fields: List["ForeignKeyField"] = None,
        owner: Optional["Model"] = None,
    ) -> None:
        self.owner = proxy(owner)
        self._related_fields = related_fields or []
        self._related_names = [field.name for field in self._related_fields]
        self._relations: Dict[str, Relation] = dict()
        for field in self._related_fields:
            self._add_relation(field)

    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._relations.get(name, None)
        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.Meta.model_fields[name].get_related_name()
        item._orm.remove(name, parent)
        parent._orm.remove(relation_name, item)

    def _get(self, name: str) -> Optional[Relation]:
        """
        Returns the actual relation and not the related model(s).

        :param name: name of the relation
        :type name: str
        :return: Relation instance
        :rtype: ormar.relations.relation.Relation
        """
        relation = self._relations.get(name, None)
        if relation is not None:
            return relation
        return None

    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

    def _add_relation(self, field: "BaseField") -> None:
        """
        Registers relation in the manager.
        Adds Relation instance under field.name.

        :param field: field with relation declaration
        :type field: BaseField
        """
        self._relations[field.name] = Relation(
            manager=self,
            type_=self._get_relation_type(field),
            field_name=field.name,
            to=field.to,
            through=getattr(field, "through", None),
        )

__contains__(item)

Checks if relation with given name is already registered.

Parameters:

Name Type Description Default
item str

name of attribute

 required

Returns:

Type Description
bool

result of the check
Source code in ormar\relations\relation_manager.py 
29
30
31
32
33
34
35
36
37
38
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.

Parameters:

Name Type Description Default
parent Model

parent model on which relation should be registered

 required
child Model

child model to register

 required
field ForeignKeyField

field with relation definition

 required
Source code in ormar\relations\relation_manager.py 
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
@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.

Parameters:

Name Type Description Default
name str

name of the relation

 required

Returns:

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

related model or list of related models if set
Source code in ormar\relations\relation_manager.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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._relations.get(name, None)
    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.

Parameters:

Name Type Description Default
name str

name of the relation

 required
child Union[NewBaseModel, Type[NewBaseModel]]

child to remove from relation

 required
Source code in ormar\relations\relation_manager.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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.

Parameters:

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

model with parent registered

 required
parent Model

parent Model

 required
name str

name of the relation

 required
Source code in ormar\relations\relation_manager.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@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.Meta.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.

Parameters:

Name Type Description Default
to_field ForeignKeyField

field with relation definition

 required
parent Model

parent model

 required
child Model

child model

 required

Returns:

Type Description
Tuple["Model", "Model", str, str]

parent, child, child_name, to_name
Source code in ormar\relations\utils.py 
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