disdantic¶

class AutoImporterMixin:
    """
    Provides recursive package directory module scanning with cache wiping.

    This mixin enables classes to dynamically discover and import submodules
    in a configured package hierarchy. It is typically integrated into registry
    systems to trigger dynamic registration of Pydantic model subclasses at runtime.

    Example:
        .. code-block:: python

            from disdantic.importer import AutoImporterMixin

            class MyRegistry(AutoImporterMixin):
                auto_package = "my_app.models"
                auto_ignore_modules = ["my_app.models.private_model"]

            # Scan and load the modules
            MyRegistry.auto_import_package_modules()
    """

    auto_package: ClassVar[str | Sequence[str] | None] = None
    """The target package or packages to scan for auto-importing."""

    auto_ignore_modules: ClassVar[Sequence[str] | None] = None
    """A list of submodule names or paths to ignore during package discovery."""

    _auto_imported_modules: ClassVar[set[str]] = set()

    @classmethod
    def auto_import_package_modules(cls) -> None:
        """
        Walks configured module layouts recursively and imports submodules.

        This method scans the configured packages, ignoring designated modules,
        and imports all discovered submodules to trigger dynamic registration.

        :raises ValueError: If the class variable 'auto_package' has not been
            properly configured and settings do not specify any auto_packages.
        :raises ImportError: If a target package name cannot be resolved.
        :returns: None.
        """
        packages = cls._resolve_auto_packages()
        if not packages:
            raise MissingPackagesError(
                f"The class variable 'auto_package' must be configured on "
                f"{cls.__name__} or 'auto_packages' configured in settings "
                f"to enable automated package discovery."
            )

        ignore_set = cls._resolve_ignore_set()

        for package_name in packages:
            try:
                package = importlib.import_module(package_name)  # nosemgrep
            except ModuleNotFoundError as err:
                raise ImportError(
                    f"Target auto_package root '{package_name}' could not be resolved."
                ) from err

            if not hasattr(package, "__path__"):
                continue

            for _, module_name, is_pkg in pkgutil.walk_packages(
                package.__path__,
                f"{package.__name__}.",
            ):
                if (
                    is_pkg
                    or module_name in ignore_set
                    or module_name in cls._auto_imported_modules
                ):
                    continue

                if module_name in sys.modules:
                    cls._auto_imported_modules.add(module_name)
                    continue

                importlib.import_module(module_name)  # nosemgrep
                cls._auto_imported_modules.add(module_name)

    @classmethod
    def reset_importer_cache(cls) -> None:
        """
        Purges cached system modules imported by this class.

        Clears the tracked imported submodules from both sys.modules and the
        internal tracking set to guarantee clean test executions.

        :returns: None.
        """
        for module_name in list(cls._auto_imported_modules):
            sys.modules.pop(module_name, None)
        cls._auto_imported_modules.clear()

    @classmethod
    def _resolve_auto_packages(cls) -> list[str]:
        # Resolves and deduplicates the list of target packages to scan.
        packages: list[str] = []
        if cls.auto_package:
            if isinstance(cls.auto_package, str):
                packages.append(cls.auto_package)
            else:
                packages.extend(cls.auto_package)

        settings = get_settings()
        if settings.auto_packages:
            packages.extend(settings.auto_packages)

        # Deduplicate while preserving order
        unique: list[str] = []
        for pkg in packages:
            if pkg not in unique:
                unique.append(pkg)
        return unique

    @classmethod
    def _resolve_ignore_set(cls) -> set[str]:
        # Resolves and merges package/settings ignore lists.
        ignore_set = set(cls.auto_ignore_modules or [])
        settings = get_settings()
        if settings.auto_ignore_modules:
            ignore_set.update(settings.auto_ignore_modules)
        return ignore_set

