Coverage for src / mafw / db / db_model.py: 97%
223 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-12 09:03 +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):
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 """
476 meta_cls = cast('PeeweeModelWithMeta', cls)
477 db = get_db(meta_cls)
479 # Normalize input
480 if __data is not None:
481 data = dict(__data)
482 data.update(mapping)
483 else:
484 data = dict(mapping)
486 # --- PK detection ---
487 pk_names, pk_fields = get_primary_key_fields(meta_cls)
489 # --- Field resolution: fix #138 ---
490 # combined includes both field and column names, we need to be sure
491 # we do not count them twice
492 combined = meta_cls._meta.combined
493 seen_fields: set[Field] = set()
495 for name in data:
496 field = combined.get(name)
497 if field is None: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true
498 raise ValueError(f'Unknown field/column: {name}')
500 # Deduplicate same Field object (raw_image vs raw_image_id)
501 if field in seen_fields: 501 ↛ 502line 501 didn't jump to line 502 because the condition on line 501 was never true
502 continue
504 seen_fields.add(field)
506 # --- Build insert ---
507 insert_query = cls.insert(data)
509 # --- Backend-specific update mapping ---
510 if isinstance(db, MySQLDatabase):
511 update_mapping = {field: fn.VALUES(field) for field in seen_fields if field.name not in pk_names}
512 return insert_query.on_conflict(update=update_mapping)
514 else:
515 update_mapping = {
516 field: getattr(EXCLUDED, field.name) for field in seen_fields if field.name not in pk_names
517 }
519 return insert_query.on_conflict(
520 conflict_target=pk_fields,
521 update=update_mapping,
522 )
524 @classmethod
525 def std_upsert_many(cls, rows: Iterable[Any], fields: list[str] | None = None) -> ModelInsert:
526 """
527 Perform a standard upsert with many rows.
529 .. seealso::
531 Read the :meth:`std_upsert` documentation for an explanation of this method.
533 :param rows: A list with the rows to be inserted. Each item can be a dictionary or a tuple of values. If a
534 tuple is provided, then the `fields` must be provided.
535 :type rows: Iterable
536 :param fields: A list of field names. Defaults to None.
537 :type fields: list[str], Optional
538 """
539 meta_cls = cast(PeeweeModelWithMeta, cls)
540 db = get_db(meta_cls)
542 # Initialise the output iterator and field names list.
543 # rows_to_insert will hold the actual data to insert (possibly split from rows for inspection)
544 # insert_field_names will hold the column names (either provided or inferred from first row)
545 rows_to_insert: Iterable[Any] = rows
546 insert_field_names: list[str] = list(fields) if fields is not None else []
548 # If fields are not explicitly provided, we need to infer them from the input data.
549 # This branch handles the case where rows is a Sequence (list, tuple) that supports indexing.
550 if fields is None:
551 if isinstance(rows, Sequence) and not isinstance(rows, (str, bytes)):
552 # Sequence types (list, tuple) allow direct indexing to inspect the first row.
553 if len(rows) > 0: 553 ↛ 570line 553 didn't jump to line 570 because the condition on line 553 was always true
554 first_row = rows[0]
555 # If the first row is a dictionary (Mapping), extract its keys as field names.
556 if isinstance(first_row, Mapping): 556 ↛ 570line 556 didn't jump to line 570 because the condition on line 556 was always true
557 insert_field_names = list(first_row.keys())
558 else:
559 # For non-Sequence iterables (generators, etc.), we need to use tee() to inspect
560 # the first element without consuming the iterator. This creates two iterators:
561 # rows_inspect for looking at the first row, rows_to_insert for actual insertion.
562 rows_inspect, rows_to_insert = tee(rows)
563 first_row = next(rows_inspect, None)
564 if isinstance(first_row, Mapping): 564 ↛ 570line 564 didn't jump to line 570 because the condition on line 564 was always true
565 insert_field_names = list(first_row.keys())
567 # Retrieve primary key field information. pk_names is used to exclude PK fields from
568 # the update operation (PK should not be updated during upsert), while pk_fields is
569 # used as the conflict target for the database.
570 pk_names, pk_fields = get_primary_key_fields(meta_cls)
572 # Build the list of fields to update during the upsert.
573 # We need to:
574 # 1. Remove duplicates (seen_names set)
575 # 2. Exclude primary key fields (they're used for conflict detection, not update)
576 # 3. Verify the field actually exists in the model (field_obj may be None)
577 seen_fields: set[Field] = set()
578 update_fields: list[tuple[Field, str]] = []
579 for field_name in insert_field_names:
580 # Get the actual field object from the model metadata
581 field_obj = meta_cls._meta.combined.get(field_name)
583 if field_obj is None:
584 raise ValueError(f'Unknown field/column: {field_name}')
586 # Skip duplicate field
587 if field_obj in seen_fields:
588 continue
590 seen_fields.add(field_obj)
592 # Skip primary key fields - they identify the row, not data to update
593 if field_obj.name in pk_names:
594 continue
596 update_fields.append((field_obj, field_name))
598 # Create the bulk insert query using insert_many
599 insert_query = cls.insert_many(rows_to_insert, fields=insert_field_names or None)
601 # Define a helper for MySQL's VALUES() syntax.
602 # NOTE: VALUES() is deprecated in MySQL 8.0.20+, but still required by Peewee
603 def _mysql_value_expression(field: Field) -> Any:
604 return fn.VALUES(field)
606 # Handle database-specific on_conflict syntax:
607 # - MySQL uses VALUES(field_name) to reference the values being inserted
608 # - PostgreSQL and SQLite use EXCLUDED.field_name to reference rejected values
609 if isinstance(db, MySQLDatabase):
610 update_mapping = {field: _mysql_value_expression(field) for field, _ in update_fields}
611 return insert_query.on_conflict(update=update_mapping)
613 # For PostgreSQL/SQLite, use EXCLUDED table to reference the rejected insert values
614 update_mapping = {field: getattr(EXCLUDED, field.name) for field, _ in update_fields}
615 return insert_query.on_conflict(
616 conflict_target=pk_fields,
617 update=update_mapping,
618 )
620 def to_dict(
621 self,
622 recurse: bool = True,
623 backrefs: bool = False,
624 only: list[str] | None = None,
625 exclude: list[str] | None = None,
626 **kwargs: Any,
627 ) -> dict[str, Any]:
628 """
629 Convert model instance to dictionary with optional parameters
631 See full documentation directly on the `peewee documentation
632 <https://docs.peewee-orm.com/en/latest/peewee/playhouse.html#dict_to_model>`__.
634 :param recurse: If to recurse through foreign keys. Default to True.
635 :type recurse: bool, Optional
636 :param backrefs: If to include backrefs. Default to False.
637 :type backrefs: bool, Optional
638 :param only: A list of fields to be included. Defaults to None.
639 :type only: list[str], Optional
640 :param exclude: A list of fields to be excluded. Defaults to None.
641 :type exclude: list[str], Optional
642 :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>`__.
643 :return: A dictionary containing the key/value of the model.
644 :rtype: dict[str, Any]
645 """
646 # the playhouse module of peewee is not typed.
647 return model_to_dict( # type: ignore[no-any-return]
648 self,
649 recurse=recurse,
650 backrefs=backrefs, # type: ignore[no-untyped-call]
651 only=only,
652 exclude=exclude,
653 **kwargs,
654 )
656 @classmethod
657 def from_dict(cls, data: dict[str, Any], ignore_unknown: bool = False) -> 'MAFwBaseModel':
658 """
659 Create a new model instance from dictionary
661 :param data: The dictionary containing the key/value pairs of the model.
662 :type data: dict[str, Any]
663 :param ignore_unknown: If unknown dictionary keys should be ignored.
664 :type ignore_unknown: bool
665 :return: A new model instance.
666 :rtype: MAFwBaseModel
667 """
668 # the playhouse module of peewee is not typed.
669 return dict_to_model(cls, data, ignore_unknown=ignore_unknown) # type: ignore[no-untyped-call,no-any-return]
671 def update_from_dict(self, data: dict[str, Any], ignore_unknown: bool = False) -> 'MAFwBaseModel':
672 """
673 Update current model instance from dictionary
675 The model instance is returned for daisy-chaining.
677 :param data: The dictionary containing the key/value pairs of the model.
678 :type data: dict[str, Any]
679 :param ignore_unknown: If unknown dictionary keys should be ignored.
680 :type ignore_unknown: bool
681 """
682 update_model_from_dict(self, data, ignore_unknown=ignore_unknown) # type: ignore[no-untyped-call]
683 return self
685 @classmethod
686 def join_dependent_models(cls, base_query: Select | None = None) -> Select:
687 """Return a query that joins every model referenced through foreign keys.
689 Each foreign-key relation defined on ``cls`` is joined and attached to the
690 resulting rows via an attribute named ``_<field name>`` so that callers have
691 the same convenient access that manual ``.join_from`` calls would have
692 provided.
694 :param base_query: Optional query to extend. If not provided a new ``Select``
695 is created that includes ``cls`` and the dependent models.
696 :type base_query: :class:`peewee.Select`, optional
697 :return: The query joining every dependent model.
698 :rtype: :class:`peewee.Select`
699 """
700 if TYPE_CHECKING:
701 assert hasattr(cls, '_meta')
703 foreign_keys = [field for field in cls._meta.fields.values() if isinstance(field, ForeignKeyField)]
705 if len(foreign_keys) == 0:
706 return base_query or cls.select(cls)
708 dependent_models: list[type[ModelBase]] = []
709 for fk in foreign_keys:
710 rel_model = fk.rel_model
711 if rel_model not in dependent_models:
712 dependent_models.append(rel_model)
714 query: Select = base_query or cls.select(cls, *dependent_models)
716 for idx, fk in enumerate(foreign_keys):
717 query = query.join(fk.rel_model, attr=f'_{fk.name}') # type: ignore[call-arg]
718 if idx != len(foreign_keys) - 1:
719 query = query.switch(cls) # type: ignore[attr-defined]
720 return query
722 class Meta:
723 """The metadata container for the Model class"""
725 database = database_proxy
726 """The reference database. A proxy is used as a placeholder that will be automatically replaced by the real
727 instance of the database at runtime."""
729 model_metadata_class = ThreadSafeDatabaseMetadata
730 """Thread-safe metadata storage to allow runtime database binding across threads."""
732 legacy_table_names = False
733 """
734 Set the default table name as the snake case of the Model camel case name.
736 So for example, a model named ThisIsMyTable will corresponds to a database table named this_is_my_table.
737 """
739 suffix = ''
740 """
741 Set the value to append to the table name.
742 """
744 prefix = ''
745 """
746 Set the value to prepend to the table name.
747 """
749 table_function = make_prefixed_suffixed_name
750 """
751 Set the table naming function.
752 """
754 automatic_creation = True
755 """
756 Whether the table linked to the model should be created automatically
757 """
759 file_trigger_auto_create = False
760 """
761 Whether to automatically create triggers to delete files once a row with a FilenameField is removed
762 """