Restore repository.
(self, entry: str, repository_data: dict[str, Any])
| 257 | |
| 258 | @callback |
| 259 | def async_restore_repository(self, entry: str, repository_data: dict[str, Any]): |
| 260 | """Restore repository.""" |
| 261 | repository: HacsRepository | None = None |
| 262 | if full_name := repository_data.get("full_name"): |
| 263 | repository = self.hacs.repositories.get_by_full_name(full_name) |
| 264 | if not repository: |
| 265 | repository = self.hacs.repositories.get_by_id(entry) |
| 266 | if not repository: |
| 267 | return |
| 268 | |
| 269 | try: |
| 270 | self.hacs.repositories.set_repository_id(repository, entry) |
| 271 | except ValueError as exception: |
| 272 | self.logger.warning("<HacsData async_restore_repository> duplicate IDs %s", exception) |
| 273 | return |
| 274 | |
| 275 | # Restore repository attributes |
| 276 | repository.data.authors = repository_data.get("authors", []) |
| 277 | repository.data.description = repository_data.get("description", "") |
| 278 | repository.data.downloads = repository_data.get("downloads", 0) |
| 279 | repository.data.last_updated = repository_data.get("last_updated", 0) |
| 280 | if self.hacs.system.generator: |
| 281 | repository.data.etag_releases = repository_data.get("etag_releases") |
| 282 | repository.data.open_issues = repository_data.get("open_issues", 0) |
| 283 | repository.data.etag_repository = repository_data.get("etag_repository") |
| 284 | repository.data.topics = [ |
| 285 | topic for topic in repository_data.get("topics", []) if topic not in TOPIC_FILTER |
| 286 | ] |
| 287 | repository.data.domain = repository_data.get("domain") |
| 288 | repository.data.stargazers_count = repository_data.get( |
| 289 | "stargazers_count" |
| 290 | ) or repository_data.get("stars", 0) |
| 291 | repository.releases.last_release = repository_data.get("last_release_tag") |
| 292 | repository.data.releases = repository_data.get("releases", False) |
| 293 | repository.data.installed = repository_data.get("installed", False) |
| 294 | repository.data.new = repository_data.get("new", False) |
| 295 | repository.data.selected_tag = repository_data.get("selected_tag") |
| 296 | repository.data.show_beta = repository_data.get("show_beta", False) |
| 297 | repository.data.last_version = repository_data.get("last_version") |
| 298 | repository.data.prerelease = repository_data.get("prerelease") |
| 299 | repository.data.last_commit = repository_data.get("last_commit") |
| 300 | repository.data.installed_version = repository_data.get("version_installed") |
| 301 | repository.data.installed_commit = repository_data.get("installed_commit") |
| 302 | repository.data.manifest_name = repository_data.get("manifest_name") |
| 303 | |
| 304 | if last_fetched := repository_data.get("last_fetched"): |
| 305 | repository.data.last_fetched = datetime.fromtimestamp(last_fetched, UTC) |
| 306 | |
| 307 | repository.repository_manifest = HacsManifest.from_dict( |
| 308 | repository_data.get("manifest") or repository_data.get("repository_manifest") or {} |
| 309 | ) |
| 310 | |
| 311 | if repository.data.prerelease == repository.data.last_version: |
| 312 | repository.data.prerelease = None |
| 313 | |
| 314 | if repository.localpath is not None and is_safe(self.hacs, repository.localpath): |
| 315 | # Set local path |
| 316 | repository.content.path.local = repository.localpath |
no test coverage detected