Add support for parsing environment "None" strings to None. (#206)

Author: kschwab

pydantic_settings/main.py

@@ -26,6 +26,7 @@ class SettingsConfigDict(ConfigDict, total=False):
     env_file_encoding: str | None
     env_ignore_empty: bool
     env_nested_delimiter: str | None
+    env_parse_none_str: str | None
     secrets_dir: str | Path | None
 
 
@@ -56,6 +57,8 @@ class BaseSettings(BaseModel):
         _env_file_encoding: The env file encoding, e.g. `'latin-1'`. Defaults to `None`.
         _env_ignore_empty: Ignore environment variables where the value is an empty string. Default to `False`.
         _env_nested_delimiter: The nested env values delimiter. Defaults to `None`.
+        _env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.)
+            into `None` type(None). Defaults to `None` type(None), which means no parsing should occur.
         _secrets_dir: The secret files directory. Defaults to `None`.
     """
 
@@ -67,6 +70,7 @@ def __init__(
         _env_file_encoding: str | None = None,
         _env_ignore_empty: bool | None = None,
         _env_nested_delimiter: str | None = None,
+        _env_parse_none_str: str | None = None,
         _secrets_dir: str | Path | None = None,
         **values: Any,
     ) -> None:
@@ -80,6 +84,7 @@ def __init__(
                 _env_file_encoding=_env_file_encoding,
                 _env_ignore_empty=_env_ignore_empty,
                 _env_nested_delimiter=_env_nested_delimiter,
+                _env_parse_none_str=_env_parse_none_str,
                 _secrets_dir=_secrets_dir,
             )
         )
@@ -117,6 +122,7 @@ def _settings_build_values(
         _env_file_encoding: str | None = None,
         _env_ignore_empty: bool | None = None,
         _env_nested_delimiter: str | None = None,
+        _env_parse_none_str: str | None = None,
         _secrets_dir: str | Path | None = None,
     ) -> dict[str, Any]:
         # Determine settings config values
@@ -134,6 +140,9 @@ def _settings_build_values(
             if _env_nested_delimiter is not None
             else self.model_config.get('env_nested_delimiter')
         )
+        env_parse_none_str = (
+            _env_parse_none_str if _env_parse_none_str is not None else self.model_config.get('env_parse_none_str')
+        )
         secrets_dir = _secrets_dir if _secrets_dir is not None else self.model_config.get('secrets_dir')
 
         # Configure built-in sources
@@ -144,6 +153,7 @@ def _settings_build_values(
             env_prefix=env_prefix,
             env_nested_delimiter=env_nested_delimiter,
             env_ignore_empty=env_ignore_empty,
+            env_parse_none_str=env_parse_none_str,
         )
         dotenv_settings = DotEnvSettingsSource(
             self.__class__,
@@ -153,6 +163,7 @@ def _settings_build_values(
             env_prefix=env_prefix,
             env_nested_delimiter=env_nested_delimiter,
             env_ignore_empty=env_ignore_empty,
+            env_parse_none_str=env_parse_none_str,
         )
 
         file_secret_settings = SecretsSettingsSource(
@@ -183,6 +194,7 @@ def _settings_build_values(
         env_file_encoding=None,
         env_ignore_empty=False,
         env_nested_delimiter=None,
+        env_parse_none_str=None,
         secrets_dir=None,
         protected_namespaces=('model_', 'settings_'),
     )

pydantic_settings/sources.py

@@ -30,6 +30,10 @@
 ENV_FILE_SENTINEL: DotenvType = Path('')
 
 
+class EnvNoneType(str):
+    pass
+
+
 class SettingsError(ValueError):
     pass
 
@@ -134,13 +138,17 @@ def __init__(
         case_sensitive: bool | None = None,
         env_prefix: str | None = None,
         env_ignore_empty: bool | None = None,
+        env_parse_none_str: str | None = None,
     ) -> None:
         super().__init__(settings_cls)
         self.case_sensitive = case_sensitive if case_sensitive is not None else self.config.get('case_sensitive', False)
         self.env_prefix = env_prefix if env_prefix is not None else self.config.get('env_prefix', '')
         self.env_ignore_empty = (
             env_ignore_empty if env_ignore_empty is not None else self.config.get('env_ignore_empty', False)
         )
+        self.env_parse_none_str = (
+            env_parse_none_str if env_parse_none_str is not None else self.config.get('env_parse_none_str')
+        )
 
     def _apply_case_sensitive(self, value: str) -> str:
         return value.lower() if not self.case_sensitive else value
@@ -244,6 +252,20 @@ class Settings(BaseSettings):
 
         return values
 
+    def _replace_env_none_type_values(self, field_value: dict[str, Any]) -> dict[str, Any]:
+        """
+        Recursively parse values that are of "None" type(EnvNoneType) to `None` type(None).
+        """
+        values: dict[str, Any] = {}
+
+        for key, value in field_value.items():
+            if not isinstance(value, EnvNoneType):
+                values[key] = value if not isinstance(value, dict) else self._replace_env_none_type_values(value)
+            else:
+                values[key] = None
+
+        return values
+
     def __call__(self) -> dict[str, Any]:
         data: dict[str, Any] = {}
 
@@ -263,6 +285,11 @@ def __call__(self) -> dict[str, Any]:
                 ) from e
 
             if field_value is not None:
+                if self.env_parse_none_str is not None:
+                    if isinstance(field_value, dict):
+                        field_value = self._replace_env_none_type_values(field_value)
+                    elif isinstance(field_value, EnvNoneType):
+                        field_value = None
                 if (
                     not self.case_sensitive
                     and lenient_issubclass(field.annotation, BaseModel)
@@ -287,8 +314,9 @@ def __init__(
         case_sensitive: bool | None = None,
         env_prefix: str | None = None,
         env_ignore_empty: bool | None = None,
+        env_parse_none_str: str | None = None,
     ) -> None:
-        super().__init__(settings_cls, case_sensitive, env_prefix, env_ignore_empty)
+        super().__init__(settings_cls, case_sensitive, env_prefix, env_ignore_empty, env_parse_none_str)
         self.secrets_dir = secrets_dir if secrets_dir is not None else self.config.get('secrets_dir')
 
     def __call__(self) -> dict[str, Any]:
@@ -376,8 +404,9 @@ def __init__(
         env_prefix: str | None = None,
         env_nested_delimiter: str | None = None,
         env_ignore_empty: bool | None = None,
+        env_parse_none_str: str | None = None,
     ) -> None:
-        super().__init__(settings_cls, case_sensitive, env_prefix, env_ignore_empty)
+        super().__init__(settings_cls, case_sensitive, env_prefix, env_ignore_empty, env_parse_none_str)
         self.env_nested_delimiter = (
             env_nested_delimiter if env_nested_delimiter is not None else self.config.get('env_nested_delimiter')
         )
@@ -386,7 +415,7 @@ def __init__(
         self.env_vars = self._load_env_vars()
 
     def _load_env_vars(self) -> Mapping[str, str | None]:
-        return parse_env_vars(os.environ, self.case_sensitive, self.env_ignore_empty)
+        return parse_env_vars(os.environ, self.case_sensitive, self.env_ignore_empty, self.env_parse_none_str)
 
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         """
@@ -570,12 +599,15 @@ def __init__(
         env_prefix: str | None = None,
         env_nested_delimiter: str | None = None,
         env_ignore_empty: bool | None = None,
+        env_parse_none_str: str | None = None,
     ) -> None:
         self.env_file = env_file if env_file != ENV_FILE_SENTINEL else settings_cls.model_config.get('env_file')
         self.env_file_encoding = (
             env_file_encoding if env_file_encoding is not None else settings_cls.model_config.get('env_file_encoding')
         )
