Skip to content

sqlalchemy

_is_db_field(field)

Alias to if check that verifies if field should be included in database.

Parameters:

Name Type Description Default
field 'BaseField'

field to check

required

Returns:

Type Description
bool

result of the check

Source code in ormar/models/helpers/sqlalchemy.py
210
211
212
213
214
215
216
217
218
def _is_db_field(field: "BaseField") -> bool:
    """
    Alias to if check that verifies if field should be included in database.
    :param field: field to check
    :type field: "BaseField"
    :return: result of the check
    :rtype: bool
    """
    return not field.pydantic_only and not field.virtual and not field.is_multi

_is_through_model_not_set(field)

Alias to if check that verifies if through model was created.

Parameters:

Name Type Description Default
field 'BaseField'

field to check

required

Returns:

Type Description
bool

result of the check

Source code in ormar/models/helpers/sqlalchemy.py
199
200
201
202
203
204
205
206
207
def _is_through_model_not_set(field: "BaseField") -> bool:
    """
    Alias to if check that verifies if through model was created.
    :param field: field to check
    :type field: "BaseField"
    :return: result of the check
    :rtype: bool
    """
    return field.is_multi and not field.through and not field.to.__class__ == ForwardRef

_process_fields(model_fields, new_model)

Helper method.

Populates pkname and columns. Trigger validation of primary_key - only one and required pk can be set, cannot be pydantic_only.

Append fields to columns if it's not pydantic_only, virtual ForeignKey or ManyToMany field.

Sets owner on each model_field as reference to newly created Model.

Parameters:

Name Type Description Default
model_fields Dict

dictionary of declared ormar model fields

required
new_model Type['Model'] required

Returns:

Type Description
Tuple[Optional[str], List[sqlalchemy.Column]]

pkname, list of sqlalchemy columns

Raises:

Type Description
ModelDefinitionError

if validation of related_names fail, or pkname validation fails.

Source code in ormar/models/helpers/sqlalchemy.py
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
def _process_fields(
    model_fields: Dict, new_model: Type["Model"]
) -> Tuple[Optional[str], List[sqlalchemy.Column]]:
    """
    Helper method.

    Populates pkname and columns.
    Trigger validation of primary_key - only one and required pk can be set,
    cannot be pydantic_only.

    Append fields to columns if it's not pydantic_only,
    virtual ForeignKey or ManyToMany field.

    Sets `owner` on each model_field as reference to newly created Model.

    :raises ModelDefinitionError: if validation of related_names fail,
    or pkname validation fails.
    :param model_fields: dictionary of declared ormar model fields
    :type model_fields: Dict[str, ormar.Field]
    :param new_model:
    :type new_model: Model class
    :return: pkname, list of sqlalchemy columns
    :rtype: Tuple[Optional[str], List[sqlalchemy.Column]]
    """
    columns = []
    pkname = None
    for field_name, field in model_fields.items():
        field.owner = new_model
        if _is_through_model_not_set(field):
            field.create_default_through_model()
        if field.primary_key:
            pkname = check_pk_column_validity(field_name, field, pkname)
        if _is_db_field(field):
            columns.append(field.get_column(field.get_alias()))
    return pkname, columns

adjust_through_many_to_many_model(model_field)

Registers m2m relation on through model. Sets ormar.ForeignKey from through model to both child and parent models. Sets sqlalchemy.ForeignKey to both child and parent models. Sets pydantic fields with child and parent model types.

Parameters:

Name Type Description Default
model_field 'ManyToManyField'

relation field defined in parent model

