Coverage for src / mafw / db / std_tables.py: 92%
55 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-09 09:08 +0000
1# Copyright 2025 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Module provides standard tables that are included in all database created by MAFw processor.
7Standard tables are automatically created and initialized by a :class:`~mafw.processor.Processor` or a
8:class:`~mafw.processor.ProcessorList` when opening a database connection.
10This means that if a processor receives a valid database object, then it will suppose that the connection was already
11opened somewhere else (either from a ProcessorList or a third party) and thus it is not creating the standard tables.
13If a processor is constructed using a database configuration dictionary, then it will first try to open a connection
14to the DB, then creating all standard tables and finally executing their :class:`StandardTable.init` method. The same
15apply for the Processor list.
17In other words, object responsible to open the database connection is taking care also of creating the standard
18tables and of initializing them. If the user opens the connection and passes it to a Processor or ProcessorList,
19then the user is responsible to create the standard tables and to initialize them.
21All standard tables must derive from the :class:`StandardTable` to have the same interface for the
22initialization.
23"""
25from types import TracebackType
26from typing import cast
28import peewee
29from peewee import AutoField, BooleanField, CharField, TextField
31from mafw.db.db_model import MAFwBaseModel
32from mafw.db.db_types import PeeweeModelWithMeta
33from mafw.db.fields import FileChecksumField, FileNameListField
36class StandardTable(MAFwBaseModel):
37 """A base class for tables that are generated automatically by the MAFw processor."""
39 @classmethod
40 def init(cls) -> None:
41 """The user must overload this method, if he wants some specific operations to be performed on the model
42 everytime the database is connected."""
43 pass
46class StandardTableDoesNotExist(Exception):
47 """An exception raised when trying to access a not existing table."""
50class TriggerStatus(StandardTable):
51 """A Model for the trigger status"""
53 trigger_type_id = AutoField(primary_key=True, help_text='Primary key')
54 trigger_type = TextField(
55 help_text='You can use it to specify the type (DELETE/INSERT/UPDATE) or the name of a specific trigger'
56 )
57 status = BooleanField(default=True, help_text='False (0) = disable / True (1) = enable')
59 # noinspection PyProtectedMember
60 @classmethod
61 def init(cls) -> None:
62 """Resets all triggers to enable when the database connection is opened."""
63 data = [
64 dict(trigger_type_id=1, trigger_type='DELETE', status=True),
65 dict(trigger_type_id=2, trigger_type='INSERT', status=True),
66 dict(trigger_type_id=3, trigger_type='UPDATE', status=True),
67 dict(trigger_type_id=4, trigger_type='DELETE_FILES', status=True),
68 ]
70 # this is used just to make mypy happy
71 # cls and meta_cls are exactly the same thing
72 meta_cls = cast(PeeweeModelWithMeta, cls)
74 db_proxy = meta_cls._meta.database
75 if isinstance(db_proxy, peewee.DatabaseProxy): 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true
76 db = cast(peewee.Database, db_proxy.obj)
77 else:
78 db = cast(peewee.Database, db_proxy)
80 if isinstance(db, peewee.PostgresqlDatabase): 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 cls.insert_many(data).on_conflict(
82 'update', conflict_target=[cls.trigger_type_id], update={cls.status: True}
83 ).execute()
84 else:
85 cls.insert_many(data).on_conflict_replace().execute()
88class TriggerStatusDoesNotExist(Exception):
89 """An exception raised when trying to access a not existing table."""
92class TriggerDisabler:
93 """
94 A helper tool to disable a specific type of triggers.
96 Not all SQL dialects allow to temporarily disable trigger execution.
98 In order overcome this limitation, MAFw has introduced a practical workaround. All types of triggers are active
99 by default but they can be temporarily disabled, by changing their status in the :class:`.TriggerStatus` table.
101 In order to disable the trigger execution, the user has to set the status of the corresponding status to 0 and also
102 add a when condition to the trigger definition.
104 Here is an example code:
106 .. code-block:: python
108 class MyTable(MAFwBaseModel):
109 id_ = AutoField(primary_key=True)
110 integer = IntegerField()
111 float_num = FloatField()
113 @classmethod
114 def triggers(cls):
115 return [
116 Trigger(
117 'mytable_after_insert',
118 (TriggerWhen.After, TriggerAction.Insert),
119 cls,
120 safe=True,
121 )
122 .add_sql(
123 'INSERT INTO target_table (id__id, half_float_num) VALUES (NEW.id_, NEW.float_num / 2)'
124 )
125 .add_when(
126 '1 == (SELECT status FROM trigger_status WHERE trigger_type_id == 1)'
127 )
128 ]
130 When you want to perform a database action with the trigger disabled, you can either use this class as context
131 manager or call the :meth:`.disable` and :meth:`.enable` methods.
133 .. code-block:: python
135 # as a context manager
136 with TriggerDisabler(trigger_type_id = 1):
137 # do something without triggering any trigger of type 1.
139 # with the explicit methods
140 disabler = TriggerDisabler(1)
141 disabler.disable()
142 # do something without triggering any trigger of type 1.
143 disabler.enable()
145 When using the two explicit methods, the responsibility to assure that the triggers are re-enabled in on the user.
146 """
148 def __init__(self, trigger_type_id: int) -> None:
149 """
150 Constructor parameters:
152 :param trigger_type_id: the id of the trigger to be temporary disabled.
153 :type trigger_type_id: int
154 """
155 self.trigger_type_id = trigger_type_id
157 def disable(self) -> None:
158 """Disable the trigger"""
159 TriggerStatus.update({TriggerStatus.status: 0}).where(
160 TriggerStatus.trigger_type_id == self.trigger_type_id
161 ).execute()
163 def enable(self) -> None:
164 """Enable the trigger"""
165 TriggerStatus.update({TriggerStatus.status: 1}).where(
166 TriggerStatus.trigger_type_id == self.trigger_type_id
167 ).execute()
169 def __enter__(self) -> 'TriggerDisabler':
170 """
171 Context enter. Disable the trigger.
172 """
173 self.disable()
174 return self
176 def __exit__(
177 self, type_: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
178 ) -> None:
179 """
180 Context exit
182 :param type_: Exception type causing the context manager to exit. Defaults to None.
183 :type type_: type[BaseException], Optional
184 :param value: Exception that caused the context manager to exit. Defaults to None.
185 :type value: BaseException, Optional
186 :param traceback: Traceback. Defaults to None.
187 :type traceback: TracebackType
188 """
189 self.enable()
192class OrphanFile(StandardTable):
193 """
194 A Model for the files to be removed from disc
196 .. versionchanged:: v2.0.0
197 The checksum field is set to allow null values.
198 The class is set not to automatically generate triggers for file removal
200 """
202 file_id = AutoField(primary_key=True, help_text='Primary key')
203 filenames = FileNameListField(help_text='The path to the file to be deleted', checksum_field='checksum')
204 checksum = FileChecksumField(help_text='The checksum of the files in the list.', null=True)
206 class Meta:
207 file_trigger_auto_create = False
210class OrphanFileDoesNotExist(peewee.DoesNotExist):
211 """An exception raised when trying to access a not existing table."""
214class PlotterOutput(StandardTable):
215 """
216 A model for the output of the plotter processors.
218 The model has a trigger activated on delete queries to insert filenames and checksum in the OrphanFile model via
219 the automatic file delete trigger generation.
220 """
222 plotter_name = CharField(primary_key=True, help_text='The plotter processor name', max_length=511)
223 filename_list = FileNameListField(help_text='The path to the output file', checksum_field='checksum')
224 checksum = FileChecksumField(help_text='The checksum of the files in the list.')
226 class Meta:
227 depends_on = [OrphanFile]
228 file_trigger_auto_create = True
231class PlotterOutputDoesNotExist(peewee.DoesNotExist):
232 """An exception raised when trying to access a not existing table."""