Coverage for src / mafw / db / fields.py: 100%
52 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 customised model fields specific for MAFw.
6"""
8from pathlib import Path
9from typing import Any
11from peewee import FieldAccessor, TextField
13import mafw.tools.file_tools
16class FileNameFieldAccessor(FieldAccessor):
17 """
18 A field accessor specialized for filename fields.
20 In the constructor of the :class:`FileNameField` and subclasses, the user can specify the name of a checksum
21 field linked to this filename. This is very useful because in this way, the user does not have to manually assign
22 any value to this field that will simply be automatically updated when the filename field is updated.
24 The user can disable this automatic feature either removing the link in the :class:`FileNameField` or simply
25 assigning a value to the :class:`FileChecksumField`.
26 """
28 # noinspection PyProtectedMember
29 def __set__(self, instance: Any, value: Any) -> None:
30 """
31 Sets the value of field in the instance data dictionary.
33 If the field has a checksum field specified and this has not been initialised, then this one as well get
34 assigned the same value.
36 :param instance: a Model instance.
37 :param value: the value to be assigned to the Field.
38 """
39 # do whatever is needed by the normal FieldAccessor
40 super().__set__(instance, value)
41 # if there is a linked field, continue here; otherwise, you have already finished.
42 if self.field.checksum_field is not None:
43 # check if the model has an attribute to store the initialisation of the linked field,
44 # if not, create one and set it to False
45 if not hasattr(instance, 'init_' + self.field.checksum_field):
46 setattr(instance, 'init_' + self.field.checksum_field, False)
48 # if the linked field has not been initialised, then set its value to the same value
49 # of this field.
50 if not getattr(instance, 'init_' + self.field.checksum_field):
51 instance.__data__[self.field.checksum_field] = value
52 instance._dirty.add(self.field.checksum_field)
55# noinspection PyProtectedMember
56class FileChecksumFieldAccessor(FieldAccessor):
57 """
58 Field accessor specialized for file checksum fields.
60 When the field is directly set, then an initialization flag in the model instance is turned to True to avoid
61 that the linked primary field will overrule this value again.
63 For each checksum field named my_checksum, the model instance will get an attribute: init_my_checksum to be
64 used as an initialization flag.
66 Once the field is manually set, to re-establish the automatic mechanism, the user has to manually toggle the
67 initialization flag.
68 """
70 def __set__(self, instance: Any, value: Any) -> None:
71 """
72 Sets the value of the field in the instance data dictionary.
74 When the field is directly set, then the initialisation flag in the instance is also turned to True to avoid
75 that the primary field will overrule this value again.
77 For each checksum field named my_checksum, the model instance will get an attribute: init_my_checksum to be
78 used as an initialisation flag.
80 Once the field is manually set, to re-establish the automatic mechanism, the user has to manually toggle the
81 initialisation flag.
83 :param instance: a Model instance.
84 :param value: the value to be assigned to the Field.
85 """
86 # do whatever is needed.
87 super().__set__(instance, value)
88 # check if the model has an attribute to store the initialisation of this field.
89 # if not, create one and set it to False
90 if not hasattr(instance, 'init_' + self.name):
91 setattr(instance, 'init_' + self.name, False)
92 # the field has been initialised, so set it to True, to avoid the primary field to overrule it.
93 setattr(instance, 'init_' + self.name, True)
96class FileNameField(TextField):
97 """
98 Field to be used for filenames.
100 It is just an overload of TextField, that allows to apply filters and python functions specific to filenames.
102 If the user specifies the name of a file checksum field, then when this field is updated, the checksum one will
103 also be automatically updated.
105 .. seealso::
107 * :class:`~mafw.db.fields.FileNameListField` for a field able to store a list of filenames.
109 * :func:`~mafw.tools.file_tools.remove_widow_db_rows` for a function removing entries from a database table
110 where the corresponding files on disk are missing.
112 * :func:`~mafw.tools.file_tools.verify_checksum` for a function comparing the actual checksum with the
113 stored one and in case removing outdated entries from the DB.
115 """
117 accessor_class = FileNameFieldAccessor
118 """The specific accessor class"""
120 def __init__(self, checksum_field: str | None = None, *args: Any, **kwargs: Any) -> None:
121 """
122 Constructor parameter:
124 :param checksum_field: The name of the checksum field linked to this filename. Defaults to None.
125 :type checksum_field: str, Optional
126 """
127 super().__init__(*args, **kwargs)
128 self.checksum_field = checksum_field
130 def db_value(self, value: str | Path) -> str:
131 """Converts the input python value into a string for the DB."""
132 return str(value)
134 def python_value(self, value: str) -> Path | None:
135 """Converts the db value from str to Path
137 The return value might also be None, if the user set the field value to null.
139 :param value: The value of the field as stored in the database.
140 :type value: str
141 :return: The converted value as a path. It can be None, if value was stored as null.
142 :rtype: Path | None
143 """
144 return Path(value) if value is not None else None
147class FileNameListField(FileNameField):
148 """
149 A field for a list of file names.
151 The evolution of the :class:`~mafw.db.fields.FileNameField`, this field is able to store a list of filenames as a
152 ';' separated string of full paths.
154 It is meant to be used when a processor is saving a bunch of correlated files that are to be used all together.
156 In a similar way as its parent class, it can be link to a checksum field, in this case, the checksum of the whole
157 file list will be calculated.
158 """
160 def db_value(self, value: list[str | Path] | str | Path) -> str:
161 """Converts the list of paths in a ';' separated string"""
162 if isinstance(value, (str, Path)):
163 value = [value]
164 return ';'.join([str(v) for v in value])
166 def python_value(self, value: str) -> list[Path]: # type: ignore[override]
167 """Converts the ';' separated string in a list of paths"""
168 if value is None: # this might be the case, when the database field is actually set to null.
169 return []
170 return [Path(p) for p in value.split(';')]
173class FileChecksumField(TextField):
174 """
175 A field to be used for file checksum.
177 It is the evolution of the TextField for storing file checksum hexadecimal digest.
179 If linked to a :class:`FileNameField` or :class:`FileNameListField`, then it will be automatically filled when the
180 primary file name (or list of file names) field is set.
182 If the user decides to set its value manually, then he can provide either the string with the hexadecimal
183 characters as calculated by :func:`~mafw.tools.file_tools.file_checksum`, or simply the filename (or filename
184 list) and the field will perform the calculation automatically.
186 """
188 accessor_class = FileChecksumFieldAccessor
189 """The specific field accessor class"""
191 def db_value(self, value: str | Path | list[str | Path]) -> str:
192 """
193 Converts the python assigned value to the DB type.
195 The checksum will be stored in the DB as a string containing only hexadecimal characters
196 (see `hexdigest <https://docs.python.org/3/library/hashlib.html#hashlib.hash.hexdigest>`_).
198 The user can provide the checksum directly or the path to the file or a list of path to files. If a Path,
199 or list of Path, is provided, then the checksum will be calculated, if a str (a path converted into a
200 string) is provided, the function will try to see if a file with that path exists. If so, the checksum will
201 be calculated, if not, the original string is assumed to be the checksum.
203 :param value: The checksum or path to the file, or list of path to files for which the checksum has to be
204 stored.
205 :type value: str | Path | list[str | Path]
206 :return: The checksum string for the database storage.
207 :rtype: str
208 """
210 if isinstance(value, Path):
211 # we got the filename, not the digest. we need to calculate it
212 value = mafw.tools.file_tools.file_checksum(value)
213 elif isinstance(value, list):
214 # we got a list of path, not the digest. we need to calculate the digest
215 # of the whole list
216 value = mafw.tools.file_tools.file_checksum(value)
217 else:
218 test_value = Path(value)
219 if test_value.exists():
220 # with a very high probability, the user passed a str to a path.
221 value = mafw.tools.file_tools.file_checksum(test_value)
222 return value
224 def python_value(self, value: str) -> str:
225 return value