✨ 允许插件从环境变量中读取配置项而不需要在envfile中声明

Author: AzideCupric

nonebot/config.py

@@ -14,14 +14,15 @@
 """
 
 import abc
-from collections.abc import Mapping
+from collections.abc import Iterable, Mapping
 from datetime import timedelta
+from functools import lru_cache
 from ipaddress import IPv4Address
 import json
 import os
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional, Union
-from typing_extensions import TypeAlias, get_args, get_origin
+from typing_extensions import TypeAlias, get_args, get_origin, override
 
 from dotenv import dotenv_values
 from pydantic import BaseModel, Field
@@ -79,7 +80,7 @@ def __repr__(self) -> str:
         return f"InitSettingsSource(init_kwargs={self.init_kwargs!r})"
 
 
-class DotEnvSettingsSource(BaseSettingsSource):
+class EnvSettingsSource(BaseSettingsSource):
     def __init__(
         self,
         settings_cls: type["BaseSettings"],
@@ -110,6 +111,11 @@ def __init__(
             else self.config.get("env_nested_delimiter", None)
         )
 
+    @abc.abstractmethod
+    def get_setting_fields(self) -> Iterable[ModelField]:
+        """获取配置类的字段信息"""
+        raise NotImplementedError
+
     def _apply_case_sensitive(self, var_name: str) -> str:
         return var_name if self.case_sensitive else var_name.lower()
 
@@ -133,6 +139,7 @@ def _read_env_file(self, file_path: Path) -> dict[str, Optional[str]]:
         file_vars = dotenv_values(file_path, encoding=self.env_file_encoding)
         return self._parse_env_vars(file_vars)
 
+    @lru_cache
     def _read_env_files(self) -> dict[str, Optional[str]]:
         env_files = self.env_file
         if env_files is None:
@@ -209,8 +216,8 @@ def __call__(self) -> dict[str, Any]:
         env_file_vars = self._read_env_files()
         env_vars = {**env_file_vars, **env_vars}
 
-        for field in model_fields(self.settings_cls):
-            field_name = field.name
+        for field in self.get_setting_fields():
+            field_name = self._parse_field_name(field)
             env_name = self._apply_case_sensitive(field_name)
 
             # try get values from env vars
@@ -283,6 +290,52 @@ def __call__(self) -> dict[str, Any]:
 
         return d
 
+    def _parse_field_name(self, field: ModelField) -> str:
+        return field.field_info.alias or field.name
+
+
+class DotEnvSettingsSource(EnvSettingsSource):
+    def __init__(
+        self,
+        settings_cls: type["BaseSettings"],
+        env_file: Optional[DOTENV_TYPE] = ENV_FILE_SENTINEL,
+        env_file_encoding: Optional[str] = None,
+        case_sensitive: Optional[bool] = None,
+        env_nested_delimiter: Optional[str] = None,
+    ) -> None:
+        super().__init__(
+            settings_cls,
+            env_file,
+            env_file_encoding,
+            case_sensitive,
+            env_nested_delimiter,
+        )
+
+    @override
+    def get_setting_fields(self) -> Iterable[ModelField]:
+        return model_fields(self.settings_cls)
+
+
+class PluginEnvSettingsSource(EnvSettingsSource):
+    def __init__(
+        self,
+        config_cls: type[BaseModel],
+        driver_config: "Config",
+    ) -> None:
+        setting_config: "SettingsConfig" = model_config(driver_config.__class__)
+        super().__init__(
+            BaseSettings,
+            env_file=setting_config.get("env_file", None),
+            env_file_encoding=setting_config.get("env_file_encoding", "utf-8"),
+            case_sensitive=setting_config.get("case_sensitive", False),
+            env_nested_delimiter=setting_config.get("env_nested_delimiter", None),
+        )
+        self.config_cls = config_cls
+
+    @override
+    def get_setting_fields(self) -> Iterable[ModelField]:
+        return model_fields(self.config_cls)
+
 
 if PYDANTIC_V2:  # pragma: pydantic-v2
 

nonebot/plugin/__init__.py

@@ -46,7 +46,8 @@
 from pydantic import BaseModel
 
 from nonebot import get_driver
-from nonebot.compat import model_dump, type_validate_python
+from nonebot.compat import type_validate_python
+from nonebot.config import PluginEnvSettingsSource
 
 C = TypeVar("C", bound=BaseModel)
 
@@ -172,7 +173,8 @@ def get_available_plugin_names() -> set[str]:
 
 def get_plugin_config(config: type[C]) -> C:
     """从全局配置获取当前插件需要的配置项。"""
-    return type_validate_python(config, model_dump(get_driver().config))
+    env_settings = PluginEnvSettingsSource(config, get_driver().config)
+    return type_validate_python(config, env_settings())
 
 
 from .load import inherit_supported_adapters as inherit_supported_adapters

tests/test_plugin/test_load.py

@@ -8,6 +8,7 @@
 import pytest
 
 import nonebot
+from nonebot.compat import model_dump
 from nonebot.plugin import (
     Plugin,
     PluginManager,
@@ -184,6 +185,18 @@ def test_plugin_metadata():
     assert plugin.metadata.get_supported_adapters() == {FakeAdapter}
 
 
+def test_plugin_load_env_config(monkeypatch: pytest.MonkeyPatch):
+    no_dummy_val = "no_dummy_val"
+    monkeypatch.setenv("CUSTOM", no_dummy_val)
+    from plugins.metadata import Config
+
+    global_config = nonebot.get_driver().config
+    assert "custom" not in model_dump(global_config)
+
+    config = nonebot.get_plugin_config(Config)
+    assert config.custom == no_dummy_val
+
+
 def test_inherit_supported_adapters_not_found():
     with pytest.raises(RuntimeError):
         inherit_supported_adapters("some_plugin_not_exist")