# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Module provides customised model fields specific for MAFw.
"""
from __future__ import annotations
from collections.abc import Iterable, Sequence
from pathlib import Path
from typing import Any
from peewee import FieldAccessor, TextField
import mafw.tools.file_tools
from mafw.storage import LogicalFileReference
_STORAGE_PATH_PREFIX = 'mafw:path:'
"""Envelope marker for physical paths that need an explicit storage type."""
_STORAGE_LOGICAL_PREFIX = 'mafw:logical:'
"""Envelope marker for logical file references."""
_STORAGE_PREFIX = 'mafw:'
"""Reserved prefix for future self-describing storage codecs."""
[docs]
def _encode_storage_value(value: str | Path | LogicalFileReference, *, list_member: bool = False) -> str:
"""Encode one physical path or logical reference without resolving it."""
if isinstance(value, LogicalFileReference):
return f'{_STORAGE_LOGICAL_PREFIX}{value.to_string()}'
if not isinstance(value, (str, Path)):
raise TypeError('storage values must be strings, pathlib.Path objects, or LogicalFileReference objects')
text = str(value)
if list_member and ';' in text:
raise ValueError('storage list path values cannot contain semicolons')
if text.startswith(_STORAGE_PREFIX):
return f'{_STORAGE_PATH_PREFIX}{text}'
return text
[docs]
def _decode_storage_value(value: str) -> Path | LogicalFileReference:
"""Decode a complete known envelope; otherwise retain legacy path text."""
if value.startswith(_STORAGE_PATH_PREFIX):
path = value[len(_STORAGE_PATH_PREFIX) :]
return Path(path) if path else Path(value)
if value.startswith(_STORAGE_LOGICAL_PREFIX):
serialized_reference = value[len(_STORAGE_LOGICAL_PREFIX) :]
try:
return LogicalFileReference.from_string(serialized_reference)
except (TypeError, ValueError):
# Incomplete reserved-looking strings may be legacy physical paths.
return Path(value)
if value.startswith(_STORAGE_PREFIX):
parts = value.split(':', maxsplit=2)
if len(parts) == 3 and parts[1] not in {'path', 'logical'} and parts[2]:
raise ValueError(f'Unknown storage value codec {parts[1]!r}')
return Path(value)
[docs]
class FileNameFieldAccessor(FieldAccessor):
"""
A field accessor specialized for filename fields.
In the constructor of the :class:`FileNameField` and subclasses, the user can specify the name of a checksum
field linked to this filename. This is very useful because in this way, the user does not have to manually assign
any value to this field that will simply be automatically updated when the filename field is updated.
The user can disable this automatic feature either removing the link in the :class:`FileNameField` or simply
assigning a value to the :class:`FileChecksumField`.
"""
# noinspection PyProtectedMember
def __set__(self, instance: Any, value: Any) -> None:
"""
Sets the value of field in the instance data dictionary.
If the field has a checksum field specified and this has not been initialised, then this one as well get
assigned the same value.
:param instance: a Model instance.
:param value: the value to be assigned to the Field.
"""
# do whatever is needed by the normal FieldAccessor
super().__set__(instance, value)
# if there is a linked field, continue here; otherwise, you have already finished.
if self.field.checksum_field is not None:
# check if the model has an attribute to store the initialisation of the linked field,
# if not, create one and set it to False
if not hasattr(instance, 'init_' + self.field.checksum_field):
setattr(instance, 'init_' + self.field.checksum_field, False)
# if the linked field has not been initialised, then set its value to the same value
# of this field.
if not getattr(instance, 'init_' + self.field.checksum_field):
instance.__data__[self.field.checksum_field] = value
instance._dirty.add(self.field.checksum_field)
# noinspection PyProtectedMember
[docs]
class FileChecksumFieldAccessor(FieldAccessor):
"""
Field accessor specialized for file checksum fields.
When the field is directly set, then an initialization flag in the model instance is turned to True to avoid
that the linked primary field will overrule this value again.
For each checksum field named my_checksum, the model instance will get an attribute: init_my_checksum to be
used as an initialization flag.
Once the field is manually set, to re-establish the automatic mechanism, the user has to manually toggle the
initialization flag.
"""
def __set__(self, instance: Any, value: Any) -> None:
"""
Sets the value of the field in the instance data dictionary.
When the field is directly set, then the initialisation flag in the instance is also turned to True to avoid
that the primary field will overrule this value again.
For each checksum field named my_checksum, the model instance will get an attribute: init_my_checksum to be
used as an initialisation flag.
Once the field is manually set, to re-establish the automatic mechanism, the user has to manually toggle the
initialisation flag.
:param instance: a Model instance.
:param value: the value to be assigned to the Field.
"""
# do whatever is needed.
super().__set__(instance, value)
# check if the model has an attribute to store the initialisation of this field.
# if not, create one and set it to False
if not hasattr(instance, 'init_' + self.name):
setattr(instance, 'init_' + self.name, False)
# the field has been initialised, so set it to True, to avoid the primary field to overrule it.
setattr(instance, 'init_' + self.name, True)
[docs]
class FileNameField(TextField):
"""
Field to be used for filenames.
It is just an overload of TextField, that allows to apply filters and python functions specific to filenames.
If the user specifies the name of a file checksum field, then when this field is updated, the checksum one will
also be automatically updated.
.. seealso::
* :class:`~mafw.db.fields.FileNameListField` for a field able to store a list of filenames.
* :func:`~mafw.tools.file_tools.remove_widow_db_rows` for a function removing entries from a database table
where the corresponding files on disk are missing.
* :func:`~mafw.tools.file_tools.verify_checksum` for a function comparing the actual checksum with the
stored one and in case removing outdated entries from the DB.
"""
accessor_class = FileNameFieldAccessor
"""The specific accessor class"""
def __init__(self, checksum_field: str | None = None, *args: Any, **kwargs: Any) -> None:
"""
Constructor parameter:
:param checksum_field: The name of the checksum field linked to this filename. Defaults to None.
:type checksum_field: str, Optional
"""
super().__init__(*args, **kwargs)
self.checksum_field = checksum_field
[docs]
def db_value(self, value: str | Path | None) -> str | None:
"""Converts the input python value into a string for the DB."""
return str(value) if value is not None else None
[docs]
def python_value(self, value: str) -> Path | None:
"""Converts the db value from str to Path
The return value might also be None, if the user set the field value to null.
:param value: The value of the field as stored in the database.
:type value: str
:return: The converted value as a path. It can be None, if value was stored as null.
:rtype: Path | None
"""
return Path(value) if value is not None else None
[docs]
class FileNameListField(FileNameField):
"""
A field for a list of file names.
The evolution of the :class:`~mafw.db.fields.FileNameField`, this field is able to store a list of filenames as a
';' separated string of full paths.
It is meant to be used when a processor is saving a bunch of correlated files that are to be used all together.
In a similar way as its parent class, it can be link to a checksum field, in this case, the checksum of the whole
file list will be calculated.
"""
[docs]
def db_value(self, value: list[str | Path] | str | Path | None) -> str | None:
"""Converts the list of paths in a ';' separated string"""
if value is None:
return None
if isinstance(value, (str, Path)):
value = [value]
return ';'.join([str(v) for v in value])
[docs]
def python_value(self, value: str) -> list[Path]: # type: ignore[override]
"""Converts the ';' separated string in a list of paths"""
if value is None: # this might be the case, when the database field is actually set to null.
return []
return [Path(p) for p in value.split(';')]
[docs]
class StorageFieldAccessor(FieldAccessor):
"""Maintain physical-only checksum propagation for storage-aware fields.
.. versionadded:: 2.4
"""
def __set__(self, instance: Any, value: Any) -> None:
"""Copy only physical values to an associated checksum field."""
super().__set__(instance, value)
checksum_field = self.field.checksum_field
if checksum_field is None:
return
flag_name = f'init_{checksum_field}'
if not hasattr(instance, flag_name):
setattr(instance, flag_name, False)
if getattr(instance, flag_name):
return
instance.__data__[checksum_field] = _storage_checksum_value(self.field, value)
instance._dirty.add(checksum_field)
[docs]
class StorageField(TextField):
"""Store a physical path or logical reference in one self-describing text field.
Ordinary physical paths retain their legacy untagged representation. Paths
beginning with the reserved ``mafw:`` prefix are written with a ``mafw:path:``
envelope. Logical references use ``mafw:logical:<volume>:<relative/path>``.
Incomplete or malformed reserved-looking values remain legacy physical paths;
a complete unknown codec identifier raises :class:`ValueError`.
A string assigned in Python is always treated as a physical path. Assign a
:class:`~mafw.storage.LogicalFileReference` object for logical data. The
codec never resolves paths or accesses a storage service.
Automatic checksum propagation occurs only for physical values. For a
logical value, provide a resolved physical path or an already computed
checksum explicitly.
:param checksum_field: Optional name of the associated checksum field.
.. versionadded:: 2.4
"""
accessor_class = StorageFieldAccessor
def __init__(self, checksum_field: str | None = None, *args: Any, **kwargs: Any) -> None:
"""Initialize a text field with an optional physical checksum association."""
super().__init__(*args, **kwargs)
self.checksum_field = checksum_field
[docs]
def db_value(self, value: str | Path | LogicalFileReference | None) -> str | None:
"""Serialize one physical or logical storage value."""
return None if value is None else _encode_storage_value(value)
[docs]
def python_value(self, value: str | None) -> Path | LogicalFileReference | None:
"""Decode storage text into a physical path or logical reference."""
return None if value is None else _decode_storage_value(value)
[docs]
class StorageListField(StorageField):
"""Store an ordered, self-describing list of physical and logical values.
Values use the existing semicolon-separated text format. A physical list
member containing ``;`` is rejected because it cannot be represented
unambiguously. Existing text is decoded using the historical split rule.
SQL NULL and empty text both produce an empty list.
:param checksum_field: Optional name of the associated checksum field.
.. versionadded:: 2.4
"""
[docs]
def db_value(
self,
value: Iterable[str | Path | LogicalFileReference] | str | Path | LogicalFileReference | None,
) -> str | None:
"""Serialize list members in order without escaping legacy delimiters."""
if value is None:
return None
if isinstance(value, (str, Path, LogicalFileReference)):
values = [value]
elif isinstance(value, (bytes, bytearray)):
raise TypeError('storage lists must contain strings, paths, or logical references')
else:
try:
values = list(value)
except TypeError as exc:
raise TypeError('storage list value must be iterable') from exc
serialized: list[str] = []
for item in values:
if item is None:
raise TypeError('storage list members cannot be None')
if isinstance(item, str) and not item:
raise ValueError('storage list members cannot be empty strings')
serialized.append(_encode_storage_value(item, list_member=True))
return ';'.join(serialized)
[docs]
def python_value(self, value: str | None) -> list[Path | LogicalFileReference]: # type: ignore[override]
"""Decode a semicolon-separated list, preserving order and duplicates."""
if value is None or value == '':
return []
return [_decode_storage_value(item) for item in value.split(';')]
[docs]
def _storage_checksum_value(field: StorageField, value: Any) -> Any:
"""Return an all-physical assignment eligible for automatic hashing."""
if value is None:
return None
if isinstance(field, StorageListField):
if isinstance(value, (str, Path, LogicalFileReference)):
values: Sequence[Any] = [value]
elif isinstance(value, Sequence):
values = value
else:
return None
return value if all(isinstance(item, (str, Path)) for item in values) else None
return value if isinstance(value, (str, Path)) else None
[docs]
class _LogicalFileReferenceFieldBase(TextField):
"""Share checksum association metadata between logical-reference fields."""
def __init__(self, checksum_field: str | None = None, *args: Any, **kwargs: Any) -> None:
"""Initialize the field and retain optional checksum association metadata."""
super().__init__(*args, **kwargs)
self.checksum_field = checksum_field
[docs]
class LogicalFileReferenceField(_LogicalFileReferenceFieldBase):
"""Store a :class:`mafw.storage.LogicalFileReference` as portable text.
``checksum_field`` records the linked checksum field name for later storage
integration. Assignment does not populate that field: logical references
are unresolved identifiers, not filesystem paths or checksum digests.
:param checksum_field: Optional name of the associated checksum field.
.. versionadded:: 2.4
"""
[docs]
def db_value(self, value: LogicalFileReference | str | None) -> str | None:
"""Convert a reference or serialized reference to database text."""
if value is None:
return None
reference = value if isinstance(value, LogicalFileReference) else LogicalFileReference.from_string(value)
return reference.to_string()
[docs]
def python_value(self, value: str | None) -> LogicalFileReference | None:
"""Parse database text into a logical reference, preserving SQL NULL."""
return LogicalFileReference.from_string(value) if value is not None else None
[docs]
class LogicalFileReferenceListField(_LogicalFileReferenceFieldBase):
"""Store an ordered list of logical references as semicolon-separated text.
Empty or null values map to an empty list. A single reference or serialized
reference string is accepted as a one-element list. Empty elements in a
non-empty serialized list are rejected.
:param checksum_field: Optional name of the associated checksum field.
.. versionadded:: 2.4
"""
[docs]
def db_value(self, value: Any) -> str:
"""Serialize an iterable of references, or one reference, to text."""
if value is None:
return ''
if isinstance(value, (LogicalFileReference, str)):
values = [value]
else:
if isinstance(value, (bytes, bytearray)):
raise TypeError('logical reference list must contain references or strings')
try:
values = list(value)
except TypeError as error:
raise TypeError('logical reference list must be an iterable') from error
serialized: list[str] = []
for item in values:
if item is None:
raise TypeError('logical reference list elements cannot be None')
if isinstance(item, LogicalFileReference):
serialized.append(item.to_string())
elif isinstance(item, str):
serialized.append(LogicalFileReference.from_string(item).to_string())
else:
raise TypeError('logical reference list elements must be references or strings')
return ';'.join(serialized)
[docs]
def python_value(self, value: str | None) -> list[LogicalFileReference]:
"""Parse the stored list, preserving its order and duplicate entries."""
if value is None or value == '':
return []
parts = value.split(';')
if any(not part for part in parts):
raise ValueError('stored logical reference list contains an empty element')
return [LogicalFileReference.from_string(part) for part in parts]
[docs]
class FileChecksumField(TextField):
"""
A field to be used for file checksum.
It is the evolution of the TextField for storing file checksum hexadecimal digest.
If linked to a :class:`FileNameField`, :class:`FileNameListField`,
:class:`StorageField`, or :class:`StorageListField`, then it is automatically
filled when the associated field is set. Storage-aware fields propagate
only all-physical values; logical or mixed values require an explicit
resolved path or digest.
If the user decides to set its value manually, then he can provide either the string with the hexadecimal
characters as calculated by :func:`~mafw.tools.file_tools.file_checksum`, or simply the filename (or filename
list) and the field will perform the calculation automatically.
"""
accessor_class = FileChecksumFieldAccessor
"""The specific field accessor class"""
[docs]
def db_value(self, value: str | Path | list[str | Path] | None) -> str | None:
"""
Converts the python assigned value to the DB type.
The checksum will be stored in the DB as a string containing only hexadecimal characters
(see `hexdigest <https://docs.python.org/3/library/hashlib.html#hashlib.hash.hexdigest>`_).
The user can provide the checksum directly or the path to the file or a list of path to files. If a Path,
or list of Path, is provided, then the checksum will be calculated, if a str (a path converted into a
string) is provided, the function will try to see if a file with that path exists. If so, the checksum will
be calculated, if not, the original string is assumed to be the checksum.
:param value: The checksum or path to the file, or list of path to files for which the checksum has to be
stored.
:type value: str | Path | list[str | Path]
:return: The checksum string for the database storage.
:rtype: str
"""
if value is None:
return None
if isinstance(value, Path):
# we got the filename, not the digest. we need to calculate it
value = mafw.tools.file_tools.file_checksum(value)
elif isinstance(value, list):
# we got a list of path, not the digest. we need to calculate the digest
# of the whole list
value = mafw.tools.file_tools.file_checksum(value)
else:
test_value = Path(value)
if test_value.exists():
# with a very high probability, the user passed a str to a path.
value = mafw.tools.file_tools.file_checksum(test_value)
return value
def python_value(self, value: str) -> str:
return value