class DiagnosticsReport(BaseModel):
    """Aggregated health status and discovery details of all checked registries.

    This class serves as the root container returned by the verification pipeline. It
    consolidates individual registry diagnostics, scanned packages, and import failure
    traces into a single report.

    Examples:
        .. code-block:: python

            from disdantic.diagnose import verify_registries

            report = verify_registries()
            if not report.is_healthy:
                print(f"Scanned packages: {report.scanned_packages}")
                print(f"Import errors: {report.import_errors}")
    """

    is_healthy: bool = Field(
        description=(
            "Indicates whether the entire check is clean, enabling safe dynamic model "
            "resolution with zero registry or import failures."
        )
    )
    scanned_packages: list[str] = Field(
        description="Maps the list of Python package names traversed during discovery."
    )
    registries: list[RegistryDiagnostics] = Field(
        default_factory=list,
        description=(
            "Configures the collection of registry-specific diagnostics reports."
        ),
    )
    import_errors: list[str] = Field(
        default_factory=list,
        description=(
            "Maps the traceback strings for import errors encountered during scanning."
        ),
    )

class InfoMixin:
    """Mixin providing runtime self-introspection to generate object structures.

    This mixin allows subclassing models to expose their public attributes,
    properties, slots, and instance dicts as sanitized primitives. It recursively
    inspects objects, resolves lazy loaders or proxies, and handles circular reference
    loops and property extraction errors without raising exceptions.
    """

    @classmethod
    def extract_from_obj(
        cls, obj: Any, visited: set[int] | None = None
    ) -> dict[str, Any]:
        """Parse complex objects into sanitized primitive dictionaries.

        This method recursively crawls the object to extract public fields,
        evaluates custom `.info` hooks if defined, and translates collections
        or nested objects into JSON/YAML compatible dictionaries.
        """
        if visited is None:
            visited = set()

        if not isinstance(obj, type) and obj is not cls:
            try:
                info_class_attr = getattr(type(obj), "info", None)
                if not _is_default_info(info_class_attr) and (
                    info_class_attr is not None or hasattr(obj, "info")
                ):
                    info_val = obj.info
                    raw_info = info_val() if callable(info_val) else info_val
                    return dict(cls._sanitize(raw_info, visited))
            except Exception:  # noqa: BLE001, S110
                pass

        obj_class = getattr(obj, "__class__", type(obj))
        obj_id = id(obj)

        visited.add(obj_id)
        try:
            attributes = cls._extract_attributes(obj, visited)
        finally:
            visited.discard(obj_id)

        return {
            "str": cls._sanitize_fallback(obj),
            "type": obj_class.__name__,
            "module": obj_class.__module__,
            "attributes": attributes,
        }

    def __repr__(self) -> str:
        """__repr__ delegating to MRO overrides or fallback."""
        return self._delegate_dunder("__repr__")

    def __str__(self) -> str:
        """__str__ delegating to MRO overrides or fallback."""
        return self._delegate_dunder("__str__")

    @property
    def info(self) -> dict[str, Any]:
        """Self-introspection dictionary representing the public state."""
        return self.extract_from_obj(self)

    def info_json(
        self,
        *,
        indent: int | None = None,
        sort_keys: bool = False,
        **kwargs: Any,
    ) -> str:
        """Serialize the introspection info dictionary into a valid JSON string."""
        prepared = self._sanitize(self.info, set())
        return json.dumps(prepared, indent=indent, sort_keys=sort_keys, **kwargs)

    def info_yaml(
        self,
        *,
        indent: int | None = None,
        sort_keys: bool = False,
        **kwargs: Any,
    ) -> str:
        """Serialize the introspection info dictionary into a valid YAML string."""
        if yaml is None:
            raise ImportError(
                "PyYAML is required for YAML serialization. "
                "Install disdantic with the 'yaml' extra: pip install disdantic[yaml]"
            )
        prepared = self._sanitize(self.info, set())
        return yaml.dump(prepared, indent=indent, sort_keys=sort_keys, **kwargs)

    def _delegate_dunder(self, name: str) -> str:
        # Delegate repr/str dunder method to MRO overrides or fallback.
        for cls_item in self.__class__.__mro__:
            if cls_item.__name__ in (
                "InfoMixin",
                "object",
                "BaseModel",
                "ReloadableBaseModel",
            ):
                continue
            if name in cls_item.__dict__:
                return str(getattr(cls_item, name)(self))
        return f"<{self.__class__.__name__} info={self.info}>"

    @classmethod
    def _is_class_variable(cls, obj: Any, key: str) -> bool:
        # Check if the key is defined on class, not instance dict/slots
        if isinstance(obj, type):
            return False

        if hasattr(obj, "__dict__") and key in obj.__dict__:
            return False

        # Slots check across MRO
        for mro_cls in type(obj).__mro__:
            slots = getattr(mro_cls, "__slots__", None)
            if slots and key in slots:
                return False

        # Check if the key is defined on any class in the MRO.
        for mro_cls in type(obj).__mro__:
            if key in mro_cls.__dict__:
                class_attr = mro_cls.__dict__[key]
                # If it has a __get__ method (e.g. property, method, descriptor),
                # it's evaluated on the instance and represents instance state.
                return not hasattr(class_attr, "__get__")

        return False

    @classmethod
    def _has_info_protocol(cls, val: Any) -> bool:
        # Check if an object implements the info protocol without evaluating it.
        try:
            return getattr(type(val), "info", None) is not None or (
                hasattr(val, "__dict__") and "info" in val.__dict__
            )
        except Exception:  # noqa: BLE001
            return False

    @classmethod
    def _extract_attributes(cls, obj: Any, visited: set[int]) -> dict[str, Any]:
        # Scrapes attributes from instance spaces, slots, and properties safely.
        attributes: dict[str, Any] = {}

        exclude_keys = set(get_settings().info_exclude_keys)
        is_pydantic = isinstance(obj, BaseModel)
        is_registry = isinstance(obj, RegistryMixin)

        for key in dir(obj):
            if key.startswith("_") or key in exclude_keys:
                continue

            if is_pydantic and key in (
                "model_fields",
                "model_computed_fields",
                "model_config",
                "model_fields_set",
                "model_extra",
            ):
                continue

            if is_registry and key in (
                "registry",
                "registry_auto_discovery",
                "registry_populated",
                "schema_discriminator",
            ):
                continue

            if cls._is_class_variable(obj, key):
                continue

            try:
                val = getattr(obj, key)
                if isinstance(val, collections.abc.Callable):
                    continue

                attributes[key] = cls._sanitize(val, visited)
            except Exception as err:  # noqa: BLE001
                attributes[key] = f"<Extraction Error: {err!r}>"

        return attributes

    @classmethod
    def _sanitize(cls, val: Any, visited: set[int]) -> Any:
        # Unified recursive sanitization and preparation for serialization.
        if isinstance(val, LazyProxy):
            val = val._resolve()  # noqa: SLF001

        val_id = id(val)
        if val_id in visited:
            res = f"<CircularReference: ID {val_id}>"
        elif isinstance(val, PRIMITIVE_TYPES):
            res = val
        elif isinstance(val, dict):
            visited.add(val_id)
            try:
                res = {
                    str(item_key): cls._sanitize(item_val, visited)
                    for item_key, item_val in val.items()
                }
            finally:
                visited.discard(val_id)
        elif isinstance(val, list | tuple | set):
            visited.add(val_id)
            try:
                res = [cls._sanitize(item, visited) for item in val]
            finally:
                visited.discard(val_id)
        elif isinstance(val, type):
            try:
                res = repr(val)
            except Exception:  # noqa: BLE001
                res = f"<{val.__name__} object at {hex(id(val))}>"
        elif cls._has_info_protocol(val):
            res = cls._sanitize_custom(val, val_id, visited)
        else:
            res = cls._sanitize_fallback(val)

        return res

    @classmethod
    def _sanitize_custom(cls, val: Any, val_id: int, visited: set[int]) -> Any:
        # Sanitize a custom object that implements the info protocol.
        try:
            info_class_attr = getattr(type(val), "info", None)
            visited.add(val_id)
            try:
                if _is_default_info(info_class_attr):
                    return cls.extract_from_obj(val, visited)
                info_val = val.info
                raw_info = info_val() if callable(info_val) else info_val
                return cls._sanitize(raw_info, visited)
            finally:
                visited.discard(val_id)
        except Exception:  # noqa: BLE001
            return cls._sanitize_fallback(val)

    @classmethod
    def _sanitize_fallback(cls, val: Any) -> str:
        # Get a safe string representation of val, avoiding infinite loops.
        val_class = getattr(val, "__class__", type(val))
        fallback = f"<{val_class.__name__} object at {hex(id(val))}>"
        try:
            mro = val_class.__mro__
            if not any(mro_cls.__name__ == "InfoMixin" for mro_cls in mro):
                return str(val)
            for cls_item in mro:
                if (
                    cls_item.__name__
                    not in (
                        "InfoMixin",
                        "object",
                        "BaseModel",
                        "ReloadableBaseModel",
                    )
                    and "__str__" in cls_item.__dict__
                ):
                    return str(val)
            return fallback
        except Exception:  # noqa: BLE001
            return fallback

