PumConfig

class PumConfig:
    """A class to hold configuration settings.

    Version Added:
        1.0.0
    """

    def __init__(
        self,
        base_path: str | Path,
        *,
        validate: bool = True,
        install_dependencies: bool = False,
        **kwargs: dict,
    ) -> None:
        """Initialize the configuration with key-value pairs.

        Args:
            base_path: The directory where the changelogs are located.
            validate: Whether to validate the changelogs and hooks and resolve dependencies. Defaults to True.
            install_dependencies: Whether to install missing dependencies into a cache directory.
            **kwargs: Key-value pairs representing configuration settings.

        Raises:
            PumConfigError: If the configuration is invalid.

        """

        if not isinstance(base_path, Path):
            base_path = Path(base_path)
        if not base_path.is_dir():
            raise PumConfigError(f"Directory `{base_path}` does not exist.")
        self._base_path = base_path

        self.dependency_path = None
        self._cached_handlers = []  # Cache handlers for cleanup

        try:
            self.config = ConfigModel(**kwargs)
        except ValidationError as e:
            logger.error("Config validation error: %s", e)
            raise PumConfigError(e) from e

        if validate:
            if self.config.pum.minimum_version and PUM_VERSION < self.config.pum.minimum_version:
                raise PumConfigError(
                    f"Minimum required version of pum is {self.config.pum.minimum_version}, but the current version is {PUM_VERSION}. Please upgrade pum."
                )
            try:
                self.validate(install_dependencies=install_dependencies)
            except (PumInvalidChangelog, PumHookError) as e:
                raise PumConfigError(
                    f"Configuration is invalid: {e}. You can disable the validation when constructing the config."
                ) from e

    @classmethod
    def from_yaml(
        cls,
        file_path: str | Path,
        *,
        validate: bool = True,
        install_dependencies: bool = False,
    ) -> "PumConfig":
        """Create a PumConfig instance from a YAML file.

        Args:
            file_path: The path to the YAML file.
            validate: Whether to validate the changelogs and hooks.
            install_dependencies: Whether to install missing dependencies into a cache directory.

        Returns:
            PumConfig: An instance of the PumConfig class.

        Raises:
            FileNotFoundError: If the file does not exist.
            yaml.YAMLError: If there is an error parsing the YAML file.

        """
        with Path.open(file_path) as file:
            data = yaml.safe_load(file)

        if "base_path" in data:
            raise PumConfigError("base_path not allowed in configuration instead.")

        base_path = Path(file_path).parent
        return cls(
            base_path=base_path,
            validate=validate,
            install_dependencies=install_dependencies,
            **data,
        )

    @property
    def base_path(self) -> Path:
        """Return the base path used for configuration and changelogs."""
        return self._base_path

    def cleanup_hook_imports(self) -> None:
        """Clean up imported modules and sys.path entries from hooks.

        This should be called when switching to a different module version to ensure
        that cached imports from the previous version don't cause conflicts.
        """
        # First, clean up sys.path additions from all cached handlers
        for handler in self._cached_handlers:
            handler.cleanup_sys_path()

        # Clear all modules that were loaded from this base_path
        base_path_str = str(self._base_path.resolve())
        modules_to_remove = []

        for module_name, module in list(sys.modules.items()):
            if module is None:
                continue
            module_file = getattr(module, "__file__", None)
            if module_file and module_file.startswith(base_path_str):
                modules_to_remove.append(module_name)

        for module_name in modules_to_remove:
            if module_name in sys.modules:
                logger.debug(f"Removing cached module: {module_name}")
                del sys.modules[module_name]

        self._cached_handlers.clear()

    def parameters(self) -> list[ParameterDefinition]:
        """Return a list of migration parameters.

        Returns:
            list[ParameterDefinition]: A list of migration parameter definitions.

        """
        return [
            ParameterDefinition(**parameter.model_dump(mode="python"))
            for parameter in self.config.parameters
        ]

    def parameter(self, name: str) -> ParameterDefinition:
        """Get a specific migration parameter by name.

        Args:
            name: The name of the parameter.

        Returns:
            ParameterDefintion: The migration parameter definition.

        Raises:
            PumConfigError: If the parameter name does not exist.

        """
        for parameter in self.config.parameters:
            if parameter.name == name:
                return ParameterDefinition(**parameter.model_dump(mode="python"))
        raise PumConfigError(f"Parameter '{name}' not found in configuration.") from None

    def last_version(
        self, min_version: str | None = None, max_version: str | None = None
    ) -> str | None:
        """Return the last version of the changelogs.
        The changelogs are sorted by version.

        Args:
            min_version (str | None): The version to start from (inclusive).
            max_version (str | None): The version to end at (inclusive).

        Returns:
            str | None: The last version of the changelogs. If no changelogs are found, None is returned.

        """
        changelogs = self.changelogs(min_version, max_version)
        if not changelogs:
            return None
        return changelogs[-1].version

    def changelogs(
        self,
        min_version: str | packaging.version.Version | None = None,
        max_version: str | packaging.version.Version | None = None,
    ) -> "list[Changelog]":
        """Return a list of changelogs.
        The changelogs are sorted by version.

        Args:
            min_version (str | None): The version to start from (inclusive).
            max_version (str | None): The version to end at (inclusive).

        Returns:
            list: A list of changelogs. Each changelog is represented by a Changelog object.

        """
        path = self._base_path / self.config.changelogs_directory
        if not path.is_dir():
            raise PumException(f"Changelogs directory `{path}` does not exist.")
        if not any(path.iterdir()):
            raise PumException(f"Changelogs directory `{path}` is empty.")

        # Local import avoids circular imports at module import time.
        from .changelog import Changelog

        changelogs = [Changelog(d) for d in path.iterdir() if d.is_dir()]

        if min_version:
            changelogs = [
                c for c in changelogs if c.version >= packaging.version.parse(min_version)
            ]
        if max_version:
            changelogs = [
                c for c in changelogs if c.version <= packaging.version.parse(max_version)
            ]

        changelogs.sort(key=lambda c: c.version)
        return changelogs

    def role_manager(self) -> RoleManager:
        """Return a RoleManager instance based on the roles defined in the configuration."""
        if not self.config.roles:
            logger.warning("No roles defined in the configuration. Returning an empty RoleManager.")
            return RoleManager([])
        return RoleManager([role.model_dump() for role in self.config.roles])

    def drop_app_handlers(self) -> list[HookHandler]:
        """Return the list of drop app hook handlers."""
        handlers = (
            [
                HookHandler(base_path=self._base_path, **hook.model_dump())
                for hook in self.config.application.drop
            ]
            if self.config.application.drop
            else []
        )
        # Cache handlers for cleanup
        self._cached_handlers.extend(handlers)
        return handlers

    def create_app_handlers(self) -> list[HookHandler]:
        """Return the list of create app hook handlers."""
        handlers = (
            [
                HookHandler(base_path=self._base_path, **hook.model_dump())
                for hook in self.config.application.create
            ]
            if self.config.application.create
            else []
        )
        # Cache handlers for cleanup
        self._cached_handlers.extend(handlers)
        return handlers

    def uninstall_handlers(self) -> list[HookHandler]:
        """Return the list of uninstall hook handlers."""
        return (
            [
                HookHandler(base_path=self._base_path, **hook.model_dump())
                for hook in self.config.uninstall
            ]
            if self.config.uninstall
            else []
        )

    def demo_data(self) -> dict[str, list[str]]:
        """Return a dictionary of demo data files defined in the configuration."""
        demo_data_files = {}
        for dm in self.config.demo_data:
            demo_data_files[dm.name] = dm.files or [dm.file]
        return demo_data_files

    def _dependency_cache_path(self) -> Path:
        """Return the pip prefix caching this configuration's dependencies.

        The key covers everything the installed content depends on, so a change of
        dependency, interpreter or platform gets its own prefix. The directory is
        kept across runs: reinstalling on every configuration load is slow, and it
        would download the dependency again on each module switch.

        Known limitation: nothing locks the prefix, and pip does not lock it
        either, so two processes installing the same module at the same time
        (two QGIS instances, or QGIS and the CLI) write into it concurrently.
        The common half-written state heals itself -- without the `dist-info`
        the dependency reads as missing and is installed again -- but the
        inverse leaves a prefix that resolves yet fails to import. Rare enough
        that a lock is not worth its failure modes; `pum cache clear` is the
        way out if it ever happens.
        """
        key = "\n".join(
            sorted(
                f"{d.name}|{d.minimum_version or ''}|{d.maximum_version or ''}"
                for d in self.config.dependencies
            )
            + [
                f"python|{sys.version_info.major}.{sys.version_info.minor}",
                f"platform|{sysconfig.get_platform()}",
            ]
        )
        digest = hashlib.sha256(key.encode()).hexdigest()[:16]
        name = _path_safe(self.config.pum.module)
        return dependency_cache_dir() / f"{name}-{digest}"

    def validate(self, install_dependencies: bool = False) -> None:
        """Validate the changelogs and hooks.

        Args:
            install_dependencies (bool): Whether to install missing dependencies into a cache directory.
        """

        if install_dependencies and self.config.dependencies:
            self.dependency_path = self._dependency_cache_path()
            self.dependency_path.mkdir(parents=True, exist_ok=True)
            # Added before resolving, so that a dependency already in the cache
            # is found and not installed again.
            _add_dependency_sys_paths(self.dependency_path)

        parameter_defaults = {}
        app_only_parameter_names = set()
        for parameter in self.config.parameters:
            parameter_defaults[parameter.name] = psycopg.sql.Literal(parameter.default)
            if parameter.app_only:
                app_only_parameter_names.add(parameter.name)

        for dependency in self.config.dependencies:
            DependencyHandler(**dependency.model_dump()).resolve(
                install_dependencies=install_dependencies, install_path=self.dependency_path
            )
            if self.dependency_path:
                # pip has only now created the site-packages directories, and the
                # next dependency must be able to see what this one pulled in.
                _add_dependency_sys_paths(self.dependency_path)

        # Validate changelogs with only non-app_only parameters.
        # app_only parameters must not be used in changelogs (migrations),
        # they are only allowed in application hooks.
        changelog_parameters = {
            k: v for k, v in parameter_defaults.items() if k not in app_only_parameter_names
        }
        for changelog in self.changelogs():
            try:
                changelog.validate(parameters=changelog_parameters)
            except (PumInvalidChangelog, PumSqlError) as e:
                # Check if the error is due to an app_only parameter being used
                error_text = _exception_chain_text(e)
                for name in app_only_parameter_names:
                    if name in error_text:
                        raise PumInvalidChangelog(
                            f"Changelog `{changelog}` uses app_only parameter `{name}`. "
                            f"App-only parameters cannot be used in changelogs (migrations), "
                            f"they are only allowed in application hooks (create/drop)."
                        ) from e
                raise PumInvalidChangelog(f"Changelog `{changelog}` is invalid.") from e

        hook_handlers = []
        if self.config.application.drop:
            hook_handlers.extend(self.drop_app_handlers())
        if self.config.application.create:
            hook_handlers.extend(self.create_app_handlers())
        for hook_handler in hook_handlers:
            try:
                hook_handler.validate(parameter_defaults)
            except PumHookError as e:
                raise PumHookError(f"Hook `{hook_handler}` is invalid.") from e