Repository: dynaconf/dynaconf
Pull request #1249 (2025-02-12 17:03:14)

[port from 3.2.8] feat: Add support for decorated hooks on settings files (#1246)
* feat: Add support for decorated hooks on settings files

Closes #1245
Port of #1246 

github-actions[bot]: ## Test Results (Python 3.12)
569 tests  +1   568 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") +1   56s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "duration of all tests") +4s
    1 suites ±0       1 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") ±0 
    1 files   ±0       0 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") ±0 

Results for commit b3de9ef7. ± Comparison against base commit 71ea887a.

[test-results]:data:application/gzip;base64,H4sIAD7VrGcC/1XMSw6DIBSF4a0Yxh0gFJVupuFxSW6q0vAYNd17KVqxw+8/yXkRhzNEcuv6S0dixnTA5qAS+rVQDMVlSbFC/nSP2ZiappYe+NwftuAUziXQI0AIPuwl5LV9fvF/uZX2WH06rD7/Gb8smAqI5hYkuJELCUwzB1ZyyhXVyqlBXrnUTPR0nMj7A5hXjgoEAQAA


Patch:
diff --git a/docs/advanced.md b/docs/advanced.md
index 8192f8e..edc69b0 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -81,6 +81,46 @@ settings = Dynaconf(post_hooks=hook_function)
 You can also set the merging individually for each settings variable as seen on
 [merging](merging.md) documentation.
 
+
+#### Decorator approach
+
+**new in 3.2.8**
+
+When the settings file is a Python file, you can also use a decorator to define hooks.
+
+```python
+from dynaconf import Dynaconf
+
+settings = Dynaconf(settings_file="settings.py")
+```
+
+```python title="settings.py"
+from dynaconf import post_hook
+
+VARIABLE = "value"
+# all regular settings here
+...
+
+@post_hook
+def set_debugging_database_url(settings):
+    data = {}
+    if settings.DEBUG:
+        data["DATABASE_URL"] = "sqlite://"
+    return data
+```
+
+Dynaconf will collect the decorated functions and execute them after the settings file is loaded.
+
+The hook decorator can only be applied to functions that are defined withing the same settings file, in the
+need to use external functions, wrap it in a local decorated function.
+
+When using `load_file` to load a python file, hooks will also be collected and immediately executed after the file is loaded, unless the `run_hooks` argument is set to `False`, this is useful when calling `load_file` multiple times and have the hooks to be all executed together at a certain point later.
+
+`execute_instance_hooks` helper method is also available to trigger execution of collected hooks, but usually not
+needed because a common pattern is to call the latest `load_file` with `run_hooks=True` to trigger all hooks at once.
+
+Notice that load_file will execute all collected hooks that have not been called yet, it does not execute hooks that have already been called (unless the hook itself is marked as `_called=False`). This is useful when you want to reload a file and re-execute hooks.
+
 ## Inspecting History
 
 > **NEW** in version 3.2.0
diff --git a/dynaconf/__init__.py b/dynaconf/__init__.py
index d59d405..ed2e47b 100644
--- a/dynaconf/__init__.py
+++ b/dynaconf/__init__.py
@@ -4,6 +4,7 @@ from dynaconf.base import LazySettings
 from dynaconf.constants import DEFAULT_SETTINGS_FILES
 from dynaconf.contrib import DjangoDynaconf
 from dynaconf.contrib import FlaskDynaconf
+from dynaconf.hooking import post_hook
 from dynaconf.utils.inspect import get_history
 from dynaconf.utils.inspect import inspect_settings
 from dynaconf.utils.parse_conf import add_converter
@@ -39,4 +40,5 @@ __all__ = [
     "get_history",
     "DynaconfFormatError",
     "DynaconfParseError",
+    "post_hook",
 ]
diff --git a/dynaconf/base.py b/dynaconf/base.py
index fa8260a..561ac16 100644
--- a/dynaconf/base.py
+++ b/dynaconf/base.py
@@ -1238,6 +1238,10 @@ class Settings:
         """Clean end Execute all loaders"""
         self.clean()
         self._loaded_hooks.clear()
+        for hook in self._post_hooks:
+            with suppress(AttributeError, TypeError):
+                hook._called = False
+
         self.execute_loaders(env, silent)
 
     def execute_loaders(
@@ -1294,7 +1298,13 @@ class Settings:
                 last_loader.load(self, env, silent, key)
 
     def load_file(
-        self, path=None, env=None, silent=True, key=None, validate=empty
+        self,
+        path=None,
+        env=None,
+        silent=True,
+        key=None,
+        validate=empty,
+        run_hooks=True,
     ):
         """Programmatically load files from ``path``.
 
@@ -1303,81 +1313,95 @@ class Settings:
         - Directory of the last loaded file
         - CWD
 
-        :param path: A single filename or a file list
+        :param path: A single filename, a glob or a file list
         :param env: Which env to load from file (default current_env)
         :param silent: Should raise errors?
         :param key: Load a single key?
         :param validate: Should trigger validation?
+        :param run_hooks: Should run collected hooks?
         """
+        files = ensure_a_list(path)
+        if not files:  # a glob pattern may return empty
+            return
+
         if validate is empty:
             validate = self.get("VALIDATE_ON_UPDATE_FOR_DYNACONF")
 
         env = (env or self.current_env).upper()
-        files = ensure_a_list(path)
-        if files:
-            # Using inspect take the filename and line number of the caller
-            # to be used in the source_metadata
-            frame = inspect.currentframe()
-            caller = inspect.getouterframes(frame)[1]
-            already_loaded = set()
-            for _filename in files:
+
+        # Using inspect take the filename and line number of the caller
+        # to be used in the source_metadata
+        frame = inspect.currentframe()
+        caller = inspect.getouterframes(frame)[1]
+
+        already_loaded = set()
+        for _filename in files:
+            # load_file() will handle validation later
+            with suppress(ValidationError):
+                source_metadata = SourceMetadata(
+                    loader=f"load_file@{caller.filename}:{caller.lineno}",
+                    identifier=_filename,
+                    env=env,
+                )
+                if py_loader.try_to_load_from_py_module_name(
+                    obj=self,
+                    name=_filename,
+                    silent=True,
+                    identifier=source_metadata,
+                ):
+                    # if it was possible to load from module name
+                    # continue the loop.
+                    continue
+
+            root_dir = str(self._root_path or os.getcwd())
+
+            # Issue #494
+            if (
+                isinstance(_filename, Path)
+                and str(_filename.parent) in root_dir
+            ):  # pragma: no cover
+                filepath = str(_filename)
+            else:
+                filepath = os.path.join(root_dir, str(_filename))
+
+            paths = [p for p in sorted(glob(filepath)) if ".local." not in p]
+            local_paths = [p for p in sorted(glob(filepath)) if ".local." in p]
+
+            # Handle possible *.globs sorted alphanumeric
+            for path in paths + local_paths:
+                if path in already_loaded:  # pragma: no cover
+                    continue
+
                 # load_file() will handle validation later
                 with suppress(ValidationError):
                     source_metadata = SourceMetadata(
                         loader=f"load_file@{caller.filename}:{caller.lineno}",
-                        identifier=_filename,
+                        identifier=path,
                         env=env,
                     )
-                    if py_loader.try_to_load_from_py_module_name(
+                    settings_loader(
                         obj=self,
-                        name=_filename,
-                        silent=True,
+                        env=env,
+                        silent=silent,
+                        key=key,
+                        filename=path,
+                        validate=validate,
                         identifier=source_metadata,
-                    ):
-                        # if it was possible to load from module name
-                        # continue the loop.
-                        continue
-
-                root_dir = str(self._root_path or os.getcwd())
-
-                # Issue #494
-                if (
-                    isinstance(_filename, Path)
-                    and str(_filename.parent) in root_dir
-                ):  # pragma: no cover
-                    filepath = str(_filename)
-                else:
-                    filepath = os.path.join(root_dir, str(_filename))
-
-                paths = [
-                    p for p in sorted(glob(filepath)) if ".local." not in p
-                ]
-                local_paths = [
-                    p for p in sorted(glob(filepath)) if ".local." in p
-                ]
-
-                # Handle possible *.globs sorted alphanumeric
-                for path in paths + local_paths:
-                    if path in already_loaded:  # pragma: no cover
-                        continue
-
-                    # load_file() will handle validation later
-                    with suppress(ValidationError):
-                        source_metadata = SourceMetadata(
-                            loader=f"load_file@{caller.filename}:{caller.lineno}",
-                            identifier=path,
-                            env=env,
-                        )
-                        settings_loader(
-                            obj=self,
-                            env=env,
-                            silent=silent,
-                            key=key,
-                            filename=path,
-                            validate=validate,
-                            identifier=source_metadata,
-                        )
-                        already_loaded.add(path)
+                    )
+                    already_loaded.add(path)
+
+        if run_hooks:
+            # this will call any collected hook that was not called yet
+            execute_instance_hooks(
+                self,
+                "post",
+                [
+                    _hook
+                    for _hook in self._post_hooks
+                    if getattr(_hook, "_dynaconf_hook", False) is True
+                    and not getattr(_hook, "_called", False)
+                ],
+            )
 
         # handle param `validate`
         if validate is True:
diff --git a/dynaconf/hooking.py b/dynaconf/hooking.py
index f2d7788..3bcd87c 100644
--- a/dynaconf/hooking.py
+++ b/dynaconf/hooking.py
@@ -18,6 +18,7 @@ __all__ = [
     "MethodValue",
     "Action",
     "HookableSettings",
+    "post_hook",
 ]
 
 
@@ -337,3 +338,34 @@ class TempSettingsHolder:
             super().__setattr__(attr, value)
         else:
             setattr(self._settings, attr, value)
+
+
+def post_hook(function: Callable) -> Callable:
+    """This decorator marks a function as a post hook.
+    This works by adding the _dynaconf_hook attribute to the function,
+    then the python loader, when reading the module, will look for
+    this attribute and register the function as a post_hook for the settings.
+
+    e.g: On a settings file with .py extension:
+
+    from dynaconf import post_hook
+    @post_hook
+    def set_log_handlers(settings) -> dict:
+        data = {}  # data to be merged into settings
+
+        # conditionals
+        if (logging := settings.get('LOGGING')) is not None:
+            # do something with logging
+            # add it back to data
+            data['LOGGING'] = logging
+        return data
+    """
+    try:
+        function._dynaconf_hook = True  # type: ignore
+        function._called = False  # type: ignore
+        function._dynaconf_hook_source = function.__module__  # type: ignore
+    except (AttributeError, TypeError):
+        raise TypeError(
+            "post_hook decorator must be applied to a function or method."
+        )
+    return function
diff --git a/dynaconf/loaders/__init__.py b/dynaconf/loaders/__init__.py
index 6ef953a..3065819 100644
--- a/dynaconf/loaders/__init__.py
+++ b/dynaconf/loaders/__init__.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import importlib
 import os
+from contextlib import suppress
 from typing import Callable
 from typing import TYPE_CHECKING
 
@@ -183,12 +184,23 @@ def _run_hook_function(
     It execute @hook_func, update the results into settings @obj and
     add it to _loaded_hook registry ([@hook_source][@hook_type])
     """
+    # if the function has a _dynaconf_hook_source attribute set
+    # hook_source to it
+    hook_source = getattr(hook_func, "_dynaconf_hook_source", hook_source)
+
     # optional settings argument
     try:
         hook_dict = hook_func(obj.dynaconf.clone())
     except TypeError:
         hook_dict = hook_func()
 
+    # mark as called so executors such as `load_file` can avoid calling it again
+    with suppress(AttributeError, TypeError):
+        # callable may not be writable, the caveat is that it will be called again in case of reload
+        # however, this must not be a problem since the function should be idempotent
+        # and documentation warns about this behavior.
+        hook_func._called = True
+
     # update obj settings
     if hook_dict:
         identifier = f"{hook_func.__name__}@{hook_source}"
diff --git a/dynaconf/loaders/py_loader.py b/dynaconf/loaders/py_loader.py
index 289588e..394d614 100644
--- a/dynaconf/loaders/py_loader.py
+++ b/dynaconf/loaders/py_loader.py
@@ -57,13 +57,13 @@ def load_from_python_object(
         file_merge = getattr(mod, "DYNACONF_MERGE", empty)
 
     for setting in dir(mod):
+        setting_value = getattr(mod, setting)
         # A setting var in a Python file should start with upper case
         # valid: A_value=1, ABC_value=3 A_BBB__default=1
         # invalid: a_value=1, MyValue=3
         # This is to avoid loading functions, classes and built-ins
         if setting.split("__")[0].isupper():
             if key is None or key == setting:
-                setting_value = getattr(mod, setting)
                 obj.set(
                     setting,
                     setting_value,
@@ -71,6 +71,14 @@ def load_from_python_object(
                     merge=file_merge,
                     validate=validate,
                 )
+        # if setting is a post_hook function it will be a callable with
+        # the _dynaconf_hook attribute set to True
+        # then we want to add it to the post_hooks list on the obj.
+        elif callable(setting_value) and getattr(
+            setting_value, "_dynaconf_hook", False
+        ):
+            if setting_value not in obj._post_hooks:
+                obj._post_hooks.append(setting_value)
 
     obj._loaded_py_modules.append(mod.__name__)
     obj._loaded_files.append(mod.__file__)


Test patch:
diff --git a/tests/test_py_loader.py b/tests/test_py_loader.py
index f6e8867..82a6b3b 100644
--- a/tests/test_py_loader.py
+++ b/tests/test_py_loader.py
@@ -96,6 +96,93 @@ def test_py_loader_from_file_dunder(clean_env, tmpdir):
     assert settings.DATABASES.default.ENGINE == "other.module"
 
 
+def test_decorated_hooks(clean_env, tmpdir):
+    # Arrange
+    settings_path = tmpdir.join("settings.py")
+    settings_extra = tmpdir.join("extra.py")
+
+    to_write = {
+        str(settings_path): [
+            "INSTALLED_APPS = ['admin']",
+            "COLORS = ['red', 'green']",
+            "DATABASES = {'default': {'NAME': 'db'}}",
+            "BANDS = ['Rush', 'Yes']",
+            "from dynaconf import post_hook",
+            "@post_hook",
+            "def post_hook(settings):",
+            "    return {",
+            "        'INSTALLED_APPS': ['admin', 'plugin'],",
+            "        'COLORS': '@merge blue',",
+            "        'DATABASES__default': '@merge PORT=5151',",
+            "        'DATABASES__default__VERSION': 42,",
+            "        'DATABASES__default__FORCED_INT': '@int 12',",
+            "        'BANDS': ['Anathema', 'dynaconf_merge'],",
+            "    }",
+        ],
+        str(settings_extra): [
+            "NAME = 'Bruce'",
+            "BASE_PATH = '/etc/frutaria'",
+            "from dynaconf import post_hook",
+            "@post_hook",
+            "def post_hook(settings):",
+            "    return {",
+            "        'INSTALLED_APPS': ['admin', 'extra'],",
+            "        'COLORS': '@merge yellow',",
+            "        'DATABASES__default': '@merge PORT=5152',",
+            "        'DATABASES__default__VERSION': 43,",
+            "        'DATABASES__default__FORCED_INT': '@int 13',",
+            "        'BANDS': ['Opeth', 'dynaconf_merge'],",
+            # Not sure @format should be supported on hooks, as settings is available on scope.
+            # maybe useful to set as a lazy value to be resolved after all hooks are executed.
+            "        'FULL_NAME': '@format {this.NAME} Rock',",
+            "        'FULL_PATH': f'{settings.BASE_PATH}/batata/',",
+            "    }",
+        ],
+    }
+
+    for path, lines in to_write.items():
+        with open(
+            str(path), "w", encoding=default_settings.ENCODING_FOR_DYNACONF
+        ) as f:
+            for line in lines:
+                f.write(line)
+                f.write("\n")
+
+    # Act
+    settings = LazySettings(settings_file="settings.py")
+
+    # Assert
+    # settings has _post_hooks attribute with one hook registered
+    assert settings._post_hooks
+    assert len(settings._post_hooks) == 1
+
+    # assert values set by the first hook only
+    assert settings.INSTALLED_APPS == ["admin", "plugin"]
+    assert settings.COLORS == ["red", "green", "blue"]
+    assert settings.DATABASES.default.NAME == "db"
+    assert settings.DATABASES.default.PORT == 5151
+    assert settings.DATABASES.default.VERSION == 42
+    assert settings.DATABASES.default.FORCED_INT == 12
+    assert settings.BANDS == ["Rush", "Yes", "Anathema"]
+
+    # Call load_file to load the second file
+    settings.load_file(str(settings_extra))
+
+    # assert _post_hooks attribute has two hooks registered
+    assert len(settings._post_hooks) == 2
+
+    # assert values set by the second hook only
+    assert settings.INSTALLED_APPS == ["admin", "extra"]
+    assert settings.COLORS == ["red", "green", "blue", "yellow"]
+    assert settings.DATABASES.default.NAME == "db"
+    assert settings.DATABASES.default.PORT == 5152
+    assert settings.DATABASES.default.VERSION == 43
+    assert settings.DATABASES.default.FORCED_INT == 13
+    assert settings.BANDS == ["Rush", "Yes", "Anathema", "Opeth"]
+    assert settings.FULL_NAME == "Bruce Rock"
+    assert settings.FULL_PATH == "/etc/frutaria/batata/"
+
+
 def test_post_load_hooks(clean_env, tmpdir):
     """Test post load hooks works