dify/api/configs/app_config.py

import logging
from pathlib import Path
from typing import Any, override

from pydantic import PrivateAttr, field_serializer, model_validator
from pydantic.fields import FieldInfo
from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource

from libs.file_utils import search_file_upwards

from .deploy import DeploymentConfig
from .enterprise import EnterpriseFeatureConfig, EnterpriseTelemetryConfig
from .extra import ExtraServiceConfig
from .feature import FeatureConfig
from .middleware import MiddlewareConfig
from .observability import ObservabilityConfig
from .packaging import PackagingInfo
from .remote_settings_sources import RemoteSettingsSource, RemoteSettingsSourceConfig, RemoteSettingsSourceName
from .remote_settings_sources.apollo import ApolloSettingsSource
from .remote_settings_sources.nacos import NacosSettingsSource
from .tokener_mapping_registry import merge_mapping_registry_file

logger = logging.getLogger(__name__)


class RemoteSettingsSourceFactory(PydanticBaseSettingsSource):
    def __init__(self, settings_cls: type[BaseSettings]):
        super().__init__(settings_cls)

    @override
    def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
        raise NotImplementedError

    @override
    def __call__(self) -> dict[str, Any]:
        current_state = self.current_state
        remote_source_name = current_state.get("REMOTE_SETTINGS_SOURCE_NAME")
        if not remote_source_name:
            return {}

        remote_source: RemoteSettingsSource | None = None
        match remote_source_name:
            case RemoteSettingsSourceName.APOLLO:
                remote_source = ApolloSettingsSource(current_state)
            case RemoteSettingsSourceName.NACOS:
                remote_source = NacosSettingsSource(current_state)
            case _:
                logger.warning("Unsupported remote source: %s", remote_source_name)
                return {}

        d: dict[str, Any] = {}

        for field_name, field in self.settings_cls.model_fields.items():
            field_value, field_key, value_is_complex = remote_source.get_field_value(field, field_name)
            field_value = remote_source.prepare_field_value(field_name, field, field_value, value_is_complex)
            if field_value is not None:
                d[field_key] = field_value

        return d


class DifyConfig(
    # Packaging info
    PackagingInfo,
    # Deployment configs
    DeploymentConfig,
    # Feature configs
    FeatureConfig,
    # Middleware configs
    MiddlewareConfig,
    # Extra service configs
    ExtraServiceConfig,
    # Observability configs
    ObservabilityConfig,
    # Remote source configs
    RemoteSettingsSourceConfig,
    # Enterprise feature configs
    # **Before using, please contact business@dify.ai by email to inquire about licensing matters.**
    EnterpriseFeatureConfig,
    # Enterprise telemetry configs
    EnterpriseTelemetryConfig,
):
    model_config = SettingsConfigDict(
        # read from dotenv format config file
        env_file=".env",
        env_file_encoding="utf-8",
        # ignore extra attributes
        extra="ignore",
    )

    # Before adding any config,
    # please consider to arrange it in the proper config group of existed or added
    # for better readability and maintainability.
    # Thanks for your concentration and consideration.

    _tokener_mapping_file_loaded: bool = PrivateAttr(default=False)
    _tokener_inline_registry: str = PrivateAttr(default="[]")

    @model_validator(mode="after")
    def load_tokener_mapping_registry_file(self) -> "DifyConfig":
        """Resolve one startup snapshot without exporting it into process env."""
        if self.TOKENER_LEGACY_MODEL_MAPPING_FILE and not self._tokener_mapping_file_loaded:
            inline = self.TOKENER_LEGACY_MODEL_MAPPING_JSON
            merged = merge_mapping_registry_file(inline, self.TOKENER_LEGACY_MODEL_MAPPING_FILE)
            self._tokener_inline_registry = inline
            self.TOKENER_LEGACY_MODEL_MAPPING_JSON = merged
            self._tokener_mapping_file_loaded = True
        return self

    @field_serializer("TOKENER_LEGACY_MODEL_MAPPING_JSON")
    def serialize_tokener_mapping_input(self, value: str) -> str:
        """Dump source settings, so dump-to-rebuild never appends the file twice.

        Runtime consumers read the effective field directly. An existing model
        keeps its snapshot when revalidated; a rebuilt model reads the file anew.
        """
        return self._tokener_inline_registry if self._tokener_mapping_file_loaded else value

    @model_validator(mode="after")
    def validate_knowledge_fs_production_origin(self) -> "DifyConfig":
        if self.DEPLOY_ENV.strip().upper() == "PRODUCTION":
            if self.KNOWLEDGE_FS_BASE_URL and not self._is_secure_or_loopback_origin(self.KNOWLEDGE_FS_BASE_URL):
                raise ValueError("KNOWLEDGE_FS_BASE_URL must use HTTPS in production unless it targets loopback")
        return self

    @classmethod
    @override
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        return (
            init_settings,
            env_settings,
            RemoteSettingsSourceFactory(settings_cls),
            dotenv_settings,
            file_secret_settings,
            TomlConfigSettingsSource(
                settings_cls=settings_cls,
                toml_file=search_file_upwards(
                    base_dir_path=Path(__file__).parent,
                    target_file_name="pyproject.toml",
                    max_search_parent_depth=2,
                ),
            ),
        )