-        super().__init__(settings_cls, case_sensitive, env_prefix, env_nested_delimiter, env_ignore_empty)
+        super().__init__(
+            settings_cls, case_sensitive, env_prefix, env_nested_delimiter, env_ignore_empty, env_parse_none_str
+        )
 
     def _load_env_vars(self) -> Mapping[str, str | None]:
         return self._read_env_files()
@@ -598,6 +630,7 @@ def _read_env_files(self) -> Mapping[str, str | None]:
                         encoding=self.env_file_encoding,
                         case_sensitive=self.case_sensitive,
                         ignore_empty=self.env_ignore_empty,
+                        parse_none_str=self.env_parse_none_str,
                     )
                 )
 
@@ -635,10 +668,21 @@ def _get_env_var_key(key: str, case_sensitive: bool = False) -> str:
     return key if case_sensitive else key.lower()
 
 
+def _parse_env_none_str(value: str | None, parse_none_str: str | None = None) -> str | None | EnvNoneType:
+    return value if not (value == parse_none_str and parse_none_str is not None) else EnvNoneType(value)
+
+
 def parse_env_vars(
-    env_vars: Mapping[str, str | None], case_sensitive: bool = False, ignore_empty: bool = False
+    env_vars: Mapping[str, str | None],
+    case_sensitive: bool = False,
+    ignore_empty: bool = False,
+    parse_none_str: str | None = None,
 ) -> Mapping[str, str | None]:
