Add a custom CSV parser as fallback for complex pydantic-settings values

Author: ChrisLovering

olli/config.py

@@ -1,9 +1,23 @@
 """Configuration loading and validation."""
 
+import json
+import typing as t
 
 from loguru import logger
-from pydantic import BaseModel, field_validator
-from pydantic_settings import BaseSettings
+from pydantic import BaseModel, field_validator, fields
+from pydantic_settings import BaseSettings, PydanticBaseSettingsSource
+
+
+def custom_decode_complex_value(
+    __: str,
+    ___: fields.FieldInfo,
+    value: t.Any,
+) -> t.Any:
+    """Parse complex values as CSV if they cannot be parsed as JSON."""
+    try:
+        return json.loads(value)
+    except ValueError:
+        return value.split(",")
 
 
 class EnvConfig(
@@ -15,6 +29,25 @@ class EnvConfig(
 ):
     """Our default configuration for models that should load from .env files."""
 
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        """Monkey patch default sources to have the custom CSV parser fallback."""
+        dotenv_settings.decode_complex_value = custom_decode_complex_value
+        env_settings.decode_complex_value = custom_decode_complex_value
+        return (
+            init_settings,
+            env_settings,
+            dotenv_settings,
+            file_secret_settings,
+        )
+
 
 class _LokiConfig(EnvConfig, env_prefix="loki_"):
     """Loki specific configuration."""
@@ -69,7 +102,7 @@ def warn_above_ten(cls, value: list[TokenConfig]) -> list[TokenConfig]:
         This is because we cannot handle more than 10 token triggers at once until we
         batch triggers into groups of 10 to distribute to the webhook.
         """
-        if len(value) > 10:  # noqa: PLR2004
+        if len(value) > 10:
             logger.warning("More than 10 token triggers in one period cannot be handled, be careful.")
 
         return value

pyproject.toml

@@ -46,12 +46,13 @@ olli = "python -m olli"
 target-version = "py311"
 extend-exclude = [".cache"]
 ignore = [
-    "ANN002", "ANN003", "ANN101", "ANN102",
+    "ANN002", "ANN003", "ANN101", "ANN102", "ANN401",
     "ARG003",
     "C901",
     "D100", "D104", "D105", "D107", "D203", "D212", "D214", "D215", "D416",
     "EM101",
-    "TRY003"
+    "PLR",
+    "TRY003",
 ]
 line-length = 120
 select = ["ALL"]