Add a proper interpolation option

CHANGELOG.rst

@@ -5,6 +5,13 @@ All notable changes to this project will be documented in this file.
 The format is inspired by `Keep a Changelog <https://keepachangelog.com/en/1.0.0/>`_
 and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0.html>`_.
 
+`v0.12.0`_ - 11-November-2023
+-----------------------------
+Added
++++++
+- Add option to interpolate variables as an alternative for simple variable
+  proxies.
+
 `v0.11.3`_ - 0-Undefined-2023
 -----------------------------
 Changed
@@ -399,6 +406,7 @@ Added
 - Initial release.
 
 
+.. _v0.12.0: https://github.com/joke2k/django-environ/compare/v0.11.3...v0.12.0
 .. _v0.11.3: https://github.com/joke2k/django-environ/compare/v0.11.2...v0.11.3
 .. _v0.11.2: https://github.com/joke2k/django-environ/compare/v0.11.1...v0.11.2
 .. _v0.11.1: https://github.com/joke2k/django-environ/compare/v0.11.0...v0.11.1

docs/tips.rst

@@ -290,20 +290,53 @@ The following example demonstrates the above:
 Proxy value
 ===========
 
-Values that being with a ``$`` may be interpolated. Pass ``interpolate=True`` to
-``environ.Env()`` to enable this feature:
+Values that begin with a ``$`` may be interpolated:
 
 .. code-block:: python
 
    import environ
 
-   env = environ.Env(interpolate=True)
+   env = environ.Env()
 
    # BAR=FOO
    # PROXY=$BAR
    >>> print(env.str('PROXY'))
    FOO
 
+Interpolation
+=============
+
+Variable interpolation can be enabled with ``interpolation``. It allows for more
+complex variable interpolation than proxies:
+
+.. code-block:: python
+
+    import environ
+
+    env = environ.Env()
+    env.interpolation = True
+
+    # FOO=abc
+    # BAR=def
+    # INTERPOLATED=prefix:$FOO@${BAR}Suffix
+    env.str('INTERPOLATED')  # prefix:abc@defSuffix
+
+Variables with escaped dollar sign (``\$``) are not interpolated.
+
+When variable does not exist, an exception will be raised. These exceptions can
+be disabled with ``raise_on_missing``:
+
+.. code-block:: python
+
+    import environ
+
+    env = environ.Env()
+    env.interpolation = True
+    env.raise_on_missing = False
+
+    # FOO=abc
+    # INTERPOLATED=prefix:$FOO@${BAR}Suffix
+    env.str('INTERPOLATED')  # prefix:abc@${BAR}Suffix
 
 Escape Proxy
 ============

environ/__init__.py

@@ -21,7 +21,7 @@
 __copyright__ = 'Copyright (C) 2013-2023 Daniele Faraglia'
 """The copyright notice of the package."""
 
-__version__ = '0.11.3'
+__version__ = '0.12.0'
 """The version of the package."""
 
 __license__ = 'MIT'

environ/environ.py

@@ -12,6 +12,7 @@
 """
 
 import ast
+import functools
 import itertools
 import logging
 import os
@@ -37,6 +38,8 @@
 )
 from .fileaware_mapping import FileAwareMapping
 
+INTERPOLATION_CACHE_SIZE = 128
+
 Openable = (str, os.PathLike)
 logger = logging.getLogger(__name__)
 
@@ -189,161 +192,174 @@ class Env:
                             for s in ('', 's')]
     CLOUDSQL = 'cloudsql'
 
+    _VAR_PATTERN = re.compile(r"""
+        (?P<to_replace>
+            (?P<escape>\\)?
+            \$(?P<name>[A-Z_][0-9A-Z_]*|\{[A-Z_][0-9A-Z]*})
+        )
+    """, re.IGNORECASE | re.VERBOSE)
+
     def __init__(self, **scheme):
         self.smart_cast = True
         self.escape_proxy = False
+        self.interpolation = False
+        self.raise_on_missing = True
         self.prefix = ""
         self.scheme = scheme
 
-    def __call__(self, var, cast=None, default=NOTSET, parse_default=False):
+    def __call__(self, var, cast=None, default=NOTSET, parse_default=False, interpolate=None):
         return self.get_value(
             var,
             cast=cast,
             default=default,
-            parse_default=parse_default
+            parse_default=parse_default,
+            interpolate=interpolate
         )
 
     def __contains__(self, var):
         return var in self.ENVIRON
 
-    def str(self, var, default=NOTSET, multiline=False):
+    def str(self, var, default=NOTSET, multiline=False, interpolate=None):
         """
         :rtype: str
         """
-        value = self.get_value(var, cast=str, default=default)
+        value = self.get_value(var, cast=str, default=default, interpolate=interpolate)
         if multiline:
             return re.sub(r'(\\r)?\\n', r'\n', value)
         return value
 
-    def bytes(self, var, default=NOTSET, encoding='utf8'):
+    def bytes(self, var, default=NOTSET, encoding='utf8', interpolate=None):
         """
         :rtype: bytes
         """
-        value = self.get_value(var, cast=str, default=default)
+        value = self.get_value(var, cast=str, default=default, interpolate=interpolate)
         if hasattr(value, 'encode'):
             return value.encode(encoding)
         return value
 
-    def bool(self, var, default=NOTSET):
+    def bool(self, var, default=NOTSET, interpolate=None):
         """
         :rtype: bool
         """
-        return self.get_value(var, cast=bool, default=default)
+        return self.get_value(var, cast=bool, default=default, interpolate=interpolate)
 
-    def int(self, var, default=NOTSET):
+    def int(self, var, default=NOTSET, interpolate=None):
         """
         :rtype: int
         """
-        return self.get_value(var, cast=int, default=default)
+        return self.get_value(var, cast=int, default=default, interpolate=interpolate)
 
-    def float(self, var, default=NOTSET):
+    def float(self, var, default=NOTSET, interpolate=None):
         """
         :rtype: float
         """
-        return self.get_value(var, cast=float, default=default)
+        return self.get_value(var, cast=float, default=default, interpolate=interpolate)
 
-    def json(self, var, default=NOTSET):
+    def json(self, var, default=NOTSET, interpolate=None):
         """
         :returns: Json parsed
         """
-        return self.get_value(var, cast=json.loads, default=default)
+        return self.get_value(var, cast=json.loads, default=default, interpolate=interpolate)
 
-    def list(self, var, cast=None, default=NOTSET):
+    def list(self, var, cast=None, default=NOTSET, interpolate=None):
         """
         :rtype: list
         """
         return self.get_value(
             var,
             cast=list if not cast else [cast],
-            default=default
+            default=default,
+            interpolate=interpolate
         )
 
-    def tuple(self, var, cast=None, default=NOTSET):
+    def tuple(self, var, cast=None, default=NOTSET, interpolate=None):
         """
         :rtype: tuple
         """
         return self.get_value(
             var,
             cast=tuple if not cast else (cast,),
-            default=default
+            default=default,
+            interpolate=interpolate
         )
 
-    def dict(self, var, cast=dict, default=NOTSET):
+    def dict(self, var, cast=dict, default=NOTSET, interpolate=None):
         """
         :rtype: dict
         """
-        return self.get_value(var, cast=cast, default=default)
+        return self.get_value(var, cast=cast, default=default, interpolate=interpolate)
 
-    def url(self, var, default=NOTSET):
+    def url(self, var, default=NOTSET, interpolate=None):
         """
         :rtype: urllib.parse.ParseResult
         """
         return self.get_value(
             var,
             cast=urlparse,
             default=default,
-            parse_default=True
+            parse_default=True,
+            interpolate=interpolate
         )
 
-    def db_url(self, var=DEFAULT_DATABASE_ENV, default=NOTSET, engine=None):
+    def db_url(self, var=DEFAULT_DATABASE_ENV, default=NOTSET, engine=None, interpolate=None):
         """Returns a config dictionary, defaulting to DATABASE_URL.
 
         The db method is an alias for db_url.
 
         :rtype: dict
         """
         return self.db_url_config(
-            self.get_value(var, default=default),
+            self.get_value(var, default=default, interpolate=interpolate),
             engine=engine
         )
 
     db = db_url
 
-    def cache_url(self, var=DEFAULT_CACHE_ENV, default=NOTSET, backend=None):
+    def cache_url(self, var=DEFAULT_CACHE_ENV, default=NOTSET, backend=None, interpolate=None):
         """Returns a config dictionary, defaulting to CACHE_URL.
 
         The cache method is an alias for cache_url.
 
         :rtype: dict
         """
         return self.cache_url_config(
-            self.url(var, default=default),
+            self.url(var, default=default, interpolate=interpolate),
             backend=backend
         )
 
     cache = cache_url
 
-    def email_url(self, var=DEFAULT_EMAIL_ENV, default=NOTSET, backend=None):
+    def email_url(self, var=DEFAULT_EMAIL_ENV, default=NOTSET, backend=None, interpolate=None):
         """Returns a config dictionary, defaulting to EMAIL_URL.
 
         The email method is an alias for email_url.
 
         :rtype: dict
         """
         return self.email_url_config(
-            self.url(var, default=default),
+            self.url(var, default=default, interpolate=interpolate),
             backend=backend
         )
 
     email = email_url
 
-    def search_url(self, var=DEFAULT_SEARCH_ENV, default=NOTSET, engine=None):
+    def search_url(self, var=DEFAULT_SEARCH_ENV, default=NOTSET, engine=None, interpolate=None):
         """Returns a config dictionary, defaulting to SEARCH_URL.
 
         :rtype: dict
         """
         return self.search_url_config(
-            self.url(var, default=default),
+            self.url(var, default=default, interpolate=interpolate),
             engine=engine
         )
 
-    def path(self, var, default=NOTSET, **kwargs):
+    def path(self, var, default=NOTSET, interpolate=None, **kwargs):
         """
         :rtype: Path
         """
-        return Path(self.get_value(var, default=default), **kwargs)
+        return Path(self.get_value(var, default=default, interpolate=interpolate), **kwargs)
 
-    def get_value(self, var, cast=None, default=NOTSET, parse_default=False):
+    def get_value(self, var, cast=None, default=NOTSET, parse_default=False, interpolate=None):
         """Return value for given environment variable.
 
         :param str var:
@@ -354,6 +370,8 @@ def get_value(self, var, cast=None, default=NOTSET, parse_default=False):
              If var not present in environ, return this instead.
         :param bool parse_default:
             Force to parse default.
+        :param bool, optional interpolate:
+            Enable or disable interpolation. Uses class-defined ``interpolation`` as default.
         :returns: Value from environment or default (if set).
         :rtype: typing.IO[typing.Any]
         """
@@ -396,9 +414,17 @@ def get_value(self, var, cast=None, default=NOTSET, parse_default=False):
         # Resolve any proxied values
         prefix = b'$' if isinstance(value, bytes) else '$'
         escape = rb'\$' if isinstance(value, bytes) else r'\$'
-        if hasattr(value, 'startswith') and value.startswith(prefix):
-            value = value.lstrip(prefix)
-            value = self.get_value(value, cast=cast, default=default)
+
+        if interpolate is None:
+            interpolate = self.interpolation
+        if interpolate:
+            # Interpolate variables
+            value = self.interpolate(value, var_name)
+        else:
+            # Resolve any proxied values
+            if hasattr(value, 'startswith') and value.startswith(prefix):
+                value = value.lstrip(prefix)
+                value = self.get_value(value, cast=cast, default=default)
 
         if self.escape_proxy and hasattr(value, 'replace'):
             value = value.replace(escape, prefix)
@@ -416,6 +442,34 @@ def get_value(self, var, cast=None, default=NOTSET, parse_default=False):
 
         return value
 
+    @functools.lru_cache(maxsize=INTERPOLATION_CACHE_SIZE, typed=True)
+    def interpolate(self, value, var_name=None):
+        """Interpolate variables in provided value
+
+        :param IO value:
+            String or bytes object to interpolate.
+        :param str, optional var_name:
+            Name of the variable whose value will be interpolated.
+
+        :returns: Interpolated value.
+        """
+        str_value = value.decode('utf-8') if isinstance(value, bytes) else value
+        for match in self._VAR_PATTERN.finditer(str_value):
+            if match.group('escape'):
+                continue  # skip escaped variables
+            to_replace = match.group('to_replace')
+            name = match.group('name').lstrip('{').rstrip('}')
+            if var_name and name == var_name:
+                error_msg = f'Variable {name} references itself'
+                raise ImproperlyConfigured(error_msg)
+            try:
+                str_value = str_value.replace(to_replace, self.get_value(name))
+            except ImproperlyConfigured:
+                if self.raise_on_missing:
+                    raise
+                logger.warning("environment variable %s is not set", name)
+        return str_value.encode('utf-8') if isinstance(value, bytes) else str_value
+
     @classmethod
     def parse_value(cls, value, cast):
         """Parse and cast provided value

tests/fixtures.py

@@ -92,4 +92,17 @@ def generate_data(cls):
                     EXPORTED_VAR=cls.EXPORTED,
                     SAML_ATTRIBUTE_MAPPING='uid=username;mail=email;cn=first_name;sn=last_name;',
                     PREFIX_TEST='foo',
+                    FOO="foo",
+                    BAR="bar",
+                    INTERPOLATE_SIMPLE="$FOO",
+                    INTERPOLATE_SIMPLE_PARENTHESES="${FOO}",
+                    INTERPOLATE_SIMPLE_ESCAPED="\\$FOO",
+                    INTERPOLATE_SIMPLE_PARENTHESES_ESCAPED="\\${FOO}",
+                    INTERPOLATE_SIMPLE_PARENTHESES_NOT_OPENED="$FOO}",
+                    INTERPOLATE_SIMPLE_PARENTHESES_NOT_CLOSED="${FOO",
+                    INTERPOLATE_PARENTHESES_MISMATCH="${FOO$BAR}",
+                    INTERPOLATE_PREFIXED="PREFIXED$FOO",
+                    INTERPOLATE_SUFFIXED="$FOO@SUFFIXED",
+                    INTERPOLATE_MULTIPLE="$FOO$BAR",
+                    INTERPOLATE_MISSING="$MISSING",
                     )