Fix #1458 - Allow for custom parsing of environment variables via env_parse

changes/3977-acmiyaguchi.md

@@ -0,0 +1 @@
+Allow for custom parsing of environment variables via `env_parse` in `Field`.

docs/examples/settings_with_custom_parsing.py

@@ -0,0 +1,17 @@
+# output-json
+import os
+from typing import List
+
+from pydantic import BaseSettings, Field
+
+
+def parse_list(s: str) -> List[int]:
+    return [int(x.strip()) for x in s.split(',')]
+
+
+class Settings(BaseSettings):
+    numbers: List[int] = Field(env_parse=parse_list)
+
+
+os.environ['numbers'] = '1,2,3'
+print(Settings().dict())

docs/examples/settings_with_custom_parsing_validator.py

@@ -0,0 +1,17 @@
+# output-json
+import os
+from typing import List
+
+from pydantic import BaseSettings, Field, validator
+
+
+class Settings(BaseSettings):
+    numbers: List[int] = Field(env_parse=str)
+
+    @validator('numbers', pre=True)
+    def validate_numbers(cls, s):
+        return [int(x.strip()) for x in s.split(',')]
+
+
+os.environ['numbers'] = '1,2,3'
+print(Settings().dict())

docs/usage/settings.md

@@ -87,7 +87,7 @@ export SUB_MODEL__DEEP__V4=v4
 You could load a settings module thus:
 {!.tmp_examples/settings_nested_env.md!}
 
-`env_nested_delimiter` can be configured via the `Config` class as shown above, or via the 
+`env_nested_delimiter` can be configured via the `Config` class as shown above, or via the
 `_env_nested_delimiter` keyword argument on instantiation.
 
 JSON is only parsed in top-level fields, if you need to parse JSON in sub-models, you will need to implement
@@ -96,6 +96,17 @@ validators on those models.
 Nested environment variables take precedence over the top-level environment variable JSON
 (e.g. in the example above, `SUB_MODEL__V2` trumps `SUB_MODEL`).
 
+You may also populate a complex type by providing your own parsing function to
+`env_parse` in the field extras.
+
+{!.tmp_examples/settings_with_custom_parsing.md!}
+
+You might choose to pass the environment string value through to pydantic and
+transform in a validator instead.
+
+{!.tmp_examples/settings_with_custom_parsing_validator.md!}
+
+
 ## Dotenv (.env) support
 
 !!! note
@@ -178,7 +189,7 @@ see [python-dotenv's documentation](https://saurabh-kumar.com/python-dotenv/#usa
 
 Placing secret values in files is a common pattern to provide sensitive configuration to an application.
 
-A secret file follows the same principal as a dotenv file except it only contains a single value and the file name 
+A secret file follows the same principal as a dotenv file except it only contains a single value and the file name
 is used as the key. A secret file will look like the following:
 
 `/var/run/database_password`:
@@ -231,7 +242,7 @@ class Settings(BaseSettings):
         secrets_dir = '/run/secrets'
 ```
 !!! note
-    By default Docker uses `/run/secrets` as the target mount point. If you want to use a different location, change 
+    By default Docker uses `/run/secrets` as the target mount point. If you want to use a different location, change
     `Config.secrets_dir` accordingly.
 
 Then, create your secret via the Docker CLI

pydantic/env_settings.py

@@ -188,12 +188,17 @@ 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 the
+                    # parsing function passed via env_parse or json_loads, then
+                    # add explode_env_vars
+                    parse_func: Callable[[str], Any] = field.field_info.extra.get(
+                        'env_parse', settings.__config__.json_loads
+                    )
                     try:
-                        env_val = settings.__config__.json_loads(env_val)
+                        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))
@@ -298,10 +303,13 @@ def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
                 if path.is_file():
                     secret_value = path.read_text().strip()
                     if field.is_complex():
+                        parse_func: Callable[[str], Any] = field.field_info.extra.get(
+                            'env_parse', settings.__config__.json_loads
+                        )
                         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
                 else:

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,
@@ -221,7 +231,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()
 
 
@@ -1054,7 +1064,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()
 
 
@@ -1215,3 +1225,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'}