Migrate file integration to config entry

homeassistant/components/file/__init__.py

@@ -1 +1,112 @@
 """The file component."""
+
+import os
+
+from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
+from homeassistant.const import CONF_FILE_PATH, CONF_FILENAME, CONF_PLATFORM, Platform
+from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers import (
+    config_validation as cv,
+    discovery,
+    issue_registry as ir,
+)
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DOMAIN
+
+CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
+
+PLATFORMS = [Platform.SENSOR]
+
+YAML_PLATFORMS = [Platform.NOTIFY, Platform.SENSOR]
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the file integration."""
+
+    # The YAML config was imported with HA Core 2024.6.0 and will be removed with
+    # HA Core 2024.12
+    ir.async_create_issue(
+        hass,
+        HOMEASSISTANT_DOMAIN,
+        f"deprecated_yaml_{DOMAIN}",
+        breaks_in_ha_version="2024.12",
+        is_fixable=False,
+        issue_domain=DOMAIN,
+        learn_more_url="https://www.home-assistant.io/integrations/file/",
+        severity=ir.IssueSeverity.WARNING,
+        translation_key="deprecated_yaml",
+        translation_placeholders={
+            "domain": DOMAIN,
+            "integration_title": "File",
+        },
+    )
+    if hass.config_entries.async_entries(DOMAIN):
+        # We skip import in case we already have config entries
+        return True
+
+    # Prepare to import the YAML config into separate config entries
+    config_data: dict[str, list[ConfigType]] = {
+        domain: [item for item in config[domain] if item.pop(CONF_PLATFORM) == DOMAIN]
+        for domain in config
+        if domain in YAML_PLATFORMS
+    }
+
+    for domain, items in config_data.items():
+        for item in items:
+            item[CONF_PLATFORM] = domain
+            hass.async_create_task(
+                hass.config_entries.flow.async_init(
+                    DOMAIN,
+                    context={"source": SOURCE_IMPORT},
+                    data=item,
+                )
+            )
+
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up a file component entry."""
+    config = dict(entry.data)
+    filepath: str = ""
+    if (platform := config[CONF_PLATFORM]) == Platform.NOTIFY:
+        filepath = os.path.join(hass.config.config_dir, config[CONF_FILENAME])
+    elif platform == Platform.SENSOR:
+        filepath = config[CONF_FILE_PATH]
+    if filepath and not await hass.async_add_executor_job(
+        hass.config.is_allowed_path, filepath
+    ):
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="dir_not_allowed",
+            translation_placeholders={"filename": filepath},
+        )
+
+    if entry.data[CONF_PLATFORM] in PLATFORMS:
+        await hass.config_entries.async_forward_entry_setups(
+            entry, [Platform(entry.data[CONF_PLATFORM])]
+        )
+    else:
+        # The notify platform is not yet set up as entry, so
+        # forward setup config through discovery to ensure setup notify service.
+        # This is needed as long as the legacy service is not migrated
+        hass.async_create_task(
+            discovery.async_load_platform(
+                hass,
+                Platform.NOTIFY,
+                DOMAIN,
+                config,
+                {},
+            )
+        )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(
+        entry, [entry.data[CONF_PLATFORM]]
+    )

homeassistant/components/file/config_flow.py