class LazyLoader:
    """Provides thread-safe lazy loading decorators for modules, classes, and variables.

    Encapsulates static utilities to defer package imports and object instantiation.
    It supports lazy module resolution, injecting descriptors into classes, and
    wrapping functional closures inside proxy containers.

    Example:
        .. code-block:: python

            from disdantic.loading import LazyLoader

            # Lazy-load a module namespace
            lazy_sys = LazyLoader.load_module_proxy("sys")
    """

    _global_lock = threading.Lock()

    @classmethod
    def module(cls, module_name: str) -> Callable[[types.ModuleType], types.ModuleType]:
        """Create a decorator to lazy-load module subpackages on demand.

        Example:
            .. code-block:: python

                import sys
                from disdantic.loading import LazyLoader

                @LazyLoader.module("disdantic")
                def loading(mod):
                    pass

        :param module_name: The fully qualified name of the package or module.
        :returns: A decorator function that replaces the target module with
            a lazy module proxy.
        """

        def decorator(_: types.ModuleType) -> types.ModuleType:
            class LazyModule(types.ModuleType):
                def __getattr__(self, name: str) -> Any:
                    with cls._global_lock:
                        try:
                            target_path = f"{module_name}.{name}"
                            return importlib.import_module(target_path)  # nosemgrep
                        except ImportError as error:
                            raise AttributeError(
                                f"Module '{module_name}' has no attribute '{name}'"
                            ) from error

            lazy_mod = LazyModule(module_name)
            sys.modules[module_name] = lazy_mod
            return lazy_mod

        return decorator

    @classmethod
    def class_attributes(
        cls, mapping: dict[str, str | Callable[[], Any]]
    ) -> Callable[[type[TargetT]], type[TargetT]]:
        """Bind lazy property descriptors to class attributes to defer instantiation.

        Accepts a dictionary mapping attribute names to either importable module paths
        or factory callables, and returning a decorator for the target class.

        Example:
            .. code-block:: python

                from disdantic.loading import LazyLoader

                @LazyLoader.class_attributes({"sys": "sys"})
                class MyClass:
                    pass

                instance = MyClass()
                # Accessing sys will load the module lazily
                print(instance.sys.path)

        :param mapping: Dictionary mapping attribute names to module paths or callables.
        :returns: A decorator function that updates the class with lazy attributes.
        """

        def decorator(target_cls: type[TargetT]) -> type[TargetT]:
            for attr_name, target in mapping.items():

                def _create_getter(
                    target_val: str | Callable[[], Any] = target,
                ) -> property:
                    proxy = LazyProxy(
                        lambda: (
                            importlib.import_module(target_val)  # nosemgrep
                            if isinstance(target_val, str)
                            else target_val()
                        )
                    )
                    return property(lambda _: proxy._resolve())  # noqa: SLF001

                setattr(target_cls, attr_name, _create_getter())
            return target_cls

        return decorator

    @classmethod
    def definition(cls, factory: Callable[[], Any]) -> LazyProxy:
        """Create an explicit variable proxy from a functional factory closure.

        Defers execution of the factory callable until attributes are accessed
        on the returned proxy.

        Example:
            .. code-block:: python

                from disdantic.loading import LazyLoader

                proxy = LazyLoader.definition(lambda: [1, 2, 3])
                # Factory remains uncalled until accessed
                print(len(proxy))

        :param factory: A zero-argument callable returning the target object.
        :returns: A LazyProxy instance wrapping the factory function.
        """
        return LazyProxy(factory)

    @classmethod
    def load_module_proxy(cls, fullname: str) -> types.ModuleType:
        """Generate a standard lazy loader proxy directly within the module mapping.

        Resolves the module spec and registers a lazy-loading module in sys.modules,
        only executing the module when its attributes are accessed.

        Example:
            .. code-block:: python

                from disdantic.loading import LazyLoader

                # Loads my_module lazily
                my_mod = LazyLoader.load_module_proxy("my_module")

        :param fullname: The fully qualified name of the module to lazy-load.
        :returns: A module proxy that triggers execution of the module on access.
        :raises ModuleNotFoundError: If the module specified by fullname cannot
            be resolved.
        """
        with cls._global_lock:
            if fullname in sys.modules:
                return sys.modules[fullname]

            spec = importlib.util.find_spec(fullname)
            if not spec or not spec.loader:
                raise ModuleNotFoundError(
                    f"No module named '{fullname}' could be resolved."
                )

            module = importlib.util.module_from_spec(spec)
            sys.modules[fullname] = module

            lazy_loader = importlib.util.LazyLoader(spec.loader)
            lazy_loader.exec_module(module)
            return module

