# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""The module provides utilities for handling file, filename, hashing and so on."""
from __future__ import annotations
import hashlib
import warnings
from collections.abc import Sequence
from pathlib import Path
from typing import Any, cast
from peewee import Model
import mafw.db.fields
from mafw.db.db_types import PeeweeModelWithMeta
from mafw.mafw_errors import ModelError
from mafw.storage import LogicalFileReference, StorageService
[docs]
def file_checksum(filenames: str | Path | Sequence[str | Path], buf_size: int = 65536) -> str:
"""
Generates the hexadecimal digest of a file or a list of files.
The digest is calculated using the sha256 algorithm.
:param filenames: The filename or the list of filenames for digest calculations.
:type filenames: str, Path, list
:param buf_size: The buffer size in bytes for reading the input files. Defaults to 64kB.
:type buf_size: int, Optional
:return: The hexadecimal digest.
:rtype: str
"""
if isinstance(filenames, str) or not isinstance(filenames, Sequence):
filenames = [filenames]
hasher = hashlib.sha256()
for filename in filenames:
with open(filename, 'rb') as file:
while True:
data = file.read(buf_size)
if not data:
break
hasher.update(data)
return hasher.hexdigest()
# noinspection PyUnresolvedReferences
[docs]
def _as_sequence(value: object) -> list[Any]:
"""Normalize one scalar or list-like model field value for file operations."""
if isinstance(value, (str, LogicalFileReference)):
return [value]
if isinstance(value, Sequence):
return list(value)
return [value]
[docs]
def _path_exists(path: Path) -> bool:
"""Distinguish a missing file from permission and filesystem failures."""
try:
path.stat()
except FileNotFoundError:
return False
return True
[docs]
def remove_widow_db_rows(
models: list[Model | type[Model]] | Model | type[Model], *, storage_service: StorageService | None = None
) -> None:
"""Removes widow rows from a database table.
According to MAFw architecture, the Database is mainly providing I/O support to the various processors.
This means that the processor retrieves a list of items from a database table for processing and subsequently
updates a result table with the newly generated outputs.
Very often the input and output data are not stored directly in the database, but rather in files saved on the
disc. In this case, the database is just providing a valid path where the input (or output) data can be found.
From this point of view, a **widow row** is a database entry in which the file referenced by the FilenameField
has been deleted. A typical example is the following: the user wants a certain processor to regenerate a given
result stored inside an output file. Instead of setting up a complex filter so that the processor receives only
this element to process, the user can delete the actual output file and ask the processor to process all new items.
The provided ``models`` can be either a list or a single element, representing either an instance of a DB model
or a model class. If a model class is provided, then a select over all its entries is performed.
The function checks physical filename fields and storage-aware fields for
existing files. Logical storage values are resolved only when a service is
provided; unresolved logical values are skipped rather than treated as
missing files.
:param models: A list or a single Model instance or Model class for widow rows removal.
:type models: list[Model | type(Model)] | Model | type(Model)
:param storage_service: Initialized service for checking logical-reference fields. When omitted, logical fields
are skipped with a warning.
:type storage_service: mafw.storage.StorageService | None
:raises TypeError: if ``models`` is not of the right type.
:raises mafw.storage.StorageError: if resolving a logical reference fails.
.. versionchanged:: 2.4
Added logical-reference checks through an optional storage service; missing service skips logical fields.
"""
from mafw.db.std_tables import TriggerDisabler
warned_logical_fields: set[tuple[type[Model], str]] = set()
# noinspection PyUnresolvedReferences,PyProtectedMember
def _check_row(r: Model) -> None:
"""Internal function performing the check and removal on a single instance"""
# this is just to make mypy happy
# r0 and r are exactly the same thing!
r0 = cast(PeeweeModelWithMeta, r)
for k, f in r0._meta.fields.items():
if isinstance(f, mafw.db.fields.StorageField):
raw_value = getattr(r, k)
if raw_value is None:
if isinstance(f, mafw.db.fields.StorageListField):
values = []
else:
r.delete_instance()
continue
else:
values = _as_sequence(raw_value)
missing = False
unresolved = False
for value in values:
if isinstance(value, LogicalFileReference):
if storage_service is None:
unresolved = True
key = (type(r), k)
if key not in warned_logical_fields:
warnings.warn(
f'Skipping logical file field {k!r}: no StorageService was provided.', stacklevel=2
)
warned_logical_fields.add(key)
continue
path = storage_service.resolve(value)
else:
path = Path(value)
if not _path_exists(path):
missing = True
break
if missing:
r.delete_instance()
elif not values and isinstance(f, mafw.db.fields.StorageListField):
continue
elif unresolved:
continue
continue
# since FileNameListField is a subclass of FileNameField, we first have to check for
# the case of a list and then of a simple field.
if isinstance(f, mafw.db.fields.FileNameListField):
files = [Path(p) if isinstance(p, str) else p for p in getattr(r, k)]
if any(not _path_exists(file) for file in files):
r.delete_instance()
elif isinstance(f, mafw.db.fields.FileNameField):
value = getattr(r, k)
if value is None or not _path_exists(Path(value) if isinstance(value, str) else value):
r.delete_instance()
elif isinstance(
f, (mafw.db.fields.LogicalFileReferenceField, mafw.db.fields.LogicalFileReferenceListField)
):
if storage_service is None:
key = (type(r), k)
if key not in warned_logical_fields:
warnings.warn(
f'Skipping logical file field {k!r}: no StorageService was provided.', stacklevel=2
)
warned_logical_fields.add(key)
continue
assert storage_service is not None
raw_value = getattr(r, k)
if raw_value is None:
r.delete_instance()
continue
values = _as_sequence(raw_value)
if not values or any(value is None for value in values):
r.delete_instance()
continue
paths = [storage_service.resolve(value) for value in values]
if any(not _path_exists(path) for path in paths):
r.delete_instance()
if isinstance(models, (Model, type(Model))):
models = [models]
with TriggerDisabler(trigger_type_id=4):
for m in models:
if isinstance(m, Model):
_check_row(m)
elif isinstance(m, type(Model)):
# this is just to make mypy happy
# m0 and m are exactly the same thing!
m0 = cast(PeeweeModelWithMeta, m)
for row in m0.select().execute():
_check_row(row)
else:
raise TypeError('models must be list[Model | type(Model)] | Model | type(Model)')
# noinspection PyUnresolvedReferences
[docs]
def verify_checksum(
models: list[Model | type[Model]] | Model | type[Model], *, storage_service: StorageService | None = None
) -> None:
"""
Verifies the goodness of FileChecksumField.
If in a model there is a FileChecksumField, this must be connected to a FileNameField or a FileNameListField in
the same model. The goal of this function is to recalculate the checksum of the FileNameField / FileNameListField
and compare it with the actual stored value. Storage-aware fields may include
logical references, which require the optional service for verification. If
the newly calculated value differs from the stored one, the
corresponding row in the model will be removed, as it is no longer valid.
If a file is missing, then the checksum check is not performed, but the row is removed right away.
This function can be CPU and I/O intensive and last a lot, so use it with care, especially when dealing with long
tables and large files.
:param models: A list or a single Model instance or Model class for checksum verification.
:type models: list[Model | type(Model)] | Model | type(Model)
:param storage_service: Initialized service for resolving logical-reference fields. When omitted, logical fields
are skipped with a warning.
:type storage_service: mafw.storage.StorageService | None
:raises TypeError: if ``models`` is not of the right type.
:raises mafw.mafw_errors.ModelError: if the FileCheckSumField is referring to a FilenameField that does not exist.
.. versionchanged:: 2.4
Added content checksum verification for logical references through an optional storage service.
"""
from mafw.db.std_tables import TriggerDisabler
warned_logical_fields: set[tuple[type[Model], str]] = set()
# noinspection PyUnresolvedReferences,PyProtectedMember
def _check_row(r: Model) -> None:
def _check_file(file: str | Path | Sequence[str | Path], stored_checksum: str) -> None:
checksum = file_checksum(file)
if checksum != getattr(r, stored_checksum):
r.delete_instance()
for k, f in r._meta.fields.items():
if isinstance(f, mafw.db.fields.StorageField):
if f.checksum_field is None:
continue
if not hasattr(r, f.checksum_field):
raise ModelError(
f'StorageField {k} is referring to {f.checksum_field}, but Model '
f'{r.__class__.__name__} has no such field'
)
raw_value = getattr(r, k)
if raw_value is None:
if isinstance(f, mafw.db.fields.StorageListField):
values = []
else:
r.delete_instance()
continue
else:
values = _as_sequence(raw_value)
if any(isinstance(value, LogicalFileReference) for value in values):
if storage_service is None:
key = (type(r), k)
if key not in warned_logical_fields:
warnings.warn(
f'Skipping checksum verification for logical field {k!r}: '
'no StorageService was provided.',
stacklevel=2,
)
warned_logical_fields.add(key)
continue
files = [
storage_service.resolve(value) if isinstance(value, LogicalFileReference) else Path(value)
for value in values
]
else:
files = [Path(value) for value in values]
if not all(_path_exists(file) for file in files):
r.delete_instance()
warnings.warn('A file is missing from the storage field, removing the whole row from the DB.')
else:
_check_file(files, f.checksum_field)
continue
logical = isinstance(
f, (mafw.db.fields.LogicalFileReferenceField, mafw.db.fields.LogicalFileReferenceListField)
)
physical = isinstance(f, mafw.db.fields.FileNameField)
if physical or logical:
# f is a filename field or a filename list field
# this might be linked to a Checksum field
if f.checksum_field is None:
continue
if logical and storage_service is None:
key = (type(r), k)
if key not in warned_logical_fields:
warnings.warn(
f'Skipping checksum verification for logical field {k!r}: no StorageService was provided.',
stacklevel=2,
)
warned_logical_fields.add(key)
continue
if logical:
assert storage_service is not None
if not hasattr(r, f.checksum_field):
raise ModelError(
f'FileNameField {k} is referring to {f.checksum_field}, but Model '
f'{r.__class__.__name__} has not such field'
)
if isinstance(f, (mafw.db.fields.FileNameListField, mafw.db.fields.LogicalFileReferenceListField)):
values = _as_sequence(getattr(r, k))
if logical and (not values or any(value is None for value in values)):
r.delete_instance()
continue
files = (
[cast(StorageService, storage_service).resolve(value) for value in values]
if logical
else [Path(p) if isinstance(p, str) else p for p in values]
)
if not all(_path_exists(file) for file in files):
r.delete_instance()
warnings.warn('A file is missing from the list, removing the whole row from the DB.')
else:
_check_file(files, f.checksum_field)
else: # isinstance(f, FileNameField)
value = getattr(r, k)
if value is None:
r.delete_instance()
continue
file = (
cast(StorageService, storage_service).resolve(value)
if logical
else (Path(value) if isinstance(value, str) else value)
)
if not _path_exists(file):
warnings.warn(f'{str(file)} does not exist, removing the corresponding row from the DB')
r.delete_instance()
else:
_check_file(file, f.checksum_field)
if isinstance(models, (Model, type(Model))):
models = [models]
with TriggerDisabler(trigger_type_id=4):
for m in models:
if isinstance(m, Model):
_check_row(m)
elif isinstance(m, type(Model)):
for row in m.select().execute():
_check_row(row)
else:
raise TypeError('models must be list[Model | type(Model)] | Model | type(Model)')