Files
dify/api/services/web_app_runtime_query_service.py
林玮 (Jade Lin)andautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> 0c443b7666 fix(api): handle invalid app icon references (#41977)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-09-09 02:56:40 +00:00

126 lines
4.3 KiB
Python

"""Application service for building the public Web app runtime bootstrap."""
import json
import logging
from collections.abc import Callable, Mapping
from typing import NamedTuple, Protocol, cast
from enums import DeploymentEdition
from services.app_definition_query_service import AppSiteConfiguration
from services.entities.feature_entities import FeatureModel
from services.errors.file import FileNotExistsError
from services.file_service import FileService
from services.icon_configuration import DEFAULT_ICON, DEFAULT_ICON_BACKGROUND, DEFAULT_ICON_TYPE
logger = logging.getLogger(__name__)
class WebAppRuntimeRecord(NamedTuple):
app_id: str
tenant_id: str
mode: str
enable_site: bool
site: AppSiteConfiguration
plan: str
tenant_status: str
# Keep this lazy: workspaces without custom branding never parsed this legacy field.
tenant_custom_config_json: str | None
class WebAppBootstrap(NamedTuple):
app_id: str
mode: str
enable_site: bool
site: Mapping[str, str | bool | None]
plan: str
can_replace_logo: bool
custom_config: Mapping[str, str | bool | None] | None
class WebAppRuntimeQuery(Protocol):
def get_runtime_record(self, app_id: str) -> WebAppRuntimeRecord | None: ...
class WebAppRuntimeUnavailableError(ValueError):
"""Raised when the admitted Web app can no longer be bootstrapped."""
_ARCHIVED_TENANT_STATUS = "archive"
class WebAppRuntimeQueryService:
def __init__(
self,
*,
runtime: WebAppRuntimeQuery,
file_service: FileService,
workspace_features: Callable[[str], FeatureModel],
files_url: str,
deployment_edition: DeploymentEdition,
) -> None:
self._runtime = runtime
self._file_service = file_service
self._workspace_features = workspace_features
self._files_url = files_url
self._deployment_edition = deployment_edition
def get_bootstrap(self, app_id: str) -> WebAppBootstrap:
record = self._runtime.get_runtime_record(app_id)
if record is None or record.tenant_status == _ARCHIVED_TENANT_STATUS:
raise WebAppRuntimeUnavailableError("Site not found")
features = self._workspace_features(record.tenant_id)
site_icon_unavailable = False
try:
site_icon_url = (
self._file_service.get_icon_url(record.site.icon, record.tenant_id)
if record.site.icon_type == "image" and record.site.icon
else None
)
except FileNotExistsError as exc:
logger.warning(
"Web app site icon is unavailable: app_id=%s tenant_id=%s file_id=%s",
record.app_id,
record.tenant_id,
record.site.icon,
)
site_icon_url = None
site_icon_unavailable = True
site = cast(dict[str, str | bool | None], record.site._asdict())
site["icon_url"] = site_icon_url
if site_icon_unavailable:
site["icon_type"] = DEFAULT_ICON_TYPE.value
site["icon"] = DEFAULT_ICON
site["icon_background"] = DEFAULT_ICON_BACKGROUND
if self._deployment_edition == DeploymentEdition.CLOUD and not features.webapp_copyright_enabled:
site["copyright"] = None
site["input_placeholder"] = None
custom_config = None
if features.can_replace_logo:
tenant_custom_config = (
cast(Mapping[str, str | bool | None], json.loads(record.tenant_custom_config_json))
if record.tenant_custom_config_json
else {}
)
replace_webapp_logo = (
f"{self._files_url}/files/workspaces/{record.tenant_id}/webapp-logo"
if tenant_custom_config.get("replace_webapp_logo")
else None
)
custom_config = {
"remove_webapp_brand": tenant_custom_config.get("remove_webapp_brand", False),
"replace_webapp_logo": replace_webapp_logo,
}
return WebAppBootstrap(
app_id=record.app_id,
mode=record.mode,
enable_site=record.enable_site,
site=site,
plan=record.plan,
can_replace_logo=features.can_replace_logo,
custom_config=custom_config,
)