Fix pydantic#1458 - Allow for custom parsing of environment variables…

… via env_parse

pydantic/env_settings.py

@@ -187,12 +187,18 @@ def __call__(self, settings: BaseSettings) -> Dict[str, Any]:  # noqa C901
                     if env_val_built:
                         d[field.alias] = env_val_built
                 else:
-                    # field is complex and there's a value, decode that as JSON, then add explode_env_vars
+                    # field is complex and there's a value, decode using an
+                    # env_parse or JSON, then add explode_env_vars
+                    is_json_parse = field.field_info.extra.get("env_parse") is None
+                    parse_func: Callable[[str], Any] = (
+                        settings.__config__.json_loads if is_json_parse else field.field_info.extra["env_parse"]
+                    )
                     try:
-                        env_val = settings.__config__.json_loads(env_val)
+                        print(parse_func)
+                        env_val = parse_func(env_val)
                     except ValueError as e:
                         if not allow_json_failure:
-                            raise SettingsError(f'error parsing JSON for "{env_name}"') from e
+                            raise SettingsError(f'error parsing envvar "{env_name}"') from e
 
                     if isinstance(env_val, dict):
                         d[field.alias] = deep_update(env_val, self.explode_env_vars(field, env_vars))
@@ -273,10 +279,14 @@ def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
                 if path.is_file():
                     secret_value = path.read_text().strip()
                     if field.is_complex():
+                        is_json_parse = field.field_info.extra.get("env_parse") is None
+                        parse_func: Callable[[str], Any] = (
+                            settings.__config__.json_loads if is_json_parse else field.field_info.extra["env_parse"]
+                        )
                         try:
-                            secret_value = settings.__config__.json_loads(secret_value)
+                            secret_value = parse_func(secret_value)
                         except ValueError as e:
-                            raise SettingsError(f'error parsing JSON for "{env_name}"') from e
+                            raise SettingsError(f'error parsing envvar "{env_name}"') from e
 
                     secrets[field.alias] = secret_value
                 elif path.exists():

tests/test_settings.py

@@ -3,11 +3,21 @@
 import uuid
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
 
 import pytest
 
-from pydantic import BaseModel, BaseSettings, Field, HttpUrl, NoneStr, SecretStr, ValidationError, dataclasses
+from pydantic import (
+    BaseModel,
+    BaseSettings,
+    Field,
+    HttpUrl,
+    NoneStr,
+    SecretStr,
+    ValidationError,
+    dataclasses,
+    validator,
+)
 from pydantic.env_settings import (
     EnvSettingsSource,
     InitSettingsSource,
@@ -194,7 +204,7 @@ def test_set_dict_model(env):
 
 def test_invalid_json(env):
     env.set('apples', '["russet", "granny smith",]')
-    with pytest.raises(SettingsError, match='error parsing JSON for "apples"'):
+    with pytest.raises(SettingsError, match='error parsing envvar "apples"'):
         ComplexSettings()
 
 
@@ -927,7 +937,7 @@ class Settings(BaseSettings):
         class Config:
             secrets_dir = tmp_path
 
-    with pytest.raises(SettingsError, match='error parsing JSON for "foo"'):
+    with pytest.raises(SettingsError, match='error parsing envvar "foo"'):
         Settings()
 
 
@@ -1087,3 +1097,60 @@ def test_builtins_settings_source_repr():
         == "EnvSettingsSource(env_file='.env', env_file_encoding='utf-8', env_nested_delimiter=None)"
     )
     assert repr(SecretsSettingsSource(secrets_dir='/secrets')) == "SecretsSettingsSource(secrets_dir='/secrets')"
+
+
+def _parse_custom_dict(value: str) -> Callable[[str], Dict[int, str]]:
+    """A custom parsing function passed into env parsing test."""
+    res = {}
+    for part in value.split(","):
+        k, v = part.split("=")
+        res[int(k)] = v
+    return res
+
+
+def test_env_setting_source_custom_env_parse(env):
+    class Settings(BaseSettings):
+        top: Dict[int, str] = Field(env_parse=_parse_custom_dict)
+
+    with pytest.raises(ValidationError):
+        Settings()
+    env.set('top', '1=apple,2=banana')
+    s = Settings()
+    assert s.top == {1: 'apple', 2: 'banana'}
+
+
+def test_env_settings_source_custom_env_parse_is_bad(env):
+    class Settings(BaseSettings):
+        top: Dict[int, str] = Field(env_parse=int)
+
+    env.set('top', '1=apple,2=banana')
+    with pytest.raises(SettingsError, match='error parsing envvar "top"'):
+        Settings()
+
+
+def test_env_settings_source_custom_env_parse_validator(env):
+    class Settings(BaseSettings):
+        top: Dict[int, str] = Field(env_parse=str)
+
+        @validator('top', pre=True)
+        def val_top(cls, v):
+            assert isinstance(v, str)
+            return _parse_custom_dict(v)
+
+    env.set('top', '1=apple,2=banana')
+    s = Settings()
+    assert s.top == {1: 'apple', 2: 'banana'}
+
+
+def test_secret_settings_source_custom_env_parse(tmp_path):
+    p = tmp_path / 'top'
+    p.write_text('1=apple,2=banana')
+
+    class Settings(BaseSettings):
+        top: Dict[int, str] = Field(env_parse=_parse_custom_dict)
+
+        class Config:
+            secrets_dir = tmp_path
+
+    s = Settings()
+    assert s.top == {1: 'apple', 2: 'banana'}