-    return {_get_env_var_key(k, case_sensitive): v for k, v in env_vars.items() if not (ignore_empty and v == '')}
+    return {
+        _get_env_var_key(k, case_sensitive): _parse_env_none_str(v, parse_none_str)
+        for k, v in env_vars.items()
+        if not (ignore_empty and v == '')
+    }
 
 
 def read_env_file(
@@ -647,9 +691,10 @@ def read_env_file(
     encoding: str | None = None,
     case_sensitive: bool = False,
     ignore_empty: bool = False,
+    parse_none_str: str | None = None,
 ) -> Mapping[str, str | None]:
     file_vars: dict[str, str | None] = dotenv_values(file_path, encoding=encoding or 'utf8')
-    return parse_env_vars(file_vars, case_sensitive, ignore_empty)
+    return parse_env_vars(file_vars, case_sensitive, ignore_empty, parse_none_str)
 
 
 def _annotation_is_complex(annotation: type[Any] | None, metadata: list[Any]) -> bool:

tests/test_settings.py

@@ -1785,6 +1785,46 @@ class Settings(BaseSettings):
     ]
 
 
+def test_env_parse_none_str(env):
+    env.set('x', 'null')
+    env.set('y', 'y_override')
+
+    class Settings(BaseSettings):
+        x: Optional[str] = 'x_default'
+        y: Optional[str] = 'y_default'
+
+    s = Settings()
+    assert s.x == 'null'
+    assert s.y == 'y_override'
+    s = Settings(_env_parse_none_str='null')
+    assert s.x is None
+    assert s.y == 'y_override'
+
+    env.set('nested__x', 'None')
+    env.set('nested__y', 'y_override')
+    env.set('nested__deep__z', 'None')
+
+    class NestedBaseModel(BaseModel):
+        x: Optional[str] = 'x_default'
+        y: Optional[str] = 'y_default'
+        deep: Optional[dict] = {'z': 'z_default'}
+        keep: Optional[dict] = {'z': 'None'}
+
+    class NestedSettings(BaseSettings, env_nested_delimiter='__'):
+        nested: Optional[NestedBaseModel] = NestedBaseModel()
+
+    s = NestedSettings()
+    assert s.nested.x == 'None'
+    assert s.nested.y == 'y_override'
+    assert s.nested.deep['z'] == 'None'
+    assert s.nested.keep['z'] == 'None'
+    s = NestedSettings(_env_parse_none_str='None')
+    assert s.nested.x is None
+    assert s.nested.y == 'y_override'
+    assert s.nested.deep['z'] is None
+    assert s.nested.keep['z'] == 'None'
+
+
 def test_env_json_field_dict(env):
     class Settings(BaseSettings):
         x: Json[Dict[str, int]]