required
Source code in ormar/models/helpers/sqlalchemy.py
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
def adjust_through_many_to_many_model(model_field: "ManyToManyField") -> None:
    """
    Registers m2m relation on through model.
    Sets ormar.ForeignKey from through model to both child and parent models.
    Sets sqlalchemy.ForeignKey to both child and parent models.
    Sets pydantic fields with child and parent model types.

    :param model_field: relation field defined in parent model
    :type model_field: ManyToManyField
    """
    parent_name = model_field.default_target_field_name()
    child_name = model_field.default_source_field_name()
    model_fields = model_field.through.Meta.model_fields
    model_fields[parent_name] = ormar.ForeignKey(  # type: ignore
        model_field.to,
        real_name=parent_name,
        ondelete="CASCADE",
        owner=model_field.through,
    )

    model_fields[child_name] = ormar.ForeignKey(  # type: ignore
        model_field.owner,
        real_name=child_name,
        ondelete="CASCADE",
        owner=model_field.through,
    )

    create_and_append_m2m_fk(
        model=model_field.to, model_field=model_field, field_name=parent_name
    )
    create_and_append_m2m_fk(
        model=model_field.owner, model_field=model_field, field_name=child_name
    )

    create_pydantic_field(parent_name, model_field.to, model_field)
    create_pydantic_field(child_name, model_field.owner, model_field)

    setattr(model_field.through, parent_name, RelationDescriptor(name=parent_name))
    setattr(model_field.through, child_name, RelationDescriptor(name=child_name))

check_for_null_type_columns_from_forward_refs(meta)

Check is any column is of NUllType() meaning it's empty column from ForwardRef

Parameters:

Name Type Description Default
meta 'ModelMeta'

Meta class of the Model without sqlalchemy table constructed

required

Returns:

Type Description
bool

result of the check

Source code in ormar/models/helpers/sqlalchemy.py
267
268
269
270
271
272
273
274
275
276
277
278
def check_for_null_type_columns_from_forward_refs(meta: "ModelMeta") -> bool:
    """
    Check is any column is of NUllType() meaning it's empty column from ForwardRef

    :param meta: Meta class of the Model without sqlalchemy table constructed
    :type meta: Model class Meta
    :return: result of the check
    :rtype: bool
    """
    return not any(
        isinstance(col.type, sqlalchemy.sql.sqltypes.NullType) for col in meta.columns
    )

check_pk_column_validity(field_name, field, pkname)

Receives the field marked as primary key and verifies if the pkname was not already set (only one allowed per model) and if field is not marked as pydantic_only as it needs to be a database field.

Parameters:

Name Type Description Default
field_name str

name of field

required
field 'BaseField'

ormar.Field

required
pkname Optional[str]

already set pkname

required

Returns:

Type Description
str

name of the field that should be set as pkname

Raises:

Type Description
ModelDefintionError

if pkname already set or field is pydantic_only

Source code in ormar/models/helpers/sqlalchemy.py
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def check_pk_column_validity(
    field_name: str, field: "BaseField", pkname: Optional[str]
) -> Optional[str]:
    """
    Receives the field marked as primary key and verifies if the pkname
    was not already set (only one allowed per model) and if field is not marked
    as pydantic_only as it needs to be a database field.

    :raises ModelDefintionError: if pkname already set or field is pydantic_only
    :param field_name: name of field
    :type field_name: str
    :param field: ormar.Field
    :type field: BaseField
    :param pkname: already set pkname
    :type pkname: Optional[str]
    :return: name of the field that should be set as pkname
    :rtype: str
    """
    if pkname is not None:
        raise ormar.ModelDefinitionError("Only one primary key column is allowed.")
    if field.pydantic_only:
        raise ormar.ModelDefinitionError("Primary key column cannot be pydantic only")
    return field_name

create_and_append_m2m_fk(model, model_field, field_name)

Registers sqlalchemy Column with sqlalchemy.ForeignKey leading to the model.

Newly created field is added to m2m relation through model Meta columns and table.

Parameters:

Name Type Description Default
field_name str

name of the column to create

required
model Type['Model']

Model class to which FK should be created

required
model_field 'ManyToManyField'

field with ManyToMany relation

required
Source code in ormar/models/helpers/sqlalchemy.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
92
93
def create_and_append_m2m_fk(
    model: Type["Model"], model_field: "ManyToManyField", field_name: str
) -> None:
    """
    Registers sqlalchemy Column with sqlalchemy.ForeignKey leading to the model.

    Newly created field is added to m2m relation through model Meta columns and table.

    :param field_name: name of the column to create
    :type field_name: str
    :param model: Model class to which FK should be created
    :type model: Model class
    :param model_field: field with ManyToMany relation
    :type model_field: ManyToManyField field
    """
    pk_alias = model.get_column_alias(model.Meta.pkname)
    pk_column = next((col for col in model.Meta.columns if col.name == pk_alias), None)
    if pk_column is None:  # pragma: no cover
        raise ormar.ModelDefinitionError(
            "ManyToMany relation cannot lead to field without pk"
        )
    column = sqlalchemy.Column(
        field_name,
        pk_column.type,
        sqlalchemy.schema.ForeignKey(
            model.Meta.tablename + "." + pk_alias,
            ondelete="CASCADE",
            onupdate="CASCADE",
            name=f"fk_{model_field.through.Meta.tablename}_{model.Meta.tablename}"
            f"_{field_name}_{pk_alias}",
        ),
    )
    model_field.through.Meta.columns.append(column)
    model_field.through.Meta.table.append_column(column)

