Env custom type casting (sanic-org#2330)

sanic/config.py

@@ -3,7 +3,7 @@
 from inspect import getmembers, isclass, isdatadescriptor
 from os import environ
 from pathlib import Path
-from typing import Any, Dict, Optional, Union
+from typing import Any, Callable, Dict, Optional, Sequence, Union
 from warnings import warn
 
 from sanic.errorpages import DEFAULT_FORMAT, check_error_format
@@ -43,6 +43,10 @@
     "WEBSOCKET_PING_TIMEOUT": 20,
 }
 
+# These values will be removed from the Config object in v22.6 and moved
+# to the application state
+DEPRECATED_CONFIG = ("SERVER_RUNNING", "RELOADER_PROCESS", "RELOADED_FILES")
+
 
 class DescriptorMeta(type):
     def __init__(cls, *_):
@@ -85,12 +89,19 @@ def __init__(
         load_env: Optional[Union[bool, str]] = True,
         env_prefix: Optional[str] = SANIC_PREFIX,
         keep_alive: Optional[bool] = None,
+        *,
+        converters: Optional[Sequence[Callable[[str], Any]]] = None,
     ):
         defaults = defaults or {}
         super().__init__({**DEFAULT_CONFIG, **defaults})
 
+        self._converters = [str, str_to_bool, float, int]
         self._LOGO = ""
 
+        if converters:
+            for converter in converters:
+                self.register_type(converter)
+
         if keep_alive is not None:
             self.KEEP_ALIVE = keep_alive
 
@@ -199,15 +210,31 @@ def load_environment_vars(self, prefix=SANIC_PREFIX):
         - ``float``
         - ``bool``
 
-        Anything else will be imported as a ``str``.
+        Anything else will be imported as a ``str``. If you would like to add
+        additional types to this list, you can use
+        :meth:`sanic.config.Config.register_type`. Just make sure that they
+        are registered before you instantiate your application.
+
+        .. code-block:: python
+
+            class Foo:
+                def __init__(self, name) -> None:
+                    self.name = name
+
+
+            config = Config(converters=[Foo])
+            app = Sanic(__name__, config=config)
+
+        `See user guide re: config
+        <https://sanicframework.org/guide/deployment/configuration.html>`__
         """
         for key, value in environ.items():
             if not key.startswith(prefix):
                 continue
 
             _, config_key = key.split(prefix, 1)
 
-            for converter in (int, float, str_to_bool, str):
+            for converter in reversed(self._converters):
                 try:
                     self[config_key] = converter(value)
                     break
@@ -282,3 +309,17 @@ class C:
         self.update(config)
 
     load = update_config
+
+    def register_type(self, converter: Callable[[str], Any]) -> None:
+        """
+        Allows for adding custom function to cast from a string value to any
+        other type. The function should raise ValueError if it is not the
+        correct type.
+        """
+        if converter in self._converters:
+            error_logger.warning(
+                f"Configuration value converter '{converter.__name__}' has "
+                "already been registered"
+            )
+            return
+        self._converters.append(converter)

tests/test_config.py

@@ -1,3 +1,5 @@
+import logging
+
 from contextlib import contextmanager
 from os import environ
 from pathlib import Path
@@ -32,6 +34,11 @@ def another_not_for_config(self):
         return self.not_for_config
 
 
+class UltimateAnswer:
+    def __init__(self, answer):
+        self.answer = int(answer)
+
+
 def test_load_from_object(app):
     app.config.load(ConfigTest)
     assert "CONFIG_VALUE" in app.config
@@ -137,6 +144,32 @@ def test_env_prefix_string_value():
     del environ["MYAPP_TEST_TOKEN"]
 
 
+def test_env_w_custom_converter():
+    environ["SANIC_TEST_ANSWER"] = "42"
+
+    config = Config(converters=[UltimateAnswer])
+    app = Sanic(name=__name__, config=config)
+    assert isinstance(app.config.TEST_ANSWER, UltimateAnswer)
+    assert app.config.TEST_ANSWER.answer == 42
+    del environ["SANIC_TEST_ANSWER"]
+
+
+def test_add_converter_multiple_times(caplog):
+    def converter():
+        ...
+
+    message = (
+        "Configuration value converter 'converter' has already been registered"
+    )
+    config = Config()
+    config.register_type(converter)
+    with caplog.at_level(logging.WARNING):
+        config.register_type(converter)
+
+    assert ("sanic.error", logging.WARNING, message) in caplog.record_tuples
+    assert len(config._converters) == 5
+
+
 def test_load_from_file(app):
     config = dedent(
         """