diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index d4434aa..24822a0 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -4,13 +4,15 @@ on: push: branches: - "**" + pull_request: + branches: ["main", "develop", "release"] jobs: tests: name: "Python ${{ matrix.python-version }}" runs-on: "ubuntu-latest" strategy: matrix: - python-version: ["3.9", "3.10", "3.11", "3.12"] + python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"] steps: - uses: "actions/checkout@v4" - uses: "actions/setup-python@v5" diff --git a/docs/release-notes.md b/docs/release-notes.md index 75b0a57..cc26a8f 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -1,5 +1,14 @@ # Release notes + +## Version 0.1.0 + +### Changed + +- Internals refactored. `base.py` is splitted now in multiple submodules. +- Allow different settings than pydantic_settings. +- Switch to semantic versioning. + ## Version 0.0.9 ### Added diff --git a/docs/tutorial.md b/docs/tutorial.md index 3628ccc..6b4ef7d 100644 --- a/docs/tutorial.md +++ b/docs/tutorial.md @@ -6,8 +6,6 @@ ``` shell pip install monkay -# or with pydantic_settings -# pip install monkay[settings] ``` ### Usage @@ -81,12 +79,9 @@ There is also a method: Which can be used to clear the caches. -##### Monkey - - #### Using settings -Settings can be an initialized pydantic settings variable or a class. +Settings pointed at via the string can be an initialized settings variable or a class. When pointing to a class the class is automatically called without arguments. Let's do the configuration like Django via environment variable: @@ -145,6 +140,45 @@ monkay = Monkay( ) ``` +##### Other settings libraries + +Here we just use pydantic_settings. But settings can be everything from a dictionary, to a dataclass. +Only requirement is that names are resolvable as attributes or as keys.` + + +``` python title="explicit_settings.py" +from typing import TypedDict + +class Settings(TypedDict): + preloads: list[str] + extensions: list[Any] + foo: str + +settings = Settings(preloads=[], extensions=[], foo="hello") +# or just a dictionary +# settings = {"preloads": [], "extensions": [], "foo": "hello"} +``` + +and + +``` python title="__init__.py" +import os +from monkay import Monkay, get_value_from_settings +monkay = Monkay( + globals(), + with_extensions=True, + with_instance=True, + settings_path=os.environ.get("MONKAY_SETTINGS", "example.default.path.settings:settings"), + settings_preloads_name="preloads", + settings_extensions_name="extensions", +) + +# attribute grabber with fallback to items +get_value_from_settings(monkay.settings, "foo") +``` + + + #### Pathes Like shown in the examples pathes end with a `:` for an attribute. But sometimes a dot is nicer. diff --git a/mkdocs.yml b/mkdocs.yml index c6158e8..a8fdb4b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -2,9 +2,59 @@ site_name: Monkay site_description: The ultimate preload, settings, lazy import manager.. site_url: https://dymmond.github.io/monkay +theme: + name: "material" + language: en + palette: + - scheme: "default" + primary: "light green" + accent: "red" + media: "(prefers-color-scheme: light)" + toggle: + icon: "material/lightbulb" + name: "Switch to dark mode" + - scheme: "slate" + media: "(prefers-color-scheme: dark)" + primary: "green" + accent: "red" + toggle: + icon: "material/lightbulb-outline" + name: "Switch to light mode" + + features: + - search.suggest + - search.highlight + - content.tabs.link + - content.code.copy + - content.code.annotate + - content.tooltips + - content.code.select + - navigation.indexes + - navigation.path + - navigation.tabs + + nav: - Home: index.md - Tutorial: tutorial.md - Helpers: helpers.md - Testing: testing.md - Specials: specials.md + - Release Notes: release-notes.md + +markdown_extensions: + - attr_list + - toc: + permalink: true + - mdx_include: + base_path: docs + - admonition + - extra + - pymdownx.superfences: + custom_fences: + - name: mermaid + class: mermaid + format: !!python/name:pymdownx.superfences.fence_code_format "" + - pymdownx.tabbed: + alternate_style: true + - md_in_html diff --git a/monkay/__about__.py b/monkay/__about__.py index cac369b..8943940 100644 --- a/monkay/__about__.py +++ b/monkay/__about__.py @@ -1,4 +1,4 @@ # SPDX-FileCopyrightText: 2024-present alex # # SPDX-License-Identifier: BSD-3-Clauses -__version__ = "0.0.9" +__version__ = "0.1.0" diff --git a/monkay/__init__.py b/monkay/__init__.py index 7df3868..cfe0c4b 100644 --- a/monkay/__init__.py +++ b/monkay/__init__.py @@ -3,12 +3,19 @@ # SPDX-License-Identifier: BSD-3-Clauses from .base import ( + InGlobalsDict, + absolutify_import, + get_value_from_settings, + load, + load_any, +) +from .core import ( + Monkay, +) +from .types import ( PRE_ADD_LAZY_IMPORT_HOOK, DeprecatedImport, ExtensionProtocol, - Monkay, - load, - load_any, ) __all__ = [ @@ -20,4 +27,6 @@ "load", "load_any", "absolutify_import", + "InGlobalsDict", + "get_value_from_settings", ] diff --git a/monkay/_monkay_exports.py b/monkay/_monkay_exports.py new file mode 100644 index 0000000..5f3ce9d --- /dev/null +++ b/monkay/_monkay_exports.py @@ -0,0 +1,306 @@ +from __future__ import annotations + +import warnings +from collections.abc import Callable, Collection +from importlib import import_module +from inspect import isclass, ismodule +from itertools import chain +from typing import ( + Any, + Literal, + cast, +) + +from .base import InGlobalsDict, absolutify_import, load +from .types import PRE_ADD_LAZY_IMPORT_HOOK, DeprecatedImport, SortedExportsEntry + + +def _stub_previous_getattr(name: str) -> Any: + raise AttributeError(f'Module has no attribute: "{name}" (Monkay).') + + +def _obj_to_full_name(obj: Any) -> str: + if ismodule(obj): + return obj.__spec__.name # type: ignore + if not isclass(obj): + obj = type(obj) + return f"{obj.__module__}.{obj.__qualname__}" + + +class MonkayExports: + package: str | None + getter: Callable[..., Any] + globals_dict: dict + _cached_imports: dict[str, Any] + pre_add_lazy_import_hook: None | PRE_ADD_LAZY_IMPORT_HOOK + post_add_lazy_import_hook: None | Callable[[str], None] + lazy_imports: dict[str, str | Callable[[], Any]] + deprecated_lazy_imports: dict[str, DeprecatedImport] + uncached_imports: set[str] + + def find_missing( + self, + *, + all_var: bool | Collection[str] = True, + search_pathes: None | Collection[str] = None, + ignore_deprecated_import_errors: bool = False, + require_search_path_all_var: bool = True, + ) -> dict[ + str, + set[ + Literal[ + "not_in_all_var", + "missing_attr", + "missing_all_var", + "import", + "shadowed", + "search_path_extra", + "search_path_import", + ] + ], + ]: + """Debug method to check""" + + assert self.getter is not None + missing: dict[ + str, + set[ + Literal[ + "not_in_all_var", + "missing_attr", + "missing_all_var", + "import", + "shadowed", + "search_path_extra", + "search_path_import", + ] + ], + ] = {} + if all_var is True: + try: + all_var = self.getter("__all__", check_globals_dict=True) + except AttributeError: + missing.setdefault(self.globals_dict["__spec__"].name, set()).add( + "missing_all_var" + ) + all_var = [] + key_set = set(chain(self.lazy_imports.keys(), self.deprecated_lazy_imports.keys())) + value_pathes_set: set[str] = set() + for name in key_set: + found_path: str = "" + if name in self.lazy_imports and isinstance(self.lazy_imports[name], str): + found_path = cast(str, self.lazy_imports[name]).replace(":", ".") + elif name in self.deprecated_lazy_imports and isinstance( + self.deprecated_lazy_imports[name]["path"], str + ): + found_path = cast(str, self.deprecated_lazy_imports[name]["path"]).replace( + ":", "." + ) + if found_path: + value_pathes_set.add(absolutify_import(found_path, self.package)) + try: + obj = self.getter(name, no_warn_deprecated=True, check_globals_dict="fail") + # also add maybe rexported path + value_pathes_set.add(_obj_to_full_name(obj)) + except InGlobalsDict: + missing.setdefault(name, set()).add("shadowed") + except ImportError: + if not ignore_deprecated_import_errors or name not in self.deprecated_lazy_imports: + missing.setdefault(name, set()).add("import") + if all_var is not False: + for export_name in cast(Collection[str], all_var): + try: + obj = self.getter( + export_name, no_warn_deprecated=True, check_globals_dict=True + ) + except AttributeError: + missing.setdefault(export_name, set()).add("missing_attr") + continue + if export_name not in key_set: + value_pathes_set.add(_obj_to_full_name(obj)) + + if search_pathes: + for search_path in search_pathes: + try: + mod = import_module(search_path, self.package) + except ImportError: + missing.setdefault(search_path, set()).add("search_path_import") + continue + try: + all_var_search = mod.__all__ + except AttributeError: + if require_search_path_all_var: + missing.setdefault(search_path, set()).add("missing_all_var") + + continue + for export_name in all_var_search: + export_path = absolutify_import(f"{search_path}.{export_name}", self.package) + try: + # for re-exports + obj = getattr(mod, export_name) + except AttributeError: + missing.setdefault(export_path, set()).add("missing_attr") + # still check check the export path + if export_path not in value_pathes_set: + missing.setdefault(export_path, set()).add("search_path_extra") + continue + if ( + export_path not in value_pathes_set + and _obj_to_full_name(obj) not in value_pathes_set + ): + missing.setdefault(export_path, set()).add("search_path_extra") + + if all_var is not False: + for name in key_set.difference(cast(Collection[str], all_var)): + missing.setdefault(name, set()).add("not_in_all_var") + + return missing + + def add_lazy_import( + self, name: str, value: str | Callable[[], Any], *, no_hooks: bool = False + ) -> None: + if not no_hooks and self.pre_add_lazy_import_hook is not None: + name, value = self.pre_add_lazy_import_hook(name, value, "lazy_import") + if name in self.lazy_imports: + raise KeyError(f'"{name}" is already a lazy import') + if name in self.deprecated_lazy_imports: + raise KeyError(f'"{name}" is already a deprecated lazy import') + self.lazy_imports[name] = value + if not no_hooks and self.post_add_lazy_import_hook is not None: + self.post_add_lazy_import_hook(name) + + def add_deprecated_lazy_import( + self, name: str, value: DeprecatedImport, *, no_hooks: bool = False + ) -> None: + if not no_hooks and self.pre_add_lazy_import_hook is not None: + name, value = self.pre_add_lazy_import_hook(name, value, "deprecated_lazy_import") + if name in self.lazy_imports: + raise KeyError(f'"{name}" is already a lazy import') + if name in self.deprecated_lazy_imports: + raise KeyError(f'"{name}" is already a deprecated lazy import') + self.deprecated_lazy_imports[name] = value + if not no_hooks and self.post_add_lazy_import_hook is not None: + self.post_add_lazy_import_hook(name) + + def sorted_exports( + self, + all_var: Collection[str] | None = None, + *, + separate_by_category: bool = True, + sort_by: Literal["export_name", "path"] = "path", + ) -> list[SortedExportsEntry]: + if all_var is None: + all_var = self.globals_dict["__all__"] + sorted_exports: list[SortedExportsEntry] = [] + # ensure all entries are only returned once + for name in set(all_var): + if name in self.lazy_imports: + sorted_exports.append( + SortedExportsEntry( + "lazy_import", + name, + cast( + str, + self.lazy_imports[name] + if isinstance(self.lazy_imports[name], str) + else f"{self.globals_dict['__spec__'].name}.{name}", + ), + ) + ) + elif name in self.deprecated_lazy_imports: + sorted_exports.append( + SortedExportsEntry( + "deprecated_lazy_import", + name, + cast( + str, + self.deprecated_lazy_imports[name]["path"] + if isinstance(self.deprecated_lazy_imports[name]["path"], str) + else f"{self.globals_dict['__spec__'].name}.{name}", + ), + ) + ) + else: + sorted_exports.append( + SortedExportsEntry( + "other", + name, + f"{self.globals_dict['__spec__'].name}.{name}", + ) + ) + if separate_by_category: + + def key_fn(ordertuple: SortedExportsEntry) -> tuple: + return ordertuple.category, getattr(ordertuple, sort_by) + else: + + def key_fn(ordertuple: SortedExportsEntry) -> tuple: + return (getattr(ordertuple, sort_by),) + + sorted_exports.sort(key=key_fn) + return sorted_exports + + def module_getter( + self, + key: str, + *, + chained_getter: Callable[[str], Any] = _stub_previous_getattr, + no_warn_deprecated: bool = False, + check_globals_dict: bool | Literal["fail"] = False, + ) -> Any: + """ + Module Getter which handles lazy imports. + The injected version containing a potential found __getattr__ handler as chained_getter + is availabe as getter attribute. + """ + if check_globals_dict and key in self.globals_dict: + if check_globals_dict == "fail": + raise InGlobalsDict(f'"{key}" is defined as real variable.') + return self.globals_dict[key] + lazy_import = self.lazy_imports.get(key) + if lazy_import is None: + deprecated = self.deprecated_lazy_imports.get(key) + if deprecated is not None: + lazy_import = deprecated["path"] + if not no_warn_deprecated: + warn_strs = [f'Attribute: "{key}" is deprecated.'] + if deprecated.get("reason"): + # Note: no dot is added, this is the responsibility of the reason author. + warn_strs.append(f"Reason: {deprecated['reason']}") + if deprecated.get("new_attribute"): + warn_strs.append(f'Use "{deprecated["new_attribute"]}" instead.') + warnings.warn("\n".join(warn_strs), DeprecationWarning, stacklevel=2) + + if lazy_import is None: + return chained_getter(key) + if key not in self._cached_imports or key in self.uncached_imports: + if callable(lazy_import): + value: Any = lazy_import() + else: + value = load(lazy_import, package=self.package) + if key in self.uncached_imports: + return value + else: + self._cached_imports[key] = value + return self._cached_imports[key] + + def update_all_var(self, all_var: Collection[str]) -> list[str] | set[str]: + if isinstance(all_var, set): + all_var_set = all_var + else: + if not isinstance(all_var, list): + all_var = list(all_var) + all_var_set = set(all_var) + + if self.lazy_imports or self.deprecated_lazy_imports: + for var in chain( + self.lazy_imports, + self.deprecated_lazy_imports, + ): + if var not in all_var_set: + if isinstance(all_var, list): + all_var.append(var) + else: + cast(set[str], all_var).add(var) + + return cast("list[str] | set[str]", all_var) diff --git a/monkay/_monkay_extensions.py b/monkay/_monkay_extensions.py new file mode 100644 index 0000000..fe7434e --- /dev/null +++ b/monkay/_monkay_extensions.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +from collections.abc import Callable, Generator, Iterable +from contextlib import contextmanager +from contextvars import ContextVar +from inspect import isclass +from typing import TYPE_CHECKING, Any, Generic, Literal, cast + +from .types import INSTANCE, SETTINGS, ExtensionProtocol + +if TYPE_CHECKING: + from .core import Monkay + + +class MonkayExtensions(Generic[INSTANCE, SETTINGS]): + # extensions are pretended to always exist, we check the _extensions_var + _extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] + _extensions_var: None | ContextVar[None | dict[str, ExtensionProtocol[INSTANCE, SETTINGS]]] = ( + None + ) + # pretend it always exists + _extensions_applied_var: ContextVar[set[str] | None] + extension_order_key_fn: None | Callable[[ExtensionProtocol[INSTANCE, SETTINGS]], Any] + # in truth a property + instance: INSTANCE | None + + def apply_extensions(self, *, use_overwrite: bool = True) -> None: + assert self._extensions_var is not None, "Monkay not enabled for extensions" + extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None = ( + self._extensions_var.get() if use_overwrite else None + ) + if extensions is None: + extensions = self._extensions + extensions_applied = self._extensions_applied_var.get() + if extensions_applied is not None: + raise RuntimeError("Other apply process in the same context is active.") + extensions_ordered: Iterable[tuple[str, ExtensionProtocol[INSTANCE, SETTINGS]]] = cast( + dict[str, ExtensionProtocol[INSTANCE, SETTINGS]], extensions + ).items() + + if self.extension_order_key_fn is not None: + extensions_ordered = sorted( + extensions_ordered, + key=self.extension_order_key_fn, # type: ignore + ) + extensions_applied = set() + token = self._extensions_applied_var.set(extensions_applied) + try: + for name, extension in extensions_ordered: + if name in extensions_applied: + continue + # despite slightly inaccurate (added before applying actually) this ensures that no loops appear + extensions_applied.add(name) + extension.apply(cast("Monkay[INSTANCE, SETTINGS]", self)) + finally: + self._extensions_applied_var.reset(token) + + def ensure_extension( + self, name_or_extension: str | ExtensionProtocol[INSTANCE, SETTINGS] + ) -> None: + assert self._extensions_var is not None, "Monkay not enabled for extensions." + extensions_applied = self._extensions_applied_var.get() + assert extensions_applied is not None, "Applying extensions not active." + extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None = ( + self._extensions_var.get() + ) + if extensions is None: + extensions = self._extensions + if isinstance(name_or_extension, str): + name = name_or_extension + extension = extensions.get(name) + elif not isclass(name_or_extension) and isinstance(name_or_extension, ExtensionProtocol): + name = name_or_extension.name + extension = extensions.get(name, name_or_extension) + else: + raise RuntimeError( + 'Provided extension "{name_or_extension}" does not implement the ExtensionProtocol' + ) + if name in extensions_applied: + return + + if extension is None: + raise RuntimeError(f'Extension: "{name}" does not exist.') + # despite slightly inaccurate (added before applying actually) this ensures that no loops appear + extensions_applied.add(name) + extension.apply(cast("Monkay[INSTANCE, SETTINGS]", self)) + + def add_extension( + self, + extension: ExtensionProtocol[INSTANCE, SETTINGS] + | type[ExtensionProtocol[INSTANCE, SETTINGS]] + | Callable[[], ExtensionProtocol[INSTANCE, SETTINGS]], + *, + use_overwrite: bool = True, + on_conflict: Literal["error", "keep", "replace"] = "error", + ) -> None: + assert self._extensions_var is not None, "Monkay not enabled for extensions" + extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None = ( + self._extensions_var.get() if use_overwrite else None + ) + if extensions is None: + extensions = self._extensions + if callable(extension) or isclass(extension): + extension = extension() + if not isinstance(extension, ExtensionProtocol): + raise ValueError(f"Extension {extension} is not compatible") + if extension.name in extensions: + if on_conflict == "error": + raise KeyError(f'Extension "{extension.name}" already exists.') + elif on_conflict == "keep": + return + extensions[extension.name] = extension + + @contextmanager + def with_extensions( + self, + extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None, + *, + apply_extensions: bool = False, + ) -> Generator[dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None]: + # why None, for temporary using the real extensions + assert self._extensions_var is not None, "Monkay not enabled for extensions" + token = self._extensions_var.set(extensions) + try: + if apply_extensions and self.instance is not None: + self.apply_extensions() + yield extensions + finally: + self._extensions_var.reset(token) diff --git a/monkay/_monkay_instance.py b/monkay/_monkay_instance.py new file mode 100644 index 0000000..55a1eb4 --- /dev/null +++ b/monkay/_monkay_instance.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +from collections.abc import Generator +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Generic + +from ._monkay_extensions import MonkayExtensions +from .types import INSTANCE, SETTINGS + + +class MonkayInstance(MonkayExtensions[INSTANCE, SETTINGS], Generic[INSTANCE, SETTINGS]): + _instance: None | INSTANCE = None + _instance_var: ContextVar[INSTANCE | None] | None = None + + @property + def instance(self) -> INSTANCE | None: # type: ignore + assert self._instance_var is not None, "Monkay not enabled for instances" + instance: INSTANCE | None = self._instance_var.get() + if instance is None: + instance = self._instance + return instance + + def set_instance( + self, + instance: INSTANCE | None, + *, + apply_extensions: bool = True, + use_extensions_overwrite: bool = True, + ) -> INSTANCE | None: + assert self._instance_var is not None, "Monkay not enabled for instances" + # need to address before the instance is swapped + if ( + apply_extensions + and self._extensions_var is not None + and self._extensions_applied_var.get() is not None + ): + raise RuntimeError("Other apply process in the same context is active.") + self._instance = instance + if apply_extensions and instance is not None and self._extensions_var is not None: + # unapply a potential instance overwrite + with self.with_instance(None): + self.apply_extensions(use_overwrite=use_extensions_overwrite) + return instance + + @contextmanager + def with_instance( + self, + instance: INSTANCE | None, + *, + apply_extensions: bool = False, + use_extensions_overwrite: bool = True, + ) -> Generator[INSTANCE | None]: + assert self._instance_var is not None, "Monkay not enabled for instances" + # need to address before the instance is swapped + if ( + apply_extensions + and self._extensions_var is not None + and self._extensions_applied_var.get() is not None + ): + raise RuntimeError("Other apply process in the same context is active.") + token = self._instance_var.set(instance) + try: + if apply_extensions and self._extensions_var is not None: + self.apply_extensions(use_overwrite=use_extensions_overwrite) + yield instance + finally: + self._instance_var.reset(token) diff --git a/monkay/_monkay_settings.py b/monkay/_monkay_settings.py new file mode 100644 index 0000000..c7dcd32 --- /dev/null +++ b/monkay/_monkay_settings.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from collections.abc import Callable, Generator +from contextlib import contextmanager +from contextvars import ContextVar +from functools import cached_property +from inspect import isclass +from typing import Generic, cast + +from .base import load +from .types import SETTINGS + + +class MonkaySettings(Generic[SETTINGS]): + package: str | None + settings_preloads_name: str + settings_extensions_name: str + _settings_var: ContextVar[SETTINGS | None] | None = None + _settings_definition: SETTINGS | type[SETTINGS] | str | Callable[[], SETTINGS] | None = None + + @cached_property + def _loaded_settings(self) -> SETTINGS | None: + # only class and string pathes + if isclass(self._settings_definition): + return self._settings_definition() + assert isinstance( + self._settings_definition, str + ), f"Not a settings object: {self._settings_definition}" + if not self._settings_definition: + return None + settings: SETTINGS | type[SETTINGS] = load(self._settings_definition, package=self.package) + if isclass(settings): + settings = settings() + return cast(SETTINGS, settings) + + @property + def settings(self) -> SETTINGS: + assert self._settings_var is not None, "Monkay not enabled for settings" + settings: SETTINGS | Callable[[], SETTINGS] | None = self._settings_var.get() + if settings is None: + # when settings_path is callable bypass the cache, for forwards + settings = ( + self._loaded_settings + if isinstance(self._settings_definition, str) or isclass(self._settings_definition) + else self._settings_definition + ) + if callable(settings): + settings = settings() + if settings is None: + raise RuntimeError( + "Settings are not set yet. Returned settings are None or settings_path is empty." + ) + return settings + + @settings.setter + def settings( + self, value: str | Callable[[], SETTINGS] | SETTINGS | type[SETTINGS] | None + ) -> None: + assert self._settings_var is not None, "Monkay not enabled for settings" + if not value: + self._settings_definition = "" + return + if not isinstance(value, str) and not callable(value) and not isclass(value): + self._settings_definition = lambda: value + else: + self._settings_definition = value + del self.settings + + @settings.deleter + def settings(self) -> None: + # clear cache + self.__dict__.pop("_loaded_settings", None) + + @contextmanager + def with_settings(self, settings: SETTINGS | None) -> Generator[SETTINGS | None]: + assert self._settings_var is not None, "Monkay not enabled for settings" + # why None, for temporary using the real settings + token = self._settings_var.set(settings) + try: + yield settings + finally: + self._settings_var.reset(token) diff --git a/monkay/base.py b/monkay/base.py index 652d312..efdc29e 100644 --- a/monkay/base.py +++ b/monkay/base.py @@ -1,75 +1,9 @@ from __future__ import annotations import warnings -from collections.abc import Callable, Collection, Generator, Iterable -from contextlib import contextmanager -from contextvars import ContextVar -from functools import cached_property, partial +from collections.abc import Collection from importlib import import_module -from inspect import isclass, ismodule -from itertools import chain -from typing import ( - TYPE_CHECKING, - Any, - Generic, - Literal, - NamedTuple, - Protocol, - TypedDict, - TypeVar, - cast, - overload, - runtime_checkable, -) - -if TYPE_CHECKING: - from pydantic_settings import BaseSettings - -INSTANCE = TypeVar("INSTANCE") -SETTINGS = TypeVar("SETTINGS", bound="BaseSettings") - - -class SortedExportsEntry(NamedTuple): - category: Literal["other", "lazy_import", "deprecated_lazy_import"] - export_name: str - path: str - - -class DeprecatedImport(TypedDict, total=False): - path: str | Callable[[], Any] - reason: str - new_attribute: str - - -DeprecatedImport.__required_keys__ = frozenset({"deprecated"}) - - -class PRE_ADD_LAZY_IMPORT_HOOK(Protocol): - @overload - @staticmethod - def __call__( - key: str, - value: str | Callable[[], Any], - type_: Literal["lazy_import"], - /, - ) -> tuple[str, str | Callable[[], Any]]: ... - - @overload - @staticmethod - def __call__( - key: str, - value: DeprecatedImport, - type_: Literal["deprecated_lazy_import"], - /, - ) -> tuple[str, DeprecatedImport]: ... - - @staticmethod - def __call__( - key: str, - value: str | Callable[[], Any] | DeprecatedImport, - type_: Literal["lazy_import", "deprecated_lazy_import"], - /, - ) -> tuple[str, str | Callable[[], Any] | DeprecatedImport]: ... +from typing import Any def load(path: str, *, allow_splits: str = ":.", package: None | str = None) -> Any: @@ -125,594 +59,12 @@ def absolutify_import(import_path: str, package: str | None) -> str: return f"{package}.{import_path.lstrip('.')}" -@runtime_checkable -class ExtensionProtocol(Protocol[INSTANCE, SETTINGS]): - name: str - - def apply(self, monkay_instance: Monkay[INSTANCE, SETTINGS]) -> None: ... - - class InGlobalsDict(Exception): pass -def _stub_previous_getattr(name: str) -> Any: - raise AttributeError(f'Module has no attribute: "{name}" (Monkay).') - - -def _obj_to_full_name(obj: Any) -> str: - if ismodule(obj): - return obj.__spec__.name # type: ignore - if not isclass(obj): - obj = type(obj) - return f"{obj.__module__}.{obj.__qualname__}" - - -class Monkay(Generic[INSTANCE, SETTINGS]): - getter: Callable[..., Any] - _instance: None | INSTANCE = None - _instance_var: ContextVar[INSTANCE | None] | None = None - # extensions are pretended to always exist, we check the _extensions_var - _extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] - _extensions_var: None | ContextVar[None | dict[str, ExtensionProtocol[INSTANCE, SETTINGS]]] = None - _extensions_applied: None | ContextVar[dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None] = None - _settings_var: ContextVar[SETTINGS | None] | None = None - _settings_definition: SETTINGS | type[SETTINGS] | str | Callable[[], SETTINGS] | None = None - - def __init__( - self, - globals_dict: dict, - *, - with_instance: str | bool = False, - with_extensions: str | bool = False, - extension_order_key_fn: None | Callable[[ExtensionProtocol[INSTANCE, SETTINGS]], Any] = None, - settings_path: str | Callable[[], SETTINGS] | SETTINGS | type[SETTINGS] | None = None, - preloads: Iterable[str] = (), - settings_preloads_name: str = "", - settings_extensions_name: str = "", - uncached_imports: Iterable[str] = (), - lazy_imports: dict[str, str | Callable[[], Any]] | None = None, - deprecated_lazy_imports: dict[str, DeprecatedImport] | None = None, - settings_ctx_name: str = "monkay_settings_ctx", - extensions_applied_ctx_name: str = "monkay_extensions_applied_ctx", - skip_all_update: bool = False, - evaluate_settings: bool = True, - pre_add_lazy_import_hook: None | PRE_ADD_LAZY_IMPORT_HOOK = None, - post_add_lazy_import_hook: None | Callable[[str], None] = None, - package: str | None = "", - ) -> None: - self.globals_dict = globals_dict - if with_instance is True: - with_instance = "monkay_instance_ctx" - with_instance = with_instance - if with_extensions is True: - with_extensions = "monkay_extensions_ctx" - with_extensions = with_extensions - if package == "" and globals_dict.get("__spec__"): - package = globals_dict["__spec__"].parent - self.package = package or None - - self._cached_imports: dict[str, Any] = {} - self.pre_add_lazy_import_hook: None | PRE_ADD_LAZY_IMPORT_HOOK = pre_add_lazy_import_hook - self.post_add_lazy_import_hook = post_add_lazy_import_hook - self.uncached_imports: set[str] = set(uncached_imports) - self.lazy_imports: dict[str, str | Callable[[], Any]] = {} - self.deprecated_lazy_imports: dict[str, DeprecatedImport] = {} - if lazy_imports: - for name, lazy_import in lazy_imports.items(): - self.add_lazy_import(name, lazy_import, no_hooks=True) - if deprecated_lazy_imports: - for name, deprecated_import in deprecated_lazy_imports.items(): - self.add_deprecated_lazy_import(name, deprecated_import, no_hooks=True) - if settings_path is not None: - self._settings_var = globals_dict[settings_ctx_name] = ContextVar(settings_ctx_name, default=None) - self.settings = settings_path # type: ignore - self.settings_preloads_name = settings_preloads_name - self.settings_extensions_name = settings_extensions_name - - if with_instance: - self._instance_var = globals_dict[with_instance] = ContextVar(with_instance, default=None) - if with_extensions: - self.extension_order_key_fn = extension_order_key_fn - self._extensions = {} - self._extensions_var = globals_dict[with_extensions] = ContextVar(with_extensions, default=None) - self._extensions_applied_var = globals_dict[extensions_applied_ctx_name] = ContextVar( - extensions_applied_ctx_name, default=None - ) - if self.lazy_imports or self.deprecated_lazy_imports: - getter: Callable[..., Any] = self.module_getter - if "__getattr__" in globals_dict: - getter = partial(getter, chained_getter=globals_dict["__getattr__"]) - globals_dict["__getattr__"] = self.getter = getter - if not skip_all_update: - all_var = globals_dict.setdefault("__all__", []) - globals_dict["__all__"] = self.update_all_var(all_var) - for preload in preloads: - splitted = preload.rsplit(":", 1) - try: - module = import_module(splitted[0], self.package) - except ImportError: - module = None - if module is not None and len(splitted) == 2: - getattr(module, splitted[1])() - if evaluate_settings and self._settings_definition: - # disables overwrite - with self.with_settings(None): - self.evaluate_settings(on_conflict="error") - - def clear_caches(self, settings_cache: bool = True, import_cache: bool = True) -> None: - if settings_cache: - del self.settings - if import_cache: - self._cached_imports.clear() - - @property - def instance(self) -> INSTANCE | None: - assert self._instance_var is not None, "Monkay not enabled for instances" - instance: INSTANCE | None = self._instance_var.get() - if instance is None: - instance = self._instance - return instance - - def set_instance( - self, - instance: INSTANCE | None, - *, - apply_extensions: bool = True, - use_extensions_overwrite: bool = True, - ) -> INSTANCE | None: - assert self._instance_var is not None, "Monkay not enabled for instances" - # need to address before the instance is swapped - if apply_extensions and self._extensions_applied_var.get() is not None: - raise RuntimeError("Other apply process in the same context is active.") - self._instance = instance - if apply_extensions and instance is not None and self._extensions_var is not None: - # unapply a potential instance overwrite - with self.with_instance(None): - self.apply_extensions(use_overwrite=use_extensions_overwrite) - return instance - - @contextmanager - def with_instance( - self, - instance: INSTANCE | None, - *, - apply_extensions: bool = False, - use_extensions_overwrite: bool = True, - ) -> Generator[INSTANCE | None]: - assert self._instance_var is not None, "Monkay not enabled for instances" - # need to address before the instance is swapped - if apply_extensions and self._extensions_var is not None and self._extensions_applied_var.get() is not None: - raise RuntimeError("Other apply process in the same context is active.") - token = self._instance_var.set(instance) - try: - if apply_extensions and self._extensions_var is not None: - self.apply_extensions(use_overwrite=use_extensions_overwrite) - yield instance - finally: - self._instance_var.reset(token) - - def apply_extensions(self, *, use_overwrite: bool = True) -> None: - assert self._extensions_var is not None, "Monkay not enabled for extensions" - extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None = ( - self._extensions_var.get() if use_overwrite else None - ) - if extensions is None: - extensions = self._extensions - extensions_applied = self._extensions_applied_var.get() - if extensions_applied is not None: - raise RuntimeError("Other apply process in the same context is active.") - extensions_ordered: Iterable[tuple[str, ExtensionProtocol[INSTANCE, SETTINGS]]] = cast( - dict[str, ExtensionProtocol[INSTANCE, SETTINGS]], extensions - ).items() - - if self.extension_order_key_fn is not None: - extensions_ordered = sorted( - extensions_ordered, - key=self.extension_order_key_fn, # type: ignore - ) - extensions_applied = set() - token = self._extensions_applied_var.set(extensions_applied) - try: - for name, extension in extensions_ordered: - if name in extensions_applied: - continue - extensions_applied.add(name) - extension.apply(self) - finally: - self._extensions_applied_var.reset(token) - - def ensure_extension(self, name_or_extension: str | ExtensionProtocol[INSTANCE, SETTINGS]) -> None: - assert self._extensions_var is not None, "Monkay not enabled for extensions" - extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None = self._extensions_var.get() - if extensions is None: - extensions = self._extensions - if isinstance(name_or_extension, str): - name = name_or_extension - extension = extensions.get(name) - elif not isclass(name_or_extension) and isinstance(name_or_extension, ExtensionProtocol): - name = name_or_extension.name - extension = extensions.get(name, name_or_extension) - else: - raise RuntimeError('Provided extension "{name_or_extension}" does not implement the ExtensionProtocol') - if name in self._extensions_applied_var.get(): - return - - if extension is None: - raise RuntimeError(f'Extension: "{name}" does not exist.') - self._extensions_applied_var.get().add(name) - extension.apply(self) - - def add_extension( - self, - extension: ExtensionProtocol[INSTANCE, SETTINGS] - | type[ExtensionProtocol[INSTANCE, SETTINGS]] - | Callable[[], ExtensionProtocol[INSTANCE, SETTINGS]], - *, - use_overwrite: bool = True, - on_conflict: Literal["error", "keep", "replace"] = "error", - ) -> None: - assert self._extensions_var is not None, "Monkay not enabled for extensions" - extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None = ( - self._extensions_var.get() if use_overwrite else None - ) - if extensions is None: - extensions = self._extensions - if callable(extension) or isclass(extension): - extension = extension() - if not isinstance(extension, ExtensionProtocol): - raise ValueError(f"Extension {extension} is not compatible") - if extension.name in extensions: - if on_conflict == "error": - raise KeyError(f'Extension "{extension.name}" already exists.') - elif on_conflict == "keep": - return - extensions[extension.name] = extension - - @contextmanager - def with_extensions( - self, - extensions: dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None, - *, - apply_extensions: bool = False, - ) -> Generator[dict[str, ExtensionProtocol[INSTANCE, SETTINGS]] | None]: - # why None, for temporary using the real extensions - assert self._extensions_var is not None, "Monkay not enabled for extensions" - token = self._extensions_var.set(extensions) - try: - if apply_extensions and self.instance is not None: - self.apply_extensions() - yield extensions - finally: - self._extensions_var.reset(token) - - def update_all_var(self, all_var: Collection[str]) -> list[str] | set[str]: - if isinstance(all_var, set): - all_var_set = all_var - else: - if not isinstance(all_var, list): - all_var = list(all_var) - all_var_set = set(all_var) - - if self.lazy_imports or self.deprecated_lazy_imports: - for var in chain( - self.lazy_imports, - self.deprecated_lazy_imports, - ): - if var not in all_var_set: - if isinstance(all_var, list): - all_var.append(var) - else: - cast(set[str], all_var).add(var) - - return cast("list[str] | set[str]", all_var) - - def find_missing( - self, - *, - all_var: bool | Collection[str] = True, - search_pathes: None | Collection[str] = None, - ignore_deprecated_import_errors: bool = False, - require_search_path_all_var: bool = True, - ) -> dict[ - str, - set[ - Literal[ - "not_in_all_var", - "missing_attr", - "missing_all_var", - "import", - "shadowed", - "search_path_extra", - "search_path_import", - ] - ], - ]: - """Debug method to check""" - - assert self.getter is not None - missing: dict[ - str, - set[ - Literal[ - "not_in_all_var", - "missing_attr", - "missing_all_var", - "import", - "shadowed", - "search_path_extra", - "search_path_import", - ] - ], - ] = {} - if all_var is True: - try: - all_var = self.getter("__all__", check_globals_dict=True) - except AttributeError: - missing.setdefault(self.globals_dict["__spec__"].name, set()).add("missing_all_var") - all_var = [] - key_set = set(chain(self.lazy_imports.keys(), self.deprecated_lazy_imports.keys())) - value_pathes_set: set[str] = set() - for name in key_set: - found_path: str = "" - if name in self.lazy_imports and isinstance(self.lazy_imports[name], str): - found_path = cast(str, self.lazy_imports[name]).replace(":", ".") - elif name in self.deprecated_lazy_imports and isinstance(self.deprecated_lazy_imports[name]["path"], str): - found_path = cast(str, self.deprecated_lazy_imports[name]["path"]).replace(":", ".") - if found_path: - value_pathes_set.add(absolutify_import(found_path, self.package)) - try: - obj = self.getter(name, no_warn_deprecated=True, check_globals_dict="fail") - # also add maybe rexported path - value_pathes_set.add(_obj_to_full_name(obj)) - except InGlobalsDict: - missing.setdefault(name, set()).add("shadowed") - except ImportError: - if not ignore_deprecated_import_errors or name not in self.deprecated_lazy_imports: - missing.setdefault(name, set()).add("import") - if all_var is not False: - for export_name in cast(Collection[str], all_var): - try: - obj = self.getter(export_name, no_warn_deprecated=True, check_globals_dict=True) - except AttributeError: - missing.setdefault(export_name, set()).add("missing_attr") - continue - if export_name not in key_set: - value_pathes_set.add(_obj_to_full_name(obj)) - - if search_pathes: - for search_path in search_pathes: - try: - mod = import_module(search_path, self.package) - except ImportError: - missing.setdefault(search_path, set()).add("search_path_import") - continue - try: - all_var_search = mod.__all__ - except AttributeError: - if require_search_path_all_var: - missing.setdefault(search_path, set()).add("missing_all_var") - - continue - for export_name in all_var_search: - export_path = absolutify_import(f"{search_path}.{export_name}", self.package) - try: - # for re-exports - obj = getattr(mod, export_name) - except AttributeError: - missing.setdefault(export_path, set()).add("missing_attr") - # still check check the export path - if export_path not in value_pathes_set: - missing.setdefault(export_path, set()).add("search_path_extra") - continue - if export_path not in value_pathes_set and _obj_to_full_name(obj) not in value_pathes_set: - missing.setdefault(export_path, set()).add("search_path_extra") - - if all_var is not False: - for name in key_set.difference(cast(Collection[str], all_var)): - missing.setdefault(name, set()).add("not_in_all_var") - - return missing - - @cached_property - def _loaded_settings(self) -> SETTINGS | None: - # only class and string pathes - if isclass(self._settings_definition): - return self._settings_definition() - assert isinstance(self._settings_definition, str), f"Not a settings object: {self._settings_definition}" - if not self._settings_definition: - return None - settings: SETTINGS | type[SETTINGS] = load(self._settings_definition, package=self.package) - if isclass(settings): - settings = settings() - return cast(SETTINGS, settings) - - @property - def settings(self) -> SETTINGS: - assert self._settings_var is not None, "Monkay not enabled for settings" - settings: SETTINGS | Callable[[], SETTINGS] | None = self._settings_var.get() - if settings is None: - # when settings_path is callable bypass the cache, for forwards - settings = ( - self._loaded_settings - if isinstance(self._settings_definition, str) or isclass(self._settings_definition) - else self._settings_definition - ) - if callable(settings): - settings = settings() - if settings is None: - raise RuntimeError("Settings are not set yet. Returned settings are None or settings_path is empty.") - return settings - - @settings.setter - def settings(self, value: str | Callable[[], SETTINGS] | SETTINGS | type[SETTINGS] | None) -> None: - assert self._settings_var is not None, "Monkay not enabled for settings" - if not value: - self._settings_definition = "" - return - if not isinstance(value, str) and not callable(value) and not isclass(value): - self._settings_definition = lambda: value - else: - self._settings_definition = value - del self.settings - - @settings.deleter - def settings(self) -> None: - # clear cache - self.__dict__.pop("_loaded_settings", None) - - @contextmanager - def with_settings(self, settings: SETTINGS | None) -> Generator[SETTINGS | None]: - assert self._settings_var is not None, "Monkay not enabled for settings" - # why None, for temporary using the real settings - token = self._settings_var.set(settings) - try: - yield settings - finally: - self._settings_var.reset(token) - - def add_lazy_import(self, name: str, value: str | Callable[[], Any], *, no_hooks: bool = False) -> None: - if not no_hooks and self.pre_add_lazy_import_hook is not None: - name, value = self.pre_add_lazy_import_hook(name, value, "lazy_import") - if name in self.lazy_imports: - raise KeyError(f'"{name}" is already a lazy import') - if name in self.deprecated_lazy_imports: - raise KeyError(f'"{name}" is already a deprecated lazy import') - self.lazy_imports[name] = value - if not no_hooks and self.post_add_lazy_import_hook is not None: - self.post_add_lazy_import_hook(name) - - def add_deprecated_lazy_import(self, name: str, value: DeprecatedImport, *, no_hooks: bool = False) -> None: - if not no_hooks and self.pre_add_lazy_import_hook is not None: - name, value = self.pre_add_lazy_import_hook(name, value, "deprecated_lazy_import") - if name in self.lazy_imports: - raise KeyError(f'"{name}" is already a lazy import') - if name in self.deprecated_lazy_imports: - raise KeyError(f'"{name}" is already a deprecated lazy import') - self.deprecated_lazy_imports[name] = value - if not no_hooks and self.post_add_lazy_import_hook is not None: - self.post_add_lazy_import_hook(name) - - def sorted_exports( - self, - all_var: Collection[str] | None = None, - *, - separate_by_category: bool = True, - sort_by: Literal["export_name", "path"] = "path", - ) -> list[SortedExportsEntry]: - if all_var is None: - all_var = self.globals_dict["__all__"] - sorted_exports: list[SortedExportsEntry] = [] - # ensure all entries are only returned once - for name in set(all_var): - if name in self.lazy_imports: - sorted_exports.append( - SortedExportsEntry( - "lazy_import", - name, - cast( - str, - self.lazy_imports[name] - if isinstance(self.lazy_imports[name], str) - else f"{self.globals_dict['__spec__'].name}.{name}", - ), - ) - ) - elif name in self.deprecated_lazy_imports: - sorted_exports.append( - SortedExportsEntry( - "deprecated_lazy_import", - name, - cast( - str, - self.deprecated_lazy_imports[name]["path"] - if isinstance(self.deprecated_lazy_imports[name]["path"], str) - else f"{self.globals_dict['__spec__'].name}.{name}", - ), - ) - ) - else: - sorted_exports.append( - SortedExportsEntry( - "other", - name, - f"{self.globals_dict['__spec__'].name}.{name}", - ) - ) - if separate_by_category: - - def key_fn(ordertuple: SortedExportsEntry) -> tuple: - return ordertuple.category, getattr(ordertuple, sort_by) - else: - - def key_fn(ordertuple: SortedExportsEntry) -> tuple: - return (getattr(ordertuple, sort_by),) - - sorted_exports.sort(key=key_fn) - return sorted_exports - - def module_getter( - self, - key: str, - *, - chained_getter: Callable[[str], Any] = _stub_previous_getattr, - no_warn_deprecated: bool = False, - check_globals_dict: bool | Literal["fail"] = False, - ) -> Any: - """ - Module Getter which handles lazy imports. - The injected version containing a potential found __getattr__ handler as chained_getter - is availabe as getter attribute. - """ - if check_globals_dict and key in self.globals_dict: - if check_globals_dict == "fail": - raise InGlobalsDict(f'"{key}" is defined as real variable.') - return self.globals_dict[key] - lazy_import = self.lazy_imports.get(key) - if lazy_import is None: - deprecated = self.deprecated_lazy_imports.get(key) - if deprecated is not None: - lazy_import = deprecated["path"] - if not no_warn_deprecated: - warn_strs = [f'Attribute: "{key}" is deprecated.'] - if deprecated.get("reason"): - # Note: no dot is added, this is the responsibility of the reason author. - warn_strs.append(f"Reason: {deprecated['reason']}") - if deprecated.get("new_attribute"): - warn_strs.append(f'Use "{deprecated["new_attribute"]}" instead.') - warnings.warn("\n".join(warn_strs), DeprecationWarning, stacklevel=2) - - if lazy_import is None: - return chained_getter(key) - if key not in self._cached_imports or key in self.uncached_imports: - if callable(lazy_import): - value: Any = lazy_import() - else: - value = load(lazy_import, package=self.package) - if key in self.uncached_imports: - return value - else: - self._cached_imports[key] = value - return self._cached_imports[key] - - def evaluate_settings( - self, - *, - on_conflict: Literal["error", "keep", "replace"] = "keep", - ) -> None: - preloads = None - if self.settings_preloads_name: - preloads = getattr(self.settings, self.settings_preloads_name) - if preloads: - for preload in preloads: - splitted = preload.rsplit(":", 1) - try: - module = import_module(splitted[0], self.package) - except ImportError: - module = None - if module is not None and len(splitted) == 2: - getattr(module, splitted[1])() - - if self.settings_extensions_name: - for extension in getattr(self.settings, self.settings_extensions_name): - self.add_extension(extension, use_overwrite=True, on_conflict=on_conflict) +def get_value_from_settings(settings: Any, name: str) -> Any: + try: + return getattr(settings, name) + except AttributeError: + return settings[name] diff --git a/monkay/core.py b/monkay/core.py new file mode 100644 index 0000000..e694487 --- /dev/null +++ b/monkay/core.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +from collections.abc import Callable, Iterable +from contextvars import ContextVar +from functools import partial +from importlib import import_module +from typing import ( + Any, + Generic, + Literal, +) + +from ._monkay_exports import MonkayExports +from ._monkay_instance import MonkayInstance +from ._monkay_settings import MonkaySettings +from .base import get_value_from_settings +from .types import ( + INSTANCE, + PRE_ADD_LAZY_IMPORT_HOOK, + SETTINGS, + DeprecatedImport, + ExtensionProtocol, +) + + +class Monkay( + MonkayInstance[INSTANCE, SETTINGS], + MonkaySettings[SETTINGS], + MonkayExports, + Generic[INSTANCE, SETTINGS], +): + def __init__( + self, + globals_dict: dict, + *, + with_instance: str | bool = False, + with_extensions: str | bool = False, + extension_order_key_fn: None + | Callable[[ExtensionProtocol[INSTANCE, SETTINGS]], Any] = None, + settings_path: str | Callable[[], SETTINGS] | SETTINGS | type[SETTINGS] | None = None, + preloads: Iterable[str] = (), + settings_preloads_name: str = "", + settings_extensions_name: str = "", + uncached_imports: Iterable[str] = (), + lazy_imports: dict[str, str | Callable[[], Any]] | None = None, + deprecated_lazy_imports: dict[str, DeprecatedImport] | None = None, + settings_ctx_name: str = "monkay_settings_ctx", + extensions_applied_ctx_name: str = "monkay_extensions_applied_ctx", + skip_all_update: bool = False, + evaluate_settings: bool = True, + pre_add_lazy_import_hook: None | PRE_ADD_LAZY_IMPORT_HOOK = None, + post_add_lazy_import_hook: None | Callable[[str], None] = None, + package: str | None = "", + ) -> None: + self.globals_dict = globals_dict + if with_instance is True: + with_instance = "monkay_instance_ctx" + with_instance = with_instance + if with_extensions is True: + with_extensions = "monkay_extensions_ctx" + with_extensions = with_extensions + if package == "" and globals_dict.get("__spec__"): + package = globals_dict["__spec__"].parent + self.package = package or None + + self._cached_imports: dict[str, Any] = {} + self.pre_add_lazy_import_hook = pre_add_lazy_import_hook + self.post_add_lazy_import_hook = post_add_lazy_import_hook + self.uncached_imports = set(uncached_imports) + self.lazy_imports = {} + self.deprecated_lazy_imports = {} + if lazy_imports: + for name, lazy_import in lazy_imports.items(): + self.add_lazy_import(name, lazy_import, no_hooks=True) + if deprecated_lazy_imports: + for name, deprecated_import in deprecated_lazy_imports.items(): + self.add_deprecated_lazy_import(name, deprecated_import, no_hooks=True) + if settings_path is not None: + self._settings_var = globals_dict[settings_ctx_name] = ContextVar( + settings_ctx_name, default=None + ) + self.settings = settings_path # type: ignore + self.settings_preloads_name = settings_preloads_name + self.settings_extensions_name = settings_extensions_name + + if with_instance: + self._instance_var = globals_dict[with_instance] = ContextVar( + with_instance, default=None + ) + if with_extensions: + self.extension_order_key_fn = extension_order_key_fn + self._extensions = {} + self._extensions_var = globals_dict[with_extensions] = ContextVar( + with_extensions, default=None + ) + self._extensions_applied_var = globals_dict[extensions_applied_ctx_name] = ContextVar( + extensions_applied_ctx_name, default=None + ) + if self.lazy_imports or self.deprecated_lazy_imports: + getter: Callable[..., Any] = self.module_getter + if "__getattr__" in globals_dict: + getter = partial(getter, chained_getter=globals_dict["__getattr__"]) + globals_dict["__getattr__"] = self.getter = getter + if not skip_all_update: + all_var = globals_dict.setdefault("__all__", []) + globals_dict["__all__"] = self.update_all_var(all_var) + for preload in preloads: + splitted = preload.rsplit(":", 1) + try: + module = import_module(splitted[0], self.package) + except ImportError: + module = None + if module is not None and len(splitted) == 2: + getattr(module, splitted[1])() + if evaluate_settings and self._settings_definition: + # disables overwrite + with self.with_settings(None): + self.evaluate_settings(on_conflict="error") + + def clear_caches(self, settings_cache: bool = True, import_cache: bool = True) -> None: + if settings_cache: + del self.settings + if import_cache: + self._cached_imports.clear() + + def evaluate_settings( + self, + *, + on_conflict: Literal["error", "keep", "replace"] = "keep", + ) -> None: + preloads = None + if self.settings_preloads_name: + preloads = get_value_from_settings(self.settings, self.settings_preloads_name) + if preloads: + for preload in preloads: + splitted = preload.rsplit(":", 1) + try: + module = import_module(splitted[0], self.package) + except ImportError: + module = None + if module is not None and len(splitted) == 2: + getattr(module, splitted[1])() + + if self.settings_extensions_name: + for extension in get_value_from_settings(self.settings, self.settings_extensions_name): + self.add_extension(extension, use_overwrite=True, on_conflict=on_conflict) diff --git a/monkay/types.py b/monkay/types.py new file mode 100644 index 0000000..94af312 --- /dev/null +++ b/monkay/types.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from collections.abc import Callable +from typing import ( + TYPE_CHECKING, + Any, + Literal, + NamedTuple, + Protocol, + TypedDict, + TypeVar, + overload, + runtime_checkable, +) + +if TYPE_CHECKING: + from .core import Monkay + + +INSTANCE = TypeVar("INSTANCE") +SETTINGS = TypeVar("SETTINGS") + + +@runtime_checkable +class ExtensionProtocol(Protocol[INSTANCE, SETTINGS]): + name: str + + def apply(self, monkay_instance: Monkay[INSTANCE, SETTINGS]) -> None: ... + + +class SortedExportsEntry(NamedTuple): + category: Literal["other", "lazy_import", "deprecated_lazy_import"] + export_name: str + path: str + + +class DeprecatedImport(TypedDict, total=False): + path: str | Callable[[], Any] + reason: str + new_attribute: str + + +DeprecatedImport.__required_keys__ = frozenset({"deprecated"}) + + +class PRE_ADD_LAZY_IMPORT_HOOK(Protocol): + @overload + @staticmethod + def __call__( + key: str, + value: str | Callable[[], Any], + type_: Literal["lazy_import"], + /, + ) -> tuple[str, str | Callable[[], Any]]: ... + + @overload + @staticmethod + def __call__( + key: str, + value: DeprecatedImport, + type_: Literal["deprecated_lazy_import"], + /, + ) -> tuple[str, DeprecatedImport]: ... + + @staticmethod + def __call__( + key: str, + value: str | Callable[[], Any] | DeprecatedImport, + type_: Literal["lazy_import", "deprecated_lazy_import"], + /, + ) -> tuple[str, str | Callable[[], Any] | DeprecatedImport]: ... diff --git a/pyproject.toml b/pyproject.toml index e19e478..24e577c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,23 +14,24 @@ authors = [ { name = "Tiago Silva", email = "tiago.arasilva@gmail.com" } ] classifiers = [ - "Development Status :: 4 - Beta", + "Operating System :: OS Independent", + "Topic :: Software Development :: Libraries :: Python Modules", + "Topic :: Software Development :: Libraries", + "Topic :: Software Development", + "Typing :: Typed", + "Development Status :: 5 - Production/Stable", "Programming Language :: Python", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy", "License :: OSI Approved :: BSD License" ] dependencies = [] -[project.optional-dependencies] -settings = [ - "pydantic-settings" -] - [project.urls] Documentation = "https://github.com/devkral/monkay#readme" Issues = "https://github.com/devkral/monkay/issues" @@ -48,6 +49,8 @@ dependencies = [ [tool.hatch.envs.docs] dependencies = [ "mkdocs", + "mkdocs-material>=9.4.4,<10.0.0", + "mdx-include>=1.4.2,<2.0.0" ] [tool.hatch.envs.docs.scripts] build = "mkdocs build" @@ -88,7 +91,7 @@ exclude_lines = [ "if TYPE_CHECKING:", ] -[ruff] +[tool.ruff] line-length = 99 fix = true @@ -96,6 +99,7 @@ fix = true select = ["E", "W", "F", "C", "B", "I", "UP", "SIM"] ignore = ["E501", "B008", "C901", "B026", "SIM115"] + [tool.ruff.lint.pycodestyle] max-line-length = 99 max-doc-length = 120 diff --git a/tests/targets/module_full.py b/tests/targets/module_full.py index 57a3d4b..ded560a 100644 --- a/tests/targets/module_full.py +++ b/tests/targets/module_full.py @@ -42,7 +42,9 @@ class FakeApp: def stringify_all_plain(separate_by_category: bool): return "[\n{}\n]".format( - "\n,".join(f'"{t[1]}"' for t in monkay.sorted_exports(separate_by_category=separate_by_category)) + "\n,".join( + f'"{t[1]}"' for t in monkay.sorted_exports(separate_by_category=separate_by_category) + ) ) diff --git a/tests/test_basic.py b/tests/test_basic.py index a1cf889..5dd7536 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -58,7 +58,10 @@ def test_attrs(): assert isinstance(mod.settings, BaseSettings) with pytest.warns(DeprecationWarning) as record: assert mod.deprecated() == "deprecated" - assert record[0].message.args[0] == 'Attribute: "deprecated" is deprecated.\nReason: old.\nUse "super_new" instead.' + assert ( + record[0].message.args[0] + == 'Attribute: "deprecated" is deprecated.\nReason: old.\nUse "super_new" instead.' + ) def test_load(): diff --git a/tests/test_missing.py b/tests/test_missing.py index 866e36f..c455b2a 100644 --- a/tests/test_missing.py +++ b/tests/test_missing.py @@ -15,7 +15,9 @@ def test_find_missing(): import tests.targets.module_full as mod # __all__ is autogenerated - assert not mod.monkay.find_missing(all_var=mod.__all__, search_pathes=["tests.targets.fn_module"]) + assert not mod.monkay.find_missing( + all_var=mod.__all__, search_pathes=["tests.targets.fn_module"] + ) # we can also use bools assert not mod.monkay.find_missing(all_var=True, search_pathes=["tests.targets.fn_module"]) assert mod.monkay.find_missing( @@ -24,7 +26,9 @@ def test_find_missing(): "tests.targets.not_existing": {"search_path_import"}, "tests.targets.module_preloaded1.not_included_export": {"search_path_extra"}, } - assert mod.monkay.find_missing(all_var={}, search_pathes=["tests.targets.module_full_preloaded1"]) == { + assert mod.monkay.find_missing( + all_var={}, search_pathes=["tests.targets.module_full_preloaded1"] + ) == { "bar": {"not_in_all_var"}, "bar2": { "not_in_all_var", diff --git a/tests/test_settings.py b/tests/test_settings.py index d77e626..691af2e 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -40,7 +40,9 @@ def test_settings_overwrite(): assert isinstance(settings_path, str) assert "tests.targets.module_settings_preloaded" not in sys.modules - new_settings = old_settings.model_copy(update={"preloads": ["tests.targets.module_settings_preloaded"]}) + new_settings = old_settings.model_copy( + update={"preloads": ["tests.targets.module_settings_preloaded"]} + ) with mod.monkay.with_settings(new_settings) as yielded: assert mod.monkay.settings is new_settings assert mod.monkay.settings is yielded @@ -59,12 +61,17 @@ def test_settings_overwrite(): assert mod.monkay.settings is not old_settings +@pytest.mark.parametrize("transform", [lambda x: x, lambda x: x.model_dump()]) @pytest.mark.parametrize("mode", ["error", "replace", "keep"]) -def test_settings_overwrite_evaluate_modes(mode): +def test_settings_overwrite_evaluate_modes(mode, transform): import tests.targets.module_full as mod with mod.monkay.with_settings( - mod.monkay.settings.model_copy(update={"preloads": ["tests.targets.module_settings_preloaded"]}) + transform( + mod.monkay.settings.model_copy( + update={"preloads": ["tests.targets.module_settings_preloaded"]} + ) + ) ) as new_settings: assert new_settings is not None if mode == "error": @@ -74,15 +81,18 @@ def test_settings_overwrite_evaluate_modes(mode): mod.monkay.evaluate_settings(on_conflict=mode) -def test_settings_overwrite_evaluate_no_conflict(): +@pytest.mark.parametrize("transform", [lambda x: x, lambda x: x.model_dump()]) +def test_settings_overwrite_evaluate_no_conflict(transform): import tests.targets.module_full as mod with mod.monkay.with_settings( - mod.monkay.settings.model_copy( - update={ - "preloads": ["tests.targets.module_settings_preloaded"], - "extensions": [], - } + transform( + mod.monkay.settings.model_copy( + update={ + "preloads": ["tests.targets.module_settings_preloaded"], + "extensions": [], + } + ) ) ) as new_settings: assert new_settings is not None