1
0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00

Delete old addon update backups when updating addon (#136977)

* Delete old addon update backups when updating addon

* Address review comments

* Add tests
This commit is contained in:
Erik Montnemery
2025-01-31 19:33:48 +01:00
committed by GitHub
parent 256157d413
commit 065cdf421f
5 changed files with 350 additions and 76 deletions

View File

@@ -685,6 +685,70 @@ class BackupManager:
return agent_errors
async def async_delete_filtered_backups(
self,
*,
include_filter: Callable[[dict[str, ManagerBackup]], dict[str, ManagerBackup]],
delete_filter: Callable[[dict[str, ManagerBackup]], dict[str, ManagerBackup]],
) -> None:
"""Delete backups parsed with a filter.
:param include_filter: A filter that should return the backups to consider for
deletion. Note: The newest of the backups returned by include_filter will
unconditionally be kept, even if delete_filter returns all backups.
:param delete_filter: A filter that should return the backups to delete.
"""
backups, get_agent_errors = await self.async_get_backups()
if get_agent_errors:
LOGGER.debug(
"Error getting backups; continuing anyway: %s",
get_agent_errors,
)
# Run the include filter first to ensure we only consider backups that
# should be included in the deletion process.
backups = include_filter(backups)
LOGGER.debug("Total automatic backups: %s", backups)
backups_to_delete = delete_filter(backups)
if not backups_to_delete:
return
# always delete oldest backup first
backups_to_delete = dict(
sorted(
backups_to_delete.items(),
key=lambda backup_item: backup_item[1].date,
)
)
if len(backups_to_delete) >= len(backups):
# Never delete the last backup.
last_backup = backups_to_delete.popitem()
LOGGER.debug("Keeping the last backup: %s", last_backup)
LOGGER.debug("Backups to delete: %s", backups_to_delete)
if not backups_to_delete:
return
backup_ids = list(backups_to_delete)
delete_results = await asyncio.gather(
*(self.async_delete_backup(backup_id) for backup_id in backups_to_delete)
)
agent_errors = {
backup_id: error
for backup_id, error in zip(backup_ids, delete_results, strict=True)
if error
}
if agent_errors:
LOGGER.error(
"Error deleting old copies: %s",
agent_errors,
)
async def async_receive_backup(
self,
*,