ui: remove do_query to tighten input_select_objects

beets/ui/__init__.py

@@ -399,7 +399,7 @@ def input_select_objects(
     objs: Sequence[T],
     rep: Callable[[T], Any],
     prompt_all: str | None = None,
-) -> Any:
+) -> Sequence[T]:
     """Prompt to user to choose all, none, or some of the given objects.
     Return the list of selected objects.
 

beets/ui/commands/modify.py

@@ -4,17 +4,16 @@
 
 from typing import TYPE_CHECKING, NamedTuple, Protocol
 
-from beets import library, ui
+from beets import ui
 from beets.dbcore import types
 from beets.exceptions import UserError
+from beets.library import Album, Item
 from beets.util.deprecation import maybe_replace_legacy_field
 
-from .utils import do_query
-
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-    from beets.library import LibModel, Library
+    from beets.library import AlbumOrItem, LibModel, Library
 
 
 class ModifyCLIOpts(Protocol):
@@ -59,34 +58,28 @@ def _check_modify_operations(
             )
 
 
-def modify_items(
+def modify_objects(
+    model_cls: type[AlbumOrItem],
+    objs: Sequence[AlbumOrItem],
     lib: Library,
     mods: dict[str, ModifyOperation],
     dels: Sequence[str],
-    query: Sequence[str],
     write: bool,
     move: bool,
-    album: bool,
     confirm: bool,
     inherit: bool,
 ) -> None:
-    """Modifies matching items according to user-specified assignments and
+    """Modifies albums or items according to user-specified assignments and
     deletions.
 
-    `mods` is a dictionary of field and value pairse indicating
+    `mods` is a dictionary of field and value pairs indicating
     assignments. `dels` is a list of fields to be deleted.
     """
     # Parse key=value specifications into a dictionary.
-    model_cls = library.Album if album else library.Item
     _check_modify_operations(model_cls, mods)
-
-    # Get the items to modify.
-    items, albums = do_query(lib, query, album, False)
-    objs = albums if album else items
-
     # Apply changes *temporarily*, preview them, and collect modified
     # objects.