@@ -0,0 +1,126 @@
+"""Config flow for file integration."""
+
+import os
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import (
+    CONF_FILE_PATH,
+    CONF_FILENAME,
+    CONF_NAME,
+    CONF_PLATFORM,
+    CONF_UNIT_OF_MEASUREMENT,
+    CONF_VALUE_TEMPLATE,
+    Platform,
+)
+from homeassistant.helpers.selector import (
+    BooleanSelector,
+    BooleanSelectorConfig,
+    TemplateSelector,
+    TemplateSelectorConfig,
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import CONF_TIMESTAMP, DEFAULT_NAME, DOMAIN
+
+BOOLEAN_SELECTOR = BooleanSelector(BooleanSelectorConfig())
+TEMPLATE_SELECTOR = TemplateSelector(TemplateSelectorConfig())
+TEXT_SELECTOR = TextSelector(TextSelectorConfig(type=TextSelectorType.TEXT))
+
+FILE_SENSOR_SCHEMA = vol.Schema(
+    {
+        vol.Optional(CONF_NAME): TEXT_SELECTOR,
+        vol.Required(CONF_FILE_PATH): TEXT_SELECTOR,
+        vol.Optional(CONF_VALUE_TEMPLATE): TEMPLATE_SELECTOR,
+        vol.Optional(CONF_UNIT_OF_MEASUREMENT): TEXT_SELECTOR,
+    }
+)
+
+FILE_NOTIFY_SCHEMA = vol.Schema(
+    {
+        vol.Optional(CONF_NAME): TEXT_SELECTOR,
+        vol.Required(CONF_FILENAME): TEXT_SELECTOR,
+        vol.Optional(CONF_TIMESTAMP, default=False): BOOLEAN_SELECTOR,
+    }
+)
+
+
+class FileConfigFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a file config flow."""
+
+    VERSION = 1
+
+    async def validate_file_path(self, file_path: str) -> bool:
+        """Ensure the file path is valid."""
+        return await self.hass.async_add_executor_job(
+            self.hass.config.is_allowed_path, file_path
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initiated by the user."""
+        return self.async_show_menu(
+            step_id="user",
+            menu_options=["notify", "sensor"],
+        )
+
+    async def async_step_notify(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle file notifier config flow."""
+        errors: dict[str, str] = {}
+        if user_input:
+            user_input[CONF_PLATFORM] = "notify"
+            self._async_abort_entries_match(user_input)
+            filepath: str = os.path.join(
+                self.hass.config.config_dir, user_input[CONF_FILENAME]
+            )
+            if not await self.validate_file_path(filepath):
+                errors[CONF_FILENAME] = "not_allowed"
+            else:
+                name: str = user_input.get(CONF_NAME, DEFAULT_NAME)
+                title = f"{name} [{user_input[CONF_FILENAME]}]"
+                return self.async_create_entry(data=user_input, title=title)
+
+        return self.async_show_form(
+            step_id="notify", data_schema=FILE_NOTIFY_SCHEMA, errors=errors
+        )
+
+    async def async_step_sensor(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle file sensor config flow."""
+        errors: dict[str, str] = {}
+        if user_input:
+            user_input[CONF_PLATFORM] = "sensor"
+            self._async_abort_entries_match(user_input)
+            if not await self.validate_file_path(user_input[CONF_FILE_PATH]):
+                errors[CONF_FILE_PATH] = "not_allowed"
+            else:
+                title = f"{DEFAULT_NAME} [{user_input[CONF_FILE_PATH]}]"
+                return self.async_create_entry(data=user_input, title=title)
+
+        return self.async_show_form(
+            step_id="sensor", data_schema=FILE_SENSOR_SCHEMA, errors=errors
+        )
+
+    async def async_step_import(
+        self, import_data: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Import `file`` config from configuration.yaml."""
+        assert import_data is not None
+        self._async_abort_entries_match(import_data)
+        platform = import_data[CONF_PLATFORM]
+        name: str = import_data.get(CONF_NAME, DEFAULT_NAME)
+        file_name: str
+        if platform == Platform.NOTIFY:
+            file_name = import_data[CONF_FILENAME]
+        else:
+            file_name = import_data[CONF_FILE_PATH]
+        title = f"{name} [{file_name}]"
+        return self.async_create_entry(title=title, data=import_data)

homeassistant/components/file/const.py

@@ -0,0 +1,8 @@
+"""Constants for the file integration."""
+
+DOMAIN = "file"
+
+CONF_TIMESTAMP = "timestamp"
+
+DEFAULT_NAME = "File"
+FILE_ICON = "mdi:file"

homeassistant/components/file/manifest.json

@@ -2,6 +2,7 @@
   "domain": "file",
   "name": "File",
   "codeowners": ["@fabaff"],
+  "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/file",
   "iot_class": "local_polling",
   "requirements": ["file-read-backwards==2.0.0"]

homeassistant/components/file/notify.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import os
 from typing import Any, TextIO
 
@@ -15,11 +16,14 @@
 )
 from homeassistant.const import CONF_FILENAME
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ServiceValidationError
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 import homeassistant.util.dt as dt_util
 
-CONF_TIMESTAMP = "timestamp"
+from .const import CONF_TIMESTAMP, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
 
 PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
     {
@@ -29,14 +33,17 @@
 )
 
 
-def get_service(
+async def async_get_service(
     hass: HomeAssistant,
     config: ConfigType,
     discovery_info: DiscoveryInfoType | None = None,
-) -> FileNotificationService:
+) -> FileNotificationService | None:
     """Get the file notification service."""
-    filename: str = config[CONF_FILENAME]
-    timestamp: bool = config[CONF_TIMESTAMP]
+    if discovery_info is None:
+        # We only set up through discovery
+        return None
+    filename: str = discovery_info[CONF_FILENAME]
+    timestamp: bool = discovery_info.get(CONF_TIMESTAMP, False)
 
     return FileNotificationService(filename, timestamp)
 
@@ -53,16 +60,23 @@ def send_message(self, message: str = "", **kwargs: Any) -> None:
         """Send a message to a file."""
         file: TextIO
         filepath: str = os.path.join(self.hass.config.config_dir, self.filename)
-        with open(filepath, "a", encoding="utf8") as file:
-            if os.stat(filepath).st_size == 0:
-                title = (
-                    f"{kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)} notifications (Log"
-                    f" started: {dt_util.utcnow().isoformat()})\n{'-' * 80}\n"
-                )
-                file.write(title)
+        try:
+            with open(filepath, "a", encoding="utf8") as file:
+                if os.stat(filepath).st_size == 0:
+                    title = (
+                        f"{kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)} notifications (Log"
+                        f" started: {dt_util.utcnow().isoformat()})\n{'-' * 80}\n"
+                    )
+                    file.write(title)
 
-            if self.add_timestamp:
-                text = f"{dt_util.utcnow().isoformat()} {message}\n"
-            else:
-                text = f"{message}\n"
-            file.write(text)
+                if self.add_timestamp:
+                    text = f"{dt_util.utcnow().isoformat()} {message}\n"
+                else:
+                    text = f"{message}\n"
+                file.write(text)
+        except Exception as exc:
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="write_access_failed",
+                translation_placeholders={"filename": filepath, "exc": f"{exc!r}"},
+            ) from exc