populate_meta_sqlalchemy_table_if_required(meta)

Constructs sqlalchemy table out of columns and parameters set on Meta class. It populates name, metadata, columns and constraints.

Parameters:

Name Type Description Default
meta 'ModelMeta'

Meta class of the Model without sqlalchemy table constructed

required
Source code in ormar/models/helpers/sqlalchemy.py
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def populate_meta_sqlalchemy_table_if_required(meta: "ModelMeta") -> None:
    """
    Constructs sqlalchemy table out of columns and parameters set on Meta class.
    It populates name, metadata, columns and constraints.

    :param meta: Meta class of the Model without sqlalchemy table constructed
    :type meta: Model class Meta
    """
    if not hasattr(meta, "table") and check_for_null_type_columns_from_forward_refs(
        meta
    ):
        set_constraint_names(meta=meta)
        table = sqlalchemy.Table(
            meta.tablename, meta.metadata, *meta.columns, *meta.constraints
        )
        meta.table = table

populate_meta_tablename_columns_and_pk(name, new_model)

Sets Model tablename if it's not already set in Meta. Default tablename if not present is class name lower + s (i.e. Bed becomes -> beds)

Checks if Model's Meta have pkname and columns set. If not calls the sqlalchemy_columns_from_model_fields to populate columns from ormar.fields definitions.

Parameters:

Name Type Description Default
name str

name of the current Model

required
new_model Type['Model']

currently constructed Model

required

Returns:

Type Description
ormar.models.metaclass.ModelMetaclass

Model with populated pkname and columns in Meta

Raises:

Type Description
ModelDefinitionError

if pkname is not present raises ModelDefinitionError. Each model has to have pk.

Source code in ormar/models/helpers/sqlalchemy.py
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def populate_meta_tablename_columns_and_pk(
    name: str, new_model: Type["Model"]
) -> Type["Model"]:
    """
    Sets Model tablename if it's not already set in Meta.
    Default tablename if not present is class name lower + s (i.e. Bed becomes -> beds)

    Checks if Model's Meta have pkname and columns set.
    If not calls the sqlalchemy_columns_from_model_fields to populate
    columns from ormar.fields definitions.

    :raises ModelDefinitionError: if pkname is not present raises ModelDefinitionError.
    Each model has to have pk.

    :param name: name of the current Model
    :type name: str
    :param new_model: currently constructed Model
    :type new_model: ormar.models.metaclass.ModelMetaclass
    :return: Model with populated pkname and columns in Meta
    :rtype: ormar.models.metaclass.ModelMetaclass
    """
    tablename = name.lower() + "s"
    new_model.Meta.tablename = (
        new_model.Meta.tablename if hasattr(new_model.Meta, "tablename") else tablename
    )
    pkname: Optional[str]

    if hasattr(new_model.Meta, "columns"):
        columns = new_model.Meta.columns
        pkname = new_model.Meta.pkname
    else:
        pkname, columns = sqlalchemy_columns_from_model_fields(
            new_model.Meta.model_fields, new_model
        )

    if pkname is None:
        raise ormar.ModelDefinitionError("Table has to have a primary key.")

    new_model.Meta.columns = columns
    new_model.Meta.pkname = pkname
    if not new_model.Meta.orders_by:
        # by default we sort by pk name if other option not provided
        new_model.Meta.orders_by.append(pkname)
    return new_model

set_constraint_names(meta)

Populates the names on IndexColumn and UniqueColumns constraints.

Parameters:

Name Type Description Default
meta 'ModelMeta'

Meta class of the Model without sqlalchemy table constructed

required
Source code in ormar/models/helpers/sqlalchemy.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def set_constraint_names(meta: "ModelMeta") -> None:
    """
    Populates the names on IndexColumn and UniqueColumns constraints.

    :param meta: Meta class of the Model without sqlalchemy table constructed
    :type meta: Model class Meta
    """
    for constraint in meta.constraints:
        if isinstance(constraint, sqlalchemy.UniqueConstraint) and not constraint.name:
            constraint.name = (
                f"uc_{meta.tablename}_"
                f'{"_".join([str(col) for col in constraint._pending_colargs])}'
            )
        elif (
            isinstance(constraint, sqlalchemy.Index)
            and constraint.name == "TEMPORARY_NAME"
        ):
            constraint.name = (
                f"ix_{meta.tablename}_"
                f'{"_".join([col for col in constraint._pending_colargs])}'
            )

sqlalchemy_columns_from_model_fields(model_fields, new_model)

Iterates over declared on Model model fields and extracts fields that should be treated as database fields.

If the model is empty it sets mandatory id field as primary key (used in through models in m2m relations).

Triggers a validation of relation_names in relation fields. If multiple fields are leading to the same related model only one can have empty related_name param. Also related_names have to be unique.

Trigger validation of primary_key - only one and required pk can be set, cannot be pydantic_only.

Append fields to columns if it's not pydantic_only, virtual ForeignKey or ManyToMany field.

Sets owner on each model_field as reference to newly created Model.

Parameters:

Name Type Description Default
model_fields Dict

dictionary of declared ormar model fields

required
new_model Type['Model'] required

Returns:

Type Description
Tuple[Optional[str], List[sqlalchemy.Column]]

pkname, list of sqlalchemy columns

Raises:

Type Description
ModelDefinitionError

if validation of related_names fail, or pkname validation fails.

Source code in ormar/models/helpers/sqlalchemy.py
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
def sqlalchemy_columns_from_model_fields(
    model_fields: Dict, new_model: Type["Model"]
) -> Tuple[Optional[str], List[sqlalchemy.Column]]:
    """
    Iterates over declared on Model model fields and extracts fields that
    should be treated as database fields.

    If the model is empty it sets mandatory id field as primary key
    (used in through models in m2m relations).

    Triggers a validation of relation_names in relation fields. If multiple fields
    are leading to the same related model only one can have empty related_name param.
    Also related_names have to be unique.

    Trigger validation of primary_key - only one and required pk can be set,
    cannot be pydantic_only.

    Append fields to columns if it's not pydantic_only,
    virtual ForeignKey or ManyToMany field.

    Sets `owner` on each model_field as reference to newly created Model.

    :raises ModelDefinitionError: if validation of related_names fail,
    or pkname validation fails.
    :param model_fields: dictionary of declared ormar model fields
    :type model_fields: Dict[str, ormar.Field]
    :param new_model:
    :type new_model: Model class
    :return: pkname, list of sqlalchemy columns
    :rtype: Tuple[Optional[str], List[sqlalchemy.Column]]
    """
    if len(model_fields.keys()) == 0:
        model_fields["id"] = ormar.Integer(name="id", primary_key=True)
        logging.warning(
            f"Table {new_model.Meta.tablename} had no fields so auto "
            "Integer primary key named `id` created."
        )
    validate_related_names_in_relations(model_fields, new_model)
    return _process_fields(model_fields=model_fields, new_model=new_model)

update_column_definition(model, field)

Updates a column with a new type column based on updated parameters in FK fields.

Parameters:

Name Type Description Default
model Union[Type['Model'], Type['NewBaseModel']]

model on which columns needs to be updated

required
field 'ForeignKeyField'

field with column definition that requires update

required

Returns:

Type Description
None

None

Source code in ormar/models/helpers/sqlalchemy.py
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def update_column_definition(
    model: Union[Type["Model"], Type["NewBaseModel"]], field: "ForeignKeyField"
) -> None:
    """
    Updates a column with a new type column based on updated parameters in FK fields.

    :param model: model on which columns needs to be updated
    :type model: Type["Model"]
    :param field: field with column definition that requires update
    :type field: ForeignKeyField
    :return: None
    :rtype: None
    """
    columns = model.Meta.columns
    for ind, column in enumerate(columns):
        if column.name == field.get_alias():
            new_column = field.get_column(field.get_alias())
            columns[ind] = new_column
            break