# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Migration-directory resolution and exclusive locking.
The effective migration directory is resolved with a fixed precedence — a CLI
override, then a :class:`~mafw.processor_library.db_init.SchemaMigrator`
``ActiveParameter``, then the steering-file database-configuration value, then a
documented default. A relative effective directory is resolved against a single
stable base directory (the steering file's directory) so that identical input
always yields the same absolute path.
Two exclusive locks keep concurrent processes from corrupting shared state: a
filesystem lock guards script writes in the directory, and a database advisory
lock guards history mutations. Each lock either acquires (waiting if needed) or
fails with a "locked" error, leaving history and scripts unaffected by a blocked
process.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import contextlib
import os
import time
from collections.abc import Iterator
from pathlib import Path
import peewee
from mafw.db.migration.errors import MigrationError, MigrationUnresolvedDirectoryError
from mafw.tools.path_resolution import PathResolutionContext, _resolve_interpolated_path
#: The documented default migration directory name, relative to the base.
DEFAULT_MIGRATION_DIRECTORY = 'migrations'
"""The documented default migration directory name, relative to the stable base."""
#: The lock file name created inside the migration directory.
_DIRECTORY_LOCK_FILE = '.mafw-migration.lock'
[docs]
class MigrationLockError(MigrationError):
"""
Raised when an exclusive migration lock cannot be acquired.
.. versionadded:: 2.4
"""
[docs]
def resolve_directory(
*,
cli_override: Path | str | None,
active_parameter: Path | str | None,
steering_value: Path | str | None,
base: Path,
default: str = DEFAULT_MIGRATION_DIRECTORY,
) -> Path:
"""
Resolve the effective migration directory following the fixed precedence.
Precedence (highest first): CLI override, SchemaMigrator ``ActiveParameter``,
steering-file value, documented default.
The effective value may be either **absolute** or **relative**:
* an **absolute** path is used as given (only normalised with ``resolve()``),
independent of the ``base`` directory;
* a **relative** path is resolved against ``base`` deterministically to one
absolute path, so identical input always yields the same result.
A leading ``~`` is expanded to the user's home directory before the
absolute/relative distinction is applied.
This function does not touch the filesystem beyond path resolution; use
:func:`validate_directory` to check that the resolved directory exists.
:param cli_override: The CLI override value, if provided.
:type cli_override: Path | str | None
:param active_parameter: The processor ``ActiveParameter`` value, if set.
:type active_parameter: Path | str | None
:param steering_value: The steering-file database-configuration value.
:type steering_value: Path | str | None
:param base: The stable base directory for resolving a *relative* value.
:type base: Path
:param default: The documented default directory name.
:type default: str
:return: The resolved absolute directory path.
:rtype: Path
"""
chosen: Path | str
for candidate in (cli_override, active_parameter, steering_value):
if candidate is not None and str(candidate) != '':
chosen = candidate
break
else:
chosen = default
context = PathResolutionContext.from_directory(base)
return _resolve_interpolated_path(chosen, context)
[docs]
def validate_directory(directory: Path) -> Path:
"""
Ensure the resolved directory exists and is a readable directory.
:param directory: The resolved directory to validate.
:type directory: Path
:return: The validated directory.
:rtype: Path
:raises MigrationUnresolvedDirectoryError: when the directory does not exist
or is not a directory.
"""
if not directory.exists() or not directory.is_dir():
raise MigrationUnresolvedDirectoryError(str(directory))
return directory
[docs]
@contextlib.contextmanager
def directory_lock(directory: Path, *, timeout: float = 30.0, poll: float = 0.1) -> Iterator[None]:
"""
Acquire an exclusive filesystem lock on the migration directory.
The lock is a sentinel file created with ``O_CREAT | O_EXCL``. The context
manager waits up to ``timeout`` seconds for a competing holder to release the
lock; if the timeout elapses it raises
:class:`~mafw.db.migration.directory.MigrationLockError` without
writing any script.
:param directory: The directory to lock.
:type directory: Path
:param timeout: The maximum number of seconds to wait for the lock.
:type timeout: float
:param poll: The polling interval in seconds while waiting.
:type poll: float
:raises MigrationLockError: when the lock cannot be acquired within the timeout.
"""
lock_path = directory / _DIRECTORY_LOCK_FILE
deadline = time.monotonic() + timeout
fd: int | None = None
while True:
try:
fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
break
except FileExistsError:
if time.monotonic() >= deadline:
raise MigrationLockError(f'The migration directory {directory} is locked by another process.') from None
time.sleep(poll)
try:
os.write(fd, str(os.getpid()).encode('utf-8'))
yield
finally:
if fd is not None:
os.close(fd)
with contextlib.suppress(FileNotFoundError):
lock_path.unlink()
[docs]
@contextlib.contextmanager
def database_lock(database: peewee.Database) -> Iterator[None]:
"""
Acquire an exclusive lock on the database before mutating the history.
The lock is implemented with the backend's native transaction/locking
facilities through :meth:`peewee.Database.atomic`, which begins an exclusive
transaction on the supported backends. If the backend reports the database is
busy, a :class:`~mafw.db.migration.directory.MigrationLockError` is raised and the history is left
unchanged.
:param database: The database to lock.
:type database: peewee.Database
:raises MigrationLockError: when the database is locked by another process.
"""
try:
with database.atomic():
yield
except peewee.OperationalError as exc: # pragma: no cover - timing dependent
raise MigrationLockError(f'The target database is locked by another process: {exc}') from exc