class LazyProxy:
    """Proxy container wrapping modules or factories until an attribute is read.

    Acts as a placeholder for objects whose construction is expensive or requires
    deferred execution. The target object is instantiated and resolved by invoking a
    provided factory function upon the first query of attributes, string representation,
    or directory listing. Resolution is fully thread-safe.

    Example:
        .. code-block:: python

            import types
            from disdantic.loading import LazyProxy

            def expensive_factory():
                return types.SimpleNamespace(data=42)

            proxy = LazyProxy(expensive_factory)
            # expensive_factory is not called yet
            print(proxy.data)  # Triggers resolution and outputs: 42
    """

    def __init__(self, factory: Callable[[], Any]) -> None:
        """Initialize the lazy proxy wrapper with a factory callable.

        :param factory: A zero-argument callable returning the target object to wrap.
        """
        self._factory = factory
        self._wrapped: Any = None
        self._resolved = False
        self._lock = threading.Lock()

    def __getattr__(self, name: str) -> Any:
        """Retrieve attributes from the lazily resolved target object.

        :param name: The name of the attribute to retrieve.
        :returns: The attribute value from the resolved object.
        """
        return getattr(self._resolve(), name)

    def __dir__(self) -> list[str]:
        """Return the directory list of attributes from the resolved target object.

        :returns: List of attributes available on the wrapped object.
        """
        return dir(self._resolve())

    def __repr__(self) -> str:
        """Return a string representation of the proxy or the wrapped target.

        :returns: A string indicating initialization state or the wrapped
            representation.
        """
        if not self._resolved:
            return f"<LazyProxy for uninitialized factory: {self._factory}>"
        return repr(self._wrapped)

    def _resolve(self) -> Any:
        # Double-checked locking implementation preventing concurrent serialization
        # or import errors.
        if not self._resolved:
            with self._lock:
                if not self._resolved:
                    self._wrapped = self._factory()
                    self._resolved = True
        return self._wrapped