Use partial() for download_web_files_in_config call sites; collapse dedup loop into a dict comp

- Both audio_file and speaker.media_player now use functools.partial to
  bind path_for=_compute_local_file_path instead of an inline lambda.
- download_content_many's dedup loop becomes a dict comprehension.
  Last-URL wins now (dict-comp semantics) instead of first-URL; in
  practice duplicate paths only arise when the URL itself is duplicated,
  so the choice is meaningless. Test updated accordingly.
This commit is contained in:
J. Nick Koston
2026-04-26 10:02:44 -05:00
parent 2e037308de
commit 2def20c646
4 changed files with 14 additions and 11 deletions
+2 -1
View File
@@ -1,4 +1,5 @@
from dataclasses import dataclass, field
from functools import partial
import hashlib
import logging
from pathlib import Path
@@ -199,7 +200,7 @@ def _validate_supported_local_file(config: list[ConfigType]) -> list[ConfigType]
CONFIG_SCHEMA = cv.All(
cv.only_on_esp32,
cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA),
lambda c: download_web_files_in_config(c, _compute_local_file_path),
partial(download_web_files_in_config, path_for=_compute_local_file_path),
_validate_supported_local_file,
)
@@ -1,5 +1,6 @@
"""Speaker Media Player Setup."""
from functools import partial
import hashlib
import logging
from pathlib import Path
@@ -277,7 +278,9 @@ CONFIG_SCHEMA = cv.All(
cv.Optional(CONF_CODEC_SUPPORT_ENABLED): cv.Any(cv.boolean, cv.string),
cv.Optional(CONF_FILES): cv.All(
cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA),
lambda c: download_web_files_in_config(c, _compute_local_file_path),
partial(
download_web_files_in_config, path_for=_compute_local_file_path
),
),
cv.Optional(CONF_TASK_STACK_IN_PSRAM): cv.All(
cv.boolean, cv.requires_component(psram.DOMAIN)
+5 -6
View File
@@ -144,13 +144,12 @@ def download_content_many(
Items are de-duplicated by `path` -- two callers asking for the same
cache file (e.g. the same URL referenced twice in a config) would
otherwise race on `download_content`'s non-atomic write.
otherwise race on `download_content`'s non-atomic write. When the
same `path` appears more than once, the last URL wins (standard dict
comprehension semantics); in practice duplicate paths only arise when
the URL is duplicated, so the choice doesn't matter.
"""
seen: dict[Path, str] = {}
for url, path in items:
if path in seen:
continue
seen[path] = url
seen: dict[Path, str] = {path: url for url, path in items}
if not seen:
return
if len(seen) == 1:
+3 -3
View File
@@ -368,7 +368,9 @@ def test_download_content_many_dedupes_by_path(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""Two items pointing at the same cache path must collapse to one
download -- otherwise concurrent writes race on the same file.
download -- otherwise concurrent writes race on the same file. Which
URL wins doesn't matter (in practice duplicate paths only arise when
the URL is duplicated), so we only assert the call count and path.
"""
path = setup_core / "shared"
items = [
@@ -378,9 +380,7 @@ def test_download_content_many_dedupes_by_path(
]
external_files.download_content_many(items)
assert mock_download_content.call_count == 1
# First-seen URL wins, matching dict-insertion order.
args, _ = mock_download_content.call_args
assert args[0] == "https://example.com/a"
assert args[1] == path