-    ui.print_(f"Modifying {len(objs)} {'album' if album else 'item'}s.")
+    ui.print_(f"Modifying {len(objs)} {model_cls.__name__.lower()}s.")
     changed = []
     for obj in objs:
         obj_mods = {
@@ -106,6 +99,7 @@ def modify_items(
         return
 
     # Confirm action.
+    selected_changes: Sequence[AlbumOrItem]
     if confirm:
         if write and move:
             extra = ", move and write tags"
@@ -116,22 +110,28 @@ def modify_items(
         else:
             extra = ""
 
-        changed = ui.input_select_objects(
-            f"Really modify{extra}",
-            changed,
-            lambda o: print_and_modify(o, mods, dels),
+        selected_changes = ui.input_select_objects(
+            f"Really modify{extra}", changed, ui.show_model_changes
         )
+    else:
+        selected_changes = changed
 
     # Apply changes to database and files
     with lib.transaction():
-        for obj in changed:
+        for obj in selected_changes:
             obj.try_sync(write, move, inherit)
 
 
+def modify_items(lib: Library, query: Sequence[str], *args, **kwargs) -> None:
+    modify_objects(Item, list(lib.items(query)), lib, *args, **kwargs)
+
+
+def modify_albums(lib: Library, query: Sequence[str], *args, **kwargs) -> None:
+    modify_objects(Album, list(lib.albums(query)), lib, *args, **kwargs)
+
+
 def print_and_modify(
-    obj: LibModel,
-    mods: dict[str, list[str]] | dict[str, ModifyOperation],
-    dels: Sequence[str],
+    obj: LibModel, mods: dict[str, list[str]], dels: Sequence[str]
 ) -> bool:
     """Print the modifications to an item and return a bool indicating
     whether any changes were made.
@@ -180,14 +180,14 @@ def modify_func(lib: Library, opts: ModifyCLIOpts, args: list[str]) -> None:
     query, mods, dels = modify_parse_args(args, is_album=opts.album)
     if not mods and not dels:
         raise UserError("no modifications specified")
-    modify_items(
+    method = modify_albums if opts.album else modify_items
+    method(
         lib,
+        query,
         mods,
         dels,
-        query,
         ui.should_write(opts.write),
         ui.should_move(opts.move),
-        opts.album,
         not opts.yes,
         opts.inherit,
     )

beets/ui/commands/move.py

@@ -3,21 +3,17 @@
 from __future__ import annotations
 
 import os
-from typing import TYPE_CHECKING, Protocol
-
-from typing_extensions import TypeIs
+from typing import TYPE_CHECKING, Literal, Protocol
 
 from beets import logging, ui
 from beets.exceptions import UserError
 from beets.util import MoveOperation, displayable_path, normpath, syspath
 from beets.util.diff import colordiff
 
-from .utils import do_query
-
 if TYPE_CHECKING:
-    from collections.abc import Iterable, Sequence
+    from collections.abc import Callable, Iterable, Sequence
 
-    from beets.library import Album, Item, Library
+    from beets.library import Album, AlbumOrItem, Item, Library
 
 # Global logger.
 log = logging.getLogger("beets")
@@ -73,42 +69,22 @@ def show_path_changes(path_changes: Iterable[tuple[bytes, bytes]]) -> None:
             ui.print_(f"{color_source} {' ' * pad} -> {color_dest}")
 
 
-def is_album_selection(
-    objects: list[Item] | list[Album], album: bool
-) -> TypeIs[list[Album]]:
-    return album
-
-
-def move_items(
-    lib: Library,
+def move_objects(
+    objs: Sequence[AlbumOrItem],
+    entity: Literal["album", "item"],
     dest: bytes | None,
-    query: Sequence[str],
+    get_paths: Callable[[Iterable[AlbumOrItem]], list[tuple[bytes, bytes]]],
+    *,
     copy: bool,
-    album: bool,
     pretend: bool,
     confirm: bool = False,
     export: bool = False,
 ) -> None:
-    """Moves or copies items to a new base directory, given by dest. If
+    """Move or copy albums or items to a new base directory, given by dest. If
     dest is None, then the library's base directory is used, making the
     command "consolidate" files.
     """
-    items, albums = do_query(lib, query, album, False)
-    objs = albums if album else items
     num_objs = len(objs)
-
-    # Filter out files that don't need to be moved.
-    def isitemmoved(item: Item) -> bool:
-        return item.path != item.destination(basedir=dest)
-
-    def isalbummoved(album: Album) -> bool:
-        return any(isitemmoved(i) for i in album.items())
-
-    if is_album_selection(objs, album):
-        objs = list(filter(isalbummoved, objs))
-    else:
-        objs = list(filter(isitemmoved, objs))
-
     num_unmoved = num_objs - len(objs)
     # Report unmoved files that match the query.
     unmoved_msg = ""
@@ -118,7 +94,6 @@ def isalbummoved(album: Album) -> bool:
     copy = copy or export  # Exporting always copies.
     action = "Copying" if copy else "Moving"
     act = "copy" if copy else "move"
-    entity = "album" if album else "item"
     log.info(
         "{} {} {}{}{}.",
         action,
@@ -131,29 +106,21 @@ def isalbummoved(album: Album) -> bool:
         return
 
     if pretend:
-        if is_album_selection(objs, album):
-            show_path_changes(
-                [
-                    (item.path, item.destination(basedir=dest))
-                    for obj in objs
-                    for item in obj.items()
-                ]
-            )
-        else:
-            show_path_changes(
-                [(obj.path, obj.destination(basedir=dest)) for obj in objs]
-            )
+        show_path_changes(get_paths(objs))
     else:
+        selected_objs: Sequence[AlbumOrItem]
         if confirm:
-            objs = ui.input_select_objects(
+            selected_objs = ui.input_select_objects(
                 f"Really {act}",
                 objs,
                 lambda o: show_path_changes(
                     [(o.path, o.destination(basedir=dest))]
                 ),
             )
+        else:
+            selected_objs = objs
 
-        for obj in objs:
+        for obj in selected_objs:
             log.debug("moving: {.filepath}", obj)
 
             if export:
@@ -169,21 +136,66 @@ def isalbummoved(album: Album) -> bool:
                     obj.move(operation=MoveOperation.MOVE, basedir=dest)
 
 
+def isitemmoved(dest: bytes | None, item: Item) -> bool:
+    """Filter out files that don't need to be moved."""
+    return item.path != item.destination(basedir=dest)
+
+
+def isalbummoved(dest: bytes | None, album: Album) -> bool:
+    return any(isitemmoved(dest, i) for i in album.items())
+
+
+def move_items(
+    lib: Library, query: Sequence[str], dest: bytes | None, *args, **kwargs
+) -> None:
+    def get_paths(objs: Iterable[Item]) -> list[tuple[bytes, bytes]]:
+        return [(obj.path, obj.destination(basedir=dest)) for obj in objs]
+
+    move_objects(
+        [i for i in lib.items(query) if isitemmoved(dest, i)],
+        "item",
+        dest,
+        get_paths,
+        *args,
+        **kwargs,
+    )
+
+
+def move_albums(
+    lib: Library, query: Sequence[str], dest: bytes | None, *args, **kwargs
+) -> None:
+    def get_paths(objs: Iterable[Album]) -> list[tuple[bytes, bytes]]:
+        return [
+            (item.path, item.destination(basedir=dest))
+            for obj in objs
+            for item in obj.items()
+        ]
+
+    move_objects(
+        [i for i in lib.albums(query) if isalbummoved(dest, i)],
+        "album",
+        dest,
+        get_paths,
+        *args,
+        **kwargs,
+    )
+
+
 def move_func(lib: Library, opts: MoveCLIOpts, args: list[str]) -> None:
     dest = normpath(opts.dest) if opts.dest else None
     if dest is not None:
         if not os.path.isdir(syspath(dest)):
             raise UserError(f"no such directory: {displayable_path(dest)}")
 
-    move_items(
+    method = move_albums if opts.album else move_items
+    method(
         lib,
-        dest,
         args,
-        opts.copy,
-        opts.album,
-        opts.pretend,
-        opts.timid,
-        opts.export,
+        dest,
+        copy=opts.copy,
+        pretend=opts.pretend,
+        confirm=opts.timid,
+        export=opts.export,
     )