Env custom type casting

sanic/config.py

@@ -1,13 +1,23 @@
 from __future__ import annotations
 
+from collections import deque
 from inspect import isclass
 from os import environ
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Optional, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    Optional,
+    Sequence,
+    Union,
+)
 from warnings import warn
 
 from sanic.errorpages import check_error_format
 from sanic.http import Http
+from sanic.log import error_logger
 from sanic.utils import load_module_from_file_location, str_to_bool
 
 
@@ -45,8 +55,24 @@
     "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 CastRegistry(deque):
+    def add(self, cast: Callable[[str], Any]) -> None:
+        if cast in self:
+            error_logger.warning(
+                f"Type cast '{cast.__name__}' has already been registered"
+            )
+            return None
+        self.appendleft(cast)
+
 
 class Config(dict):
+    __registry__ = CastRegistry((int, float, str_to_bool, str))
+
     ACCESS_LOG: bool
     AUTO_RELOAD: bool
     EVENT_AUTOREGISTER: bool
@@ -81,13 +107,17 @@ def __init__(
         keep_alive: Optional[bool] = None,
         *,
         app: Optional[Sanic] = None,
+        converters: Optional[Sequence[Callable[[str], Any]]] = None,
     ):
         defaults = defaults or {}
         super().__init__({**DEFAULT_CONFIG, **defaults})
 
         self._app = app
         self._LOGO = ""
 
+        if converters:
+            self.register_type(*converters)
+
         if keep_alive is not None:
             self.KEEP_ALIVE = keep_alive
 
@@ -184,15 +214,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 self.__registry__:
                 try:
                     self[config_key] = converter(value)
                     break
@@ -267,3 +313,12 @@ class C:
         self.update(config)
 
     load = update_config
+
+    def register_type(self, *cast: 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.
+        """
+        for item in cast:
+            self.__registry__.add(item)

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"]
+    config.__registry__.remove(UltimateAnswer)
+
+
+def test_add_converter_multiple_times(caplog):
+    def converter():
+        ...
+
+    message = "Type cast '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.__registry__) == 5
+    config.__registry__.remove(converter)
+
+
 def test_load_from_file(app):
     config = dedent(
         """