feat: add support for new backup logic in HA Core 2025.1.0

custom_components/auto_backup/__init__.py

@@ -37,7 +37,6 @@
     CONF_AUTO_PURGE,
     CONF_BACKUP_TIMEOUT,
     DEFAULT_BACKUP_TIMEOUT,
-    PATCH_NAME,
 )
 from .handlers import SupervisorHandler, HassioAPIError, BackupHandler, HandlerBase
 
@@ -156,7 +155,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
     if is_hassio(hass):
         handler = SupervisorHandler(getenv("SUPERVISOR"), async_get_clientsession(hass))
     else:
-        handler = BackupHandler(hass.data[DOMAIN_BACKUP])
+        handler = BackupHandler(hass, hass.data[DOMAIN_BACKUP])
 
     auto_backup = AutoBackup(hass, options, handler)
     hass.data.setdefault(DOMAIN, {})[DATA_AUTO_BACKUP] = auto_backup
@@ -296,13 +295,6 @@ def generate_backup_name(self) -> str:
     def validate_backup_config(self, config: Dict):
         """Validate the backup config."""
         if not self._supervised:
-            disallowed_options = [ATTR_PASSWORD]
-            for option in disallowed_options:
-                if config.get(option):
-                    raise HomeAssistantError(
-                        f"The '{option}' option is not supported on non-supervised installations."
-                    )
-
             # allow `include` if it only contains the configuration
             if ATTR_INCLUDE in config and not config.get(ATTR_EXCLUDE):
                 # ensure no addons were included
@@ -317,9 +309,6 @@ def validate_backup_config(self, config: Dict):
                     "Partial backups (e.g. include/exclude) are not supported on non-supervised installations."
                 )
 
-            if config.get(ATTR_NAME):
-                config[PATCH_NAME] = True
-
         if not config.get(ATTR_NAME):
             config[ATTR_NAME] = self.generate_backup_name()
 

custom_components/auto_backup/const.py

@@ -15,6 +15,4 @@
 EVENT_BACKUP_FAILED = f"{DOMAIN}.backup_failed"
 EVENT_BACKUPS_PURGED = f"{DOMAIN}.purged_backups"
 
-PATCH_NAME = "patch.name"
-
 PLATFORMS = [Platform.SENSOR]

custom_components/auto_backup/handlers.py

@@ -9,10 +9,11 @@
 import aiohttp
 from aiohttp.hdrs import AUTHORIZATION
 from homeassistant.components.backup import BackupManager
+from homeassistant.components.hassio import ATTR_PASSWORD
 from homeassistant.const import ATTR_NAME
-from homeassistant.exceptions import HomeAssistantError
+from homeassistant.core import HomeAssistant
 
-from .const import DEFAULT_BACKUP_TIMEOUT_SECONDS, PATCH_NAME
+from .const import DEFAULT_BACKUP_TIMEOUT_SECONDS
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -168,7 +169,8 @@ async def download_backup(
 
 
 class BackupHandler(HandlerBase):
-    def __init__(self, manager: BackupManager):
+    def __init__(self, hass: HomeAssistant, manager: BackupManager):
+        self._hass = hass
         self._manager = manager
 
     async def get_addons(self):
@@ -178,69 +180,40 @@ async def get_addons(self):
     async def create_backup(
         self, config: Dict, partial: bool = False, timeout: Optional[int] = None
     ) -> Dict:
-        if not config.get(PATCH_NAME):
-            if hasattr(self._manager, "async_create_backup"):
-                backup = await self._manager.async_create_backup()
-            else:
-                backup = await self._manager.generate_backup()
-        else:
-            _LOGGER.debug("Support name is set: %s", config)
-            if not hasattr(self._manager, "_mkdir_and_generate_backup_contents"):
-                if not hasattr(self._manager, "_generate_backup_contents"):
-                    raise HomeAssistantError(
-                        "Unable to patch '_generate_backup_contents' function."
-                    )
-                raise HomeAssistantError(
-                    "Unable to patch '_mkdir_and_generate_backup_contents' function."
-                )
+        agent_id = list(self._manager.local_backup_agents)[0]
+        backup = await self._manager.async_create_backup(
+            agent_ids=[agent_id],
+            name=config.get(ATTR_NAME),
+            include_database=True,
+            include_folders=None,
+            include_homeassistant=True,
+            password=config.get(ATTR_PASSWORD),
+            # don't exist on HA Core
+            include_all_addons=False,
+            include_addons=None,
+        )
+        [backup, agent_errors] = await self._manager.async_get_backup(
+            backup.backup_job_id
+        )
 
-            def wrapper(*args, **kwargs):
-                if len(args) != 2 or kwargs or not isinstance(args[1], dict):
-                    raise HomeAssistantError(
-                        "Wrapper of '_mkdir_and_generate_backup_contents' called with wrong arguments"
-                    )
-
-                args[1]["name"] = config[ATTR_NAME]
-                return old_function(*args, **kwargs)
-
-            if hasattr(self._manager, "_generate_backup_contents"):
-                old_function = self._manager._generate_backup_contents
-            else:
-                old_function = self._manager._mkdir_and_generate_backup_contents
-
-            try:
-                if hasattr(self._manager, "_generate_backup_contents"):
-                    self._manager._generate_backup_contents = wrapper
-                else:
-                    self._manager._mkdir_and_generate_backup_contents = wrapper
-                if hasattr(self._manager, "async_create_backup"):
-                    backup = await self._manager.async_create_backup()
-                else:
-                    backup = await self._manager.generate_backup()
-                backup.name = config[ATTR_NAME]
-            finally:
-                if hasattr(self._manager, "_generate_backup_contents"):
-                    self._manager._generate_backup_contents = old_function
-                else:
-                    self._manager._mkdir_and_generate_backup_contents = old_function
-
-        return backup.as_dict()
+        return {"slug": backup.backup_id, **backup.as_dict()}
 
     async def remove_backup(self, slug):
-        if hasattr(self._manager, "async_remove_backup"):
-            await self._manager.async_remove_backup(slug=slug)
-        else:
-            await self._manager.remove_backup(slug)
+        await self._manager.async_delete_backup(slug)
 
     async def download_backup(
         self, slug: str, destination: str, timeout: int = DEFAULT_BACKUP_TIMEOUT_SECONDS
     ):
-        if hasattr(self._manager, "async_get_backup"):
-            backup = await self._manager.async_get_backup(slug=slug)
-        else:
-            backup = await self._manager.get_backup(slug)
+        [backup, agent_errors] = await self._manager.async_get_backup(slug)
         if backup:
-            shutil.copyfile(backup.path, destination)
+            agent_id = list(self._manager.local_backup_agents)[0]
+            agent = self._manager.local_backup_agents[agent_id]
+            backup_path = agent.get_backup_path(backup.backup_id)
+
+            def _copyfile():
+                shutil.copyfile(backup_path, destination)
+
+            await self._hass.async_add_executor_job(_copyfile)
         else:
             _LOGGER.error(
                 "Cannot move backup (%s) to '%s' as it does not exist.",

hacs.json

@@ -3,5 +3,5 @@
   "zip_release": true,
   "hide_default_branch": true,
   "filename": "auto_backup.zip",
-  "homeassistant": "2024.11.0"
+  "homeassistant": "2025.1.0"
 }