Coverage for src / mafw / db / db_model.py: 91%
216 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-30 16:10 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5The module provides functionality to MAFw to interface to a DB.
6"""
8import warnings
9from collections.abc import Mapping, Sequence
10from itertools import tee
11from typing import TYPE_CHECKING, Any, Iterable, cast
13# mypy is complaining that peewee does not have make_snake_case.
14from peewee import ( # type: ignore[attr-defined]
15 EXCLUDED,
16 SQL,
17 Database,
18 DatabaseProxy,
19 Field,
20 ForeignKeyField,
21 ModelBase,
22 ModelInsert,
23 MySQLDatabase,
24 Select,
25 Value,
26 fn,
27 make_snake_case,
28)
29from playhouse.shortcuts import ThreadSafeDatabaseMetadata, dict_to_model, model_to_dict, update_model_from_dict
31# noinspection PyUnresolvedReferences
32from playhouse.signals import Model
34from mafw.db import trigger
35from mafw.db.db_types import PeeweeModelWithMeta
36from mafw.db.fields import FileNameField
37from mafw.db.model_register import ModelRegister
38from mafw.db.trigger import Trigger
39from mafw.mafw_errors import MAFwException, UnsupportedDatabaseError
41database_proxy = DatabaseProxy()
42"""This is a placeholder for the real database object that will be known only at run time"""
45mafw_model_register = ModelRegister()
46"""
47This is the instance of the ModelRegister
49 .. seealso::
51 :class:`.ModelRegister` for more information on how to retrieve models and :class:`.RegisteredMeta` and :class:`MAFwBaseModel` for the automatic registration of models`
53"""
56class RegisteredMeta(ModelBase):
57 """
58 Metaclass for registering models with the MAFw model registry.
60 This metaclass automatically registers model classes with the global model registry
61 when they are defined, allowing for dynamic discovery and management of database models.
62 It ensures that only concrete model classes (not the base classes themselves) are registered.
64 The registration process uses the table name from the model's metadata or generates
65 a snake_case version of the class name if no explicit table name is set.
66 """
68 reserved_field_names = ['__logic__', '__conditional__']
70 def __new__(cls, name: str, bases: tuple[type, ...], attrs: dict[str, Any], **kwargs: dict[str, Any]) -> type:
71 """
72 Create a new model class and register it if applicable.
74 This method is called during class definition to create the actual class
75 and register it with the MAFw model registry if it's a concrete model class.
77 :param name: The name of the class being created.
78 :type name: str
79 :param bases: The base classes of the class being created.
80 :type bases: tuple
81 :param attrs: The attributes of the class being created.
82 :type attrs: dict
83 :param kwargs: Other keyword attributes passed to the class.
84 :type kwargs: dict[str, Any]
85 :return: The newly created class.
86 :rtype: type
87 """
88 # store here any extra argument that is passed to the class defition.
89 extra_args = kwargs
91 meta = attrs.get('Meta', None)
92 if meta:
93 custom_table_name = meta.__dict__.get('table_name', None)
94 else:
95 custom_table_name = None
96 model = cast(RegisteredMeta, super().__new__(cls, name, bases, attrs)) # type: ignore[no-untyped-call]
98 if TYPE_CHECKING:
99 assert hasattr(model, '_meta')
101 # for consistency, add the extra_args to the meta instance
102 model._meta.extra_args = extra_args
104 # check for the use of reserved field names.
105 field_names = model._meta.sorted_field_names
106 if any([reserved_name in field_names for reserved_name in RegisteredMeta.reserved_field_names]):
107 warnings.warn(
108 f'Model {model.__name__} is using a reserved field name. Filter configuration might not '
109 f'work properly. Reserved names are f{RegisteredMeta.reserved_field_names}.'
110 )
112 # this super call is actually creating the Model class according to peewee ModelBase (that is a metaclass)
113 # the model class will have an attribute (_meta) created by the ModelBase using the inner class Meta as a template.
114 # The additional attributes (like suffix, prefix, ...) that we have added in the MAFwBaseModel Meta class will be
115 # available. There is one problem, that the _meta.table_name is generated with our make_prefixed_suffixed_table_name
116 # but before that the additional attributes are actually transferred to the instance of the meta class.
117 # for this reason we need to regenerate the table_name now, when the whole Meta class is available.
118 # We do so only if the user didn't specify a custom table name
119 if custom_table_name is None:
120 if model._meta.table_function is None:
121 table_name = make_snake_case(model.__name__)
122 else:
123 table_name = model._meta.table_function(model)
124 model._meta.table_name = table_name
126 # Do we need to register the Model?
127 # By default we want to register (so skip_registration = False)
128 skip_registration = extra_args.get('do_not_register', False)
129 if skip_registration:
130 # no need to register, so just skip the registration.
131 return model
133 # Register only concrete model classes that are not base classes
134 # Check that we're not processing base classes themselves
135 is_base_class = name in ['Model', 'MAFwBaseModel', 'StandardTable']
137 # Check that we have valid bases and that at least one inherits from Model
138 has_valid_bases = bases and any(issubclass(base, Model) for base in bases)
140 if not is_base_class and has_valid_bases:
141 table_name = model._meta.table_name
142 mafw_model_register.register_model(table_name, model)
144 if hasattr(model._meta, 'suffix'): 144 ↛ 147line 144 didn't jump to line 147 because the condition on line 144 was always true
145 mafw_model_register.register_suffix(model._meta.suffix)
147 if hasattr(model._meta, 'prefix'): 147 ↛ 150line 147 didn't jump to line 150 because the condition on line 147 was always true
148 mafw_model_register.register_prefix(model._meta.prefix)
150 return model
153class MAFwBaseModelDoesNotExist(MAFwException):
154 """Raised when the base model class is not existing."""
157def get_db(model_class: PeeweeModelWithMeta) -> Database:
158 """
159 Extract the actual database instance from a model class.
161 This function retrieves the database object from a model class's metadata,
162 handling the case where the database is a :class:`peewee.DatabaseProxy`. If the
163 database is a proxy, it extracts the underlying database object.
165 :param model_class: The model class from which to extract the database.
166 :type model_class: PeeweeModelWithMeta
167 :return: The actual database instance.
168 :rtype: Database
169 """
170 db = model_class._meta.database
171 if isinstance(db, DatabaseProxy): 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 db = db.obj
173 return cast(Database, db)
176def get_primary_key_fields(model_class: PeeweeModelWithMeta) -> tuple[list[str], list[Field]]:
177 """
178 Extract primary key field names and field objects from a model metadata class.
180 This function retrieves the primary key information from a Peewee model metadata
181 class, handling both single-column primary keys and composite (multi-column) keys.
182 It returns two parallel lists: the first containing the field names as strings,
183 and the second containing the corresponding field objects.
185 :param model_class: The model class from which to extract primary key information.
186 :type model_class: RegisteredMeta
187 :return: A tuple containing two lists: the first with primary key field names (as strings)
188 and the second with the corresponding field objects.
189 :rtype: tuple[list[str], list[Field]]
190 """
191 if model_class._meta.composite_key:
192 pk_fields = [model_class._meta.fields[name] for name in model_class._meta.primary_key.field_names]
193 else:
194 pk_fields = [model_class._meta.primary_key]
196 return [f.name for f in pk_fields], pk_fields
199def make_prefixed_suffixed_name(model_class: RegisteredMeta) -> str:
200 """
201 Generate a table name with optional prefix and suffix for a given model class.
203 This function constructs a table name by combining the prefix, the snake_case
204 version of the model class name, and the suffix. If either prefix or suffix
205 are not defined in the model's metadata, empty strings are used instead.
207 The prefix, table name, and suffix are joined using underscores. For example:
209 - If a model class is named "UserAccount" with prefix="app", suffix="data",
210 the resulting table name will be "app_user_account_data"
212 - If a model class is named "Product" with prefix="ecommerce", suffix="_latest",
213 the resulting table name will be "ecommerce_product_latest"
215 .. note::
217 Underscores (_) will be automatically added to prefix and suffix if not already present.
219 :param model_class: The model class for which to generate the table name.
220 :type model_class: RegisteredMeta
221 :return: The constructed table name including prefix and suffix if applicable.
222 :rtype: str
223 """
224 if TYPE_CHECKING:
225 assert hasattr(model_class, '_meta')
227 if hasattr(model_class._meta, 'suffix') and model_class._meta.suffix is not None:
228 suffix = model_class._meta.suffix
229 else:
230 suffix = ''
232 if hasattr(model_class._meta, 'prefix') and model_class._meta.prefix is not None:
233 prefix = model_class._meta.prefix
234 else:
235 prefix = ''
237 if not suffix.startswith('_') and suffix != '':
238 suffix = '_' + suffix
240 if not prefix.endswith('_') and prefix != '':
241 prefix = prefix + '_'
243 return f'{prefix}{make_snake_case(model_class.__name__)}{suffix}'
246class MAFwBaseModel(Model, metaclass=RegisteredMeta):
247 """The base model for the MAFw library.
249 Every model class (table) that the user wants to interface must inherit from this base.
251 This class extends peewee's Model with several additional features:
253 1. Automatic model registration: Models are automatically registered with the MAFw model registry
254 during class definition, enabling dynamic discovery and management of database models.
256 2. Trigger support: The class supports defining database triggers through the :meth:`.triggers` method,
257 which are automatically created when the table is created. File removal triggers can also be automatically
258 generated using the `file_trigger_auto_create` boolean flag in the :ref:`meta class <auto_triggers>`. See also
259 :meth:`.file_removal_triggers`.
261 3. Standard upsert operations: Provides :meth:`.std_upsert` and :meth:`.std_upsert_many` methods for
262 performing upsert operations that work with SQLite and PostgreSQL.
264 4. Dictionary conversion utilities: Includes :meth:`.to_dict`, :meth:`.from_dict`, and :meth:`.update_from_dict`
265 methods for easy serialization and deserialization of model instances.
267 5. Customizable table naming: Supports table name prefixes and suffixes through the Meta class
268 with `prefix` and `suffix` attributes. See :func:`.make_prefixed_suffixed_name`.
270 6. Automatic table creation control: The `automatic_creation` Meta attribute controls whether
271 tables are automatically created when the application starts.
273 .. note::
275 The automatic model registration can be disabled for one single model class using the keyword argument
276 `do_not_register` passed to the :class:`.RegisteredMeta` meta-class. For example:
278 .. code-block:: python
280 class AutoRegisterModel(MAFwBaseModel):
281 pass
284 class NoRegisterModel(MAFwBaseModel, do_not_register=True):
285 pass
287 the first class will be automatically registered, while the second one will not. This is particularly useful if
288 the user wants to define a base model class for the whole project without having it in the register where
289 only concrete Model implementations are stored.
291 """
293 @classmethod
294 def get_fields_by_type(cls, field_type: type[Field]) -> dict[str, Field]:
295 """
296 Return a dict {field_name: field_object} for all fields of the given type.
298 .. versionadded:: v2.0.0
300 :param field_type: Field type
301 :type field_type: peewee.Field
302 :return: A dict {field_name: field_object} for all fields of the given type.
303 :rtype: dict[str, peewee.Field]
304 """
305 if TYPE_CHECKING:
306 assert hasattr(cls, '_meta')
308 return {name: field for name, field in cls._meta.fields.items() if isinstance(field, field_type)}
310 @classmethod
311 def file_removal_triggers(cls) -> list[Trigger]:
312 """
313 Generate a list of triggers for automatic file removal when records are deleted.
315 This method creates database triggers that automatically handle file cleanup when
316 records containing :class:`~mafw.db.fields.FileNameField` fields are removed from
317 the database table. The triggers insert the filenames and checksums into the
318 :class:`~mafw.db.std_tables.OrphanFile` table for later processing.
320 The triggers are only created if the model has at least one field of type
321 :class:`~mafw.db.fields.FileNameField`. If no such fields exist, an empty list
322 is returned.
324 :class:`.FileNameListField` is a subclass of :class:`.FileNameField` and is treated in the same
325 way.
327 .. versionadded:: v2.0.0
329 .. note::
331 This functionality requires the ``file_trigger_auto_create`` attribute in the
332 model's Meta class to be set to ``True`` for automatic trigger creation.
334 :return: A list containing the trigger object for file removal, or an empty list
335 if no :class:`~mafw.db.fields.FileNameField` fields are found.
336 :rtype: list[:class:`~mafw.db.trigger.Trigger`]
337 """
338 from mafw.db.std_tables import OrphanFile, TriggerStatus
340 # it includes also FileNameListField that is a subclass of FileNameField
341 file_fields = cls.get_fields_by_type(FileNameField)
343 if len(file_fields) == 0:
344 return []
346 if TYPE_CHECKING:
347 assert hasattr(cls, '_meta')
349 new_trigger = Trigger(
350 trigger_name=cls._meta.table_name + '_delete_files',
351 trigger_type=(trigger.TriggerWhen.Before, trigger.TriggerAction.Delete),
352 source_table=cls,
353 safe=True,
354 for_each_row=True,
355 )
356 sub_query = TriggerStatus.select(TriggerStatus.status).where(TriggerStatus.trigger_type == 'DELETE_FILES')
357 trigger_condition = Value(1) == sub_query
358 new_trigger.add_when(trigger_condition)
359 data = []
360 for f in file_fields:
361 c = cast(FileNameField, file_fields[f]).checksum_field or f
362 # the checksum is not really used in the OrphanFile.
363 # in versions before 2.0.0, the checksum field in OrphanFile was not nullable,
364 # so it should contain something. In order to be backward compatible, in case of a missing checksum field
365 # we will just use the filename
366 # note that the automatic filename to checksum conversion will not work in this case because the trigger
367 # lives in the database and not in the application.
368 data.append({'filenames': SQL(f'OLD.{f}'), 'checksum': SQL(f'OLD.{c}')})
369 insert_query = OrphanFile.insert_many(data)
370 new_trigger.add_sql(insert_query)
372 return [new_trigger]
374 @classmethod
375 def triggers(cls) -> list[Trigger]:
376 """
377 Returns an iterable of :class:`~mafw.db.trigger.Trigger` objects to create upon table creation.
379 The user must overload this returning all the triggers that must be created along with this class.
380 """
381 return []
383 # noinspection PyUnresolvedReferences
384 @classmethod
385 def create_table(cls, safe: bool = True, **options: Any) -> None:
386 """
387 Create the table in the underlying DB and all the related trigger as well.
389 If the creation of a trigger fails, then the whole table dropped, and the original exception is re-raised.
391 .. warning::
393 Trigger creation has been extensively tested with :link:`SQLite`, but not with the other database implementation.
394 Please report any malfunction.
396 :param safe: Flag to add an IF NOT EXISTS to the creation statement. Defaults to True.
397 :type safe: bool, Optional
398 :param options: Additional options passed to the super method.
399 """
400 super().create_table(safe, **options)
402 # this is just use to make mypy happy.
403 meta_cls = cast(PeeweeModelWithMeta, cls)
405 # Get the database instance, it is used for trigger creation
406 db = meta_cls._meta.database
408 triggers_list = cls.triggers()
410 if meta_cls._meta.file_trigger_auto_create:
411 triggers_list.extend(cls.file_removal_triggers())
413 if len(triggers_list):
414 # Create tables with appropriate error handling
415 try:
416 for trigger in triggers_list:
417 trigger.set_database(db)
418 try:
419 db.execute_sql(trigger.create())
420 except UnsupportedDatabaseError as e:
421 warnings.warn(f'Skipping unsupported trigger {trigger.trigger_name}: {str(e)}')
422 except Exception:
423 raise
424 except:
425 # If an error occurs, drop the table and any created triggers
426 meta_cls._meta.database.drop_tables([cls], safe=safe)
427 for trigger in triggers_list:
428 try:
429 db.execute_sql(trigger.drop(True))
430 except Exception:
431 pass # Ignore errors when dropping triggers during cleanup
432 raise
434 @classmethod
435 def std_upsert(cls, __data: dict[str, Any] | None = None, **mapping: Any) -> ModelInsert:
436 """
437 Perform a so-called standard upsert.
439 An upsert statement is not part of the standard SQL and different databases have different ways to implement it.
440 This method will work for the three DB backends generating a valid query for the three dialects.
442 An upsert is a statement in which we try to insert some data in a table where there are some constraints.
443 If one constraint is failing, then instead of inserting a new row, we will try to update the existing row
444 causing the constraint violation.
446 A standard upsert, in the naming convention of MAFw, is setting the conflict cause to the primary key with all
447 other fields being updated. In other words, the database will try to insert the data provided in the table, but
448 if the primary key already exists, then all other fields will be updated.
450 This method is equivalent to the following:
452 .. code-block:: python
454 class Sample(MAFwBaseModel):
455 sample_id = AutoField(
456 primary_key=True,
457 help_text='The sample id primary key',
458 )
459 sample_name = TextField(help_text='The sample name')
462 (
463 Sample.insert(sample_id=1, sample_name='my_sample')
464 .on_conflict(
465 preserve=[Sample.sample_name]
466 ) # use the value we would have inserted
467 .execute()
468 )
470 :param __data: A dictionary containing the key/value pair for the insert. The key is the column name.
471 Defaults to None
472 :type __data: dict, Optional
473 :param mapping: Keyword arguments representing the value to be inserted.
475 """
477 meta_cls = cast('PeeweeModelWithMeta', cls)
478 db = get_db(meta_cls)
480 # Normalize input
481 if __data is not None:
482 data = dict(__data)
483 data.update(mapping)
484 else:
485 data = dict(mapping)
487 # --- PK detection ---
488 pk_names, pk_fields = get_primary_key_fields(meta_cls)
490 # --- Build insert ---
491 insert_query = cls.insert(data)
493 # --- Backend-specific update mapping ---
494 if isinstance(db, MySQLDatabase): 494 ↛ 496line 494 didn't jump to line 496 because the condition on line 494 was never true
495 # MySQL: VALUES(col)
496 update_mapping = {
497 meta_cls._meta.fields[name]: fn.VALUES(meta_cls._meta.fields[name])
498 for name in data
499 if name not in pk_names
500 }
501 q = insert_query.on_conflict(update=update_mapping)
502 return q
504 else:
505 # PostgreSQL / SQLite: EXCLUDED.col
506 update_mapping = {
507 meta_cls._meta.fields[name]: getattr(EXCLUDED, name) for name in data if name not in pk_names
508 }
510 q = insert_query.on_conflict(conflict_target=pk_fields, update=update_mapping)
511 return q
513 # noinspection PyProtectedMember
514 @classmethod
515 def std_upsert_many(cls, rows: Iterable[Any], fields: list[str] | None = None) -> ModelInsert:
516 """
517 Perform a standard upsert with many rows.
519 .. seealso::
521 Read the :meth:`std_upsert` documentation for an explanation of this method.
523 :param rows: A list with the rows to be inserted. Each item can be a dictionary or a tuple of values. If a
524 tuple is provided, then the `fields` must be provided.
525 :type rows: Iterable
526 :param fields: A list of field names. Defaults to None.
527 :type fields: list[str], Optional
528 """
529 meta_cls = cast(PeeweeModelWithMeta, cls)
530 db = get_db(meta_cls)
532 # Initialise the output iterator and field names list.
533 # rows_to_insert will hold the actual data to insert (possibly split from rows for inspection)
534 # insert_field_names will hold the column names (either provided or inferred from first row)
535 rows_to_insert: Iterable[Any] = rows
536 insert_field_names: list[str] = list(fields) if fields is not None else []
538 # If fields are not explicitly provided, we need to infer them from the input data.
539 # This branch handles the case where rows is a Sequence (list, tuple) that supports indexing.
540 if fields is None:
541 if isinstance(rows, Sequence) and not isinstance(rows, (str, bytes)): 541 ↛ 552line 541 didn't jump to line 552 because the condition on line 541 was always true
542 # Sequence types (list, tuple) allow direct indexing to inspect the first row.
543 if len(rows) > 0: 543 ↛ 560line 543 didn't jump to line 560 because the condition on line 543 was always true
544 first_row = rows[0]
545 # If the first row is a dictionary (Mapping), extract its keys as field names.
546 if isinstance(first_row, Mapping): 546 ↛ 560line 546 didn't jump to line 560 because the condition on line 546 was always true
547 insert_field_names = list(first_row.keys())
548 else:
549 # For non-Sequence iterables (generators, etc.), we need to use tee() to inspect
550 # the first element without consuming the iterator. This creates two iterators:
551 # rows_inspect for looking at the first row, rows_to_insert for actual insertion.
552 rows_inspect, rows_to_insert = tee(rows)
553 first_row = next(rows_inspect, None)
554 if isinstance(first_row, Mapping):
555 insert_field_names = list(first_row.keys())
557 # Retrieve primary key field information. pk_names is used to exclude PK fields from
558 # the update operation (PK should not be updated during upsert), while pk_fields is
559 # used as the conflict target for the database.
560 pk_names, pk_fields = get_primary_key_fields(meta_cls)
562 # Build the list of fields to update during the upsert.
563 # We need to:
564 # 1. Remove duplicates (seen_names set)
565 # 2. Exclude primary key fields (they're used for conflict detection, not update)
566 # 3. Verify the field actually exists in the model (field_obj may be None)
567 seen_names: set[str] = set()
568 update_fields: list[tuple[Field, str]] = []
569 for field_name in insert_field_names:
570 # Skip duplicate field names
571 if field_name in seen_names: 571 ↛ 572line 571 didn't jump to line 572 because the condition on line 571 was never true
572 continue
573 seen_names.add(field_name)
574 # Skip primary key fields - they identify the row, not data to update
575 if field_name in pk_names:
576 continue
577 # Get the actual field object from the model metadata
578 field_obj = meta_cls._meta.fields.get(field_name)
579 if field_obj is None: 579 ↛ 580line 579 didn't jump to line 580 because the condition on line 579 was never true
580 continue
581 update_fields.append((field_obj, field_name))
583 # Create the bulk insert query using insert_many
584 insert_query = cls.insert_many(rows_to_insert, fields)
586 # Define a helper for MySQL's VALUES() syntax.
587 # NOTE: VALUES() is deprecated in MySQL 8.0.20+, but still required by Peewee
588 def _mysql_value_expression(field: Field) -> Any:
589 return fn.VALUES(field)
591 # Handle database-specific on_conflict syntax:
592 # - MySQL uses VALUES(field_name) to reference the values being inserted
593 # - PostgreSQL and SQLite use EXCLUDED.field_name to reference rejected values
594 if isinstance(db, MySQLDatabase): 594 ↛ 595line 594 didn't jump to line 595 because the condition on line 594 was never true
595 update_mapping = {field: _mysql_value_expression(field) for field, _ in update_fields}
596 return insert_query.on_conflict(update=update_mapping)
598 # For PostgreSQL/SQLite, use EXCLUDED table to reference the rejected insert values
599 update_mapping = {field: getattr(EXCLUDED, field_name) for field, field_name in update_fields}
600 return insert_query.on_conflict(
601 conflict_target=pk_fields,
602 update=update_mapping,
603 )
605 def to_dict(
606 self,
607 recurse: bool = True,
608 backrefs: bool = False,
609 only: list[str] | None = None,
610 exclude: list[str] | None = None,
611 **kwargs: Any,
612 ) -> dict[str, Any]:
613 """
614 Convert model instance to dictionary with optional parameters
616 See full documentation directly on the `peewee documentation
617 <https://docs.peewee-orm.com/en/latest/peewee/playhouse.html#dict_to_model>`__.
619 :param recurse: If to recurse through foreign keys. Default to True.
620 :type recurse: bool, Optional
621 :param backrefs: If to include backrefs. Default to False.
622 :type backrefs: bool, Optional
623 :param only: A list of fields to be included. Defaults to None.
624 :type only: list[str], Optional
625 :param exclude: A list of fields to be excluded. Defaults to None.
626 :type exclude: list[str], Optional
627 :param kwargs: Other keyword arguments to be passed to peewee `playhouse shortcut <https://docs.peewee-orm.com/en/latest/peewee/playhouse.html#dict_to_model>`__.
628 :return: A dictionary containing the key/value of the model.
629 :rtype: dict[str, Any]
630 """
631 # the playhouse module of peewee is not typed.
632 return model_to_dict( # type: ignore[no-any-return]
633 self,
634 recurse=recurse,
635 backrefs=backrefs, # type: ignore[no-untyped-call]
636 only=only,
637 exclude=exclude,
638 **kwargs,
639 )
641 @classmethod
642 def from_dict(cls, data: dict[str, Any], ignore_unknown: bool = False) -> 'MAFwBaseModel':
643 """
644 Create a new model instance from dictionary
646 :param data: The dictionary containing the key/value pairs of the model.
647 :type data: dict[str, Any]
648 :param ignore_unknown: If unknown dictionary keys should be ignored.
649 :type ignore_unknown: bool
650 :return: A new model instance.
651 :rtype: MAFwBaseModel
652 """
653 # the playhouse module of peewee is not typed.
654 return dict_to_model(cls, data, ignore_unknown=ignore_unknown) # type: ignore[no-untyped-call,no-any-return]
656 def update_from_dict(self, data: dict[str, Any], ignore_unknown: bool = False) -> 'MAFwBaseModel':
657 """
658 Update current model instance from dictionary
660 The model instance is returned for daisy-chaining.
662 :param data: The dictionary containing the key/value pairs of the model.
663 :type data: dict[str, Any]
664 :param ignore_unknown: If unknown dictionary keys should be ignored.
665 :type ignore_unknown: bool
666 """
667 update_model_from_dict(self, data, ignore_unknown=ignore_unknown) # type: ignore[no-untyped-call]
668 return self
670 @classmethod
671 def join_dependent_models(cls, base_query: Select | None = None) -> Select:
672 """Return a query that joins every model referenced through foreign keys.
674 Each foreign-key relation defined on ``cls`` is joined and attached to the
675 resulting rows via an attribute named ``_<field name>`` so that callers have
676 the same convenient access that manual ``.join_from`` calls would have
677 provided.
679 :param base_query: Optional query to extend. If not provided a new ``Select``
680 is created that includes ``cls`` and the dependent models.
681 :type base_query: :class:`peewee.Select`, optional
682 :return: The query joining every dependent model.
683 :rtype: :class:`peewee.Select`
684 """
685 if TYPE_CHECKING:
686 assert hasattr(cls, '_meta')
688 foreign_keys = [field for field in cls._meta.fields.values() if isinstance(field, ForeignKeyField)]
690 if len(foreign_keys) == 0:
691 return base_query or cls.select(cls)
693 dependent_models: list[type[ModelBase]] = []
694 for fk in foreign_keys:
695 rel_model = fk.rel_model
696 if rel_model not in dependent_models: 696 ↛ 694line 696 didn't jump to line 694 because the condition on line 696 was always true
697 dependent_models.append(rel_model)
699 query: Select = base_query or cls.select(cls, *dependent_models)
701 for idx, fk in enumerate(foreign_keys):
702 query = query.join(fk.rel_model, attr=f'_{fk.name}') # type: ignore[call-arg]
703 if idx != len(foreign_keys) - 1:
704 query = query.switch(cls) # type: ignore[attr-defined]
705 return query
707 class Meta:
708 """The metadata container for the Model class"""
710 database = database_proxy
711 """The reference database. A proxy is used as a placeholder that will be automatically replaced by the real
712 instance of the database at runtime."""
714 model_metadata_class = ThreadSafeDatabaseMetadata
715 """Thread-safe metadata storage to allow runtime database binding across threads."""
717 legacy_table_names = False
718 """
719 Set the default table name as the snake case of the Model camel case name.
721 So for example, a model named ThisIsMyTable will corresponds to a database table named this_is_my_table.
722 """
724 suffix = ''
725 """
726 Set the value to append to the table name.
727 """
729 prefix = ''
730 """
731 Set the value to prepend to the table name.
732 """
734 table_function = make_prefixed_suffixed_name
735 """
736 Set the table naming function.
737 """
739 automatic_creation = True
740 """
741 Whether the table linked to the model should be created automatically
742 """
744 file_trigger_auto_create = False
745 """
746 Whether to automatically create triggers to delete files once a row with a FilenameField is removed
747 """