offset: int = 0,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
album_types: list[AlbumType] | None = None,
) -> list[Album]:
"""Get in-database albums.
:param offset: Number of items to skip.
:param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
:param provider: Filter by provider instance ID (single string or list).
- :param extra_query: Additional SQL query string.
- :param extra_query_params: Additional query parameters.
:param album_types: Filter by album types.
"""
- extra_query_params = extra_query_params or {}
- extra_query_parts: list[str] = [extra_query] if extra_query else []
+ extra_query_params: dict[str, Any] = {}
+ extra_query_parts: list[str] = []
extra_join_parts: list[str] = []
artist_table_joined = False
# optional album type filter
)
artist_table_joined = True
extra_query_params["search_artist"] = f"%{artist_str}%"
- result = await self._get_library_items_by_query(
+ result = await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
extra_query_params["search_artist"] = f"%{search}%"
existing_uris = {item.uri for item in result}
- for album in await self._get_library_items_by_query(
+ for album in await self.get_library_items_by_query(
favorite=favorite,
search=None,
limit=remaining_limit,
item_id: str | int,
) -> list[Track]:
"""Return in-database album tracks for the given database album."""
- return await self.mass.music.tracks._get_library_items_by_query(
- extra_query_parts=[f"WHERE album_tracks.album_id = {item_id}"],
+ db_id = int(item_id) # ensure integer
+ return await self.mass.music.tracks.get_library_items_by_query(
+ extra_query_parts=["WHERE album_tracks.album_id = :album_id"],
+ extra_query_params={"album_id": db_id},
)
async def add_item_mapping_as_album_to_library(self, item: ItemMapping) -> Album:
offset: int = 0,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
album_artists_only: bool = False,
) -> list[Artist]:
"""Get in-database (album) artists.
:param offset: Number of items to skip.
:param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
:param provider: Filter by provider instance ID (single string or list).
- :param extra_query: Additional SQL query string.
- :param extra_query_params: Additional query parameters.
:param album_artists_only: Only return artists that have albums.
"""
- extra_query_params = extra_query_params or {}
- extra_query_parts: list[str] = [extra_query] if extra_query else []
+ extra_query_params: dict[str, Any] = {}
+ extra_query_parts: list[str] = []
if album_artists_only:
extra_query_parts.append(
f"artists.item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
f"from {DB_TABLE_ALBUM_ARTISTS})"
)
- return await self._get_library_items_by_query(
+ return await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
# recursively also remove artist albums
for db_row in await self.mass.music.database.get_rows_from_query(
- f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = {db_id}",
+ f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id",
+ {"artist_id": db_id},
limit=5000,
):
if not recursive:
await self.mass.music.albums.remove_item_from_library(db_row["album_id"])
# recursively also remove artist tracks
for db_row in await self.mass.music.database.get_rows_from_query(
- f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = {db_id}",
+ f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id",
+ {"artist_id": db_id},
limit=5000,
):
if not recursive:
item_id,
provider_instance_id_or_domain,
):
- artist_id = db_artist.item_id
- subquery = (
- f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = {artist_id}"
- )
+ db_artist_id = int(db_artist.item_id) # ensure integer
+ subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
query = f"tracks.item_id in ({subquery})"
- return await self.mass.music.tracks._get_library_items_by_query(
- extra_query_parts=[query], provider_filter=[provider_instance_id_or_domain]
+ return await self.mass.music.tracks.get_library_items_by_query(
+ extra_query_parts=[query],
+ extra_query_params={"artist_id": db_artist_id},
+ provider_filter=[provider_instance_id_or_domain],
)
return []
item_id: str | int,
) -> list[Track]:
"""Return all tracks for an artist in the library/db."""
- subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = {item_id}"
+ db_id = int(item_id) # ensure integer
+ subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
query = f"tracks.item_id in ({subquery})"
- return await self.mass.music.tracks.library_items(extra_query=query)
+ return await self.mass.music.tracks.get_library_items_by_query(
+ extra_query_parts=[query],
+ extra_query_params={"artist_id": db_id},
+ )
async def get_provider_artist_albums(
self,
item_id,
provider_instance_id_or_domain,
):
- artist_id = db_artist.item_id
- subquery = (
- f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = {artist_id}"
- )
+ db_artist_id = int(db_artist.item_id) # ensure integer
+ subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
query = f"albums.item_id in ({subquery})"
- return await self.mass.music.albums._get_library_items_by_query(
- extra_query_parts=[query], provider_filter=[provider_instance_id_or_domain]
+ return await self.mass.music.albums.get_library_items_by_query(
+ extra_query_parts=[query],
+ extra_query_params={"artist_id": db_artist_id},
+ provider_filter=[provider_instance_id_or_domain],
)
return []
item_id: str | int,
) -> list[Album]:
"""Return all in-library albums for an artist."""
- subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = {item_id}"
+ db_id = int(item_id) # ensure integer
+ subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
query = f"albums.item_id in ({subquery})"
- return await self.mass.music.albums.library_items(extra_query=query)
+ return await self.mass.music.albums.get_library_items_by_query(
+ extra_query_parts=[query],
+ extra_query_params={"artist_id": db_id},
+ )
async def _add_library_item(
self, item: Artist | ItemMapping, overwrite_existing: bool = False
offset: int = 0,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
) -> list[Audiobook]:
"""Get in-database audiobooks.
:param offset: Number of items to skip.
:param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
:param provider: Filter by provider instance ID (single string or list).
- :param extra_query: Additional SQL query string.
- :param extra_query_params: Additional query parameters.
"""
- extra_query_params = extra_query_params or {}
- extra_query_parts: list[str] = [extra_query] if extra_query else []
- result = await self._get_library_items_by_query(
+ extra_query_params: dict[str, Any] = {}
+ extra_query_parts: list[str] = []
+ result = await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
"WHERE audiobooks.authors LIKE :search or audiobooks.narrators LIKE :search",
]
extra_query_params["search"] = f"%{search}%"
- return result + await self._get_library_items_by_query(
+ return result + await self.get_library_items_by_query(
favorite=favorite,
search=None,
limit=limit,
# search by (exact) name match
query = f"{self.db_table}.name = :name OR {self.db_table}.sort_name = :sort_name"
query_params = {"name": item.name, "sort_name": item.sort_name}
- for db_item in await self._get_library_items_by_query(
+ for db_item in await self.get_library_items_by_query(
extra_query_parts=[query], extra_query_params=query_params
):
if compare_media_item(db_item, item, True):
offset: int = 0,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
) -> list[ItemCls]:
- """Get in-database items."""
- return await self._get_library_items_by_query(
+ """
+ Get the library items for this mediatype.
+
+ :param favorite: Filter by favorite status.
+ :param search: Filter by search query.
+ :param limit: Maximum number of items to return.
+ :param offset: Number of items to skip.
+ :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
+ :param provider: Filter by provider instance ID (single string or list).
+ """
+ return await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
offset=offset,
order_by=order_by,
provider_filter=self._ensure_provider_filter(provider),
- extra_query_parts=[extra_query] if extra_query else None,
- extra_query_params=extra_query_params,
)
async def iter_library_items(
search: str | None = None,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
) -> AsyncGenerator[ItemCls, None]:
"""Iterate all in-database items."""
limit: int = 500
else:
provider_filter = None
while True:
- next_items = await self._get_library_items_by_query(
+ next_items = await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
offset=offset,
order_by=order_by,
provider_filter=provider_filter,
- extra_query_parts=[extra_query] if extra_query else None,
- extra_query_params=extra_query_params,
)
for item in next_items:
yield item
async def get_library_item(self, item_id: int | str) -> ItemCls:
"""Get single library item by id."""
db_id = int(item_id) # ensure integer
- extra_query = f"WHERE {self.db_table}.item_id = {item_id}"
- for db_item in await self._get_library_items_by_query(
+ extra_query = f"WHERE {self.db_table}.item_id = :item_id"
+ for db_item in await self.get_library_items_by_query(
extra_query_parts=[extra_query],
+ extra_query_params={"item_id": db_id},
):
return db_item
msg = f"{self.media_type.value} not found in library: {db_id}"
external_id_str = f'%"{external_id_type}","{external_id}"%'
else:
external_id_str = f'%"{external_id}"%'
- for item in await self._get_library_items_by_query(
+ for item in await self.get_library_items_by_query(
extra_query_parts=[query],
extra_query_params={"external_id_str": external_id_str},
):
query_params["item_id"] = provider_item_id
subquery = f"SELECT item_id FROM provider_mappings WHERE {' AND '.join(subquery_parts)}"
query = f"WHERE {self.db_table}.item_id IN ({subquery})"
- return await self._get_library_items_by_query(
+ return await self.get_library_items_by_query(
limit=limit,
offset=offset,
extra_query_parts=[query],
"""
@final
- async def _get_library_items_by_query(
+ async def get_library_items_by_query(
self,
favorite: bool | None = None,
search: str | None = None,
# Apply the provider filter
if provider_filter:
provider_conditions = []
- for prov in provider_filter:
- provider_conditions.append(f"provider_mappings.provider_instance = '{prov}'")
+ for idx, prov in enumerate(provider_filter):
+ param_name = f"provider_filter_{idx}"
+ provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
+ query_params[param_name] = prov
+ query_params["provider_media_type"] = self.media_type.value
join_parts.append(
f"JOIN provider_mappings ON provider_mappings.item_id = {self.db_table}.item_id "
- f"AND provider_mappings.media_type = '{self.media_type.value}' "
- f"AND provider_mappings.in_library = 1 "
+ "AND provider_mappings.media_type = :provider_media_type "
+ "AND provider_mappings.in_library = 1 "
f"AND ({' OR '.join(provider_conditions)})"
)
offset: int = 0,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
) -> list[Podcast]:
"""Get in-database podcasts.
:param offset: Number of items to skip.
:param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
:param provider: Filter by provider instance ID (single string or list).
- :param extra_query: Additional SQL query string.
- :param extra_query_params: Additional query parameters.
"""
- extra_query_params = extra_query_params or {}
- extra_query_parts: list[str] = [extra_query] if extra_query else []
- result = await self._get_library_items_by_query(
+ result = await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
offset=offset,
order_by=order_by,
provider_filter=self._ensure_provider_filter(provider),
- extra_query_parts=extra_query_parts,
- extra_query_params=extra_query_params,
)
if search and len(result) < 25 and not offset:
# append publisher items to result
- extra_query_parts = [
+ extra_query_parts: list[str] = [
"WHERE podcasts.publisher LIKE :search",
]
- extra_query_params["search"] = f"%{search}%"
- return result + await self._get_library_items_by_query(
+ extra_query_params: dict[str, Any] = {
+ "search": f"%{search}%",
+ }
+ return result + await self.get_library_items_by_query(
favorite=favorite,
search=None,
limit=limit,
offset: int = 0,
order_by: str = "sort_name",
provider: str | list[str] | None = None,
- extra_query: str | None = None,
- extra_query_params: dict[str, Any] | None = None,
) -> list[Track]:
"""Get in-database tracks.
:param offset: Number of items to skip.
:param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
:param provider: Filter by provider instance ID (single string or list).
- :param extra_query: Additional SQL query string.
- :param extra_query_params: Additional query parameters.
"""
- extra_query_params = extra_query_params or {}
- extra_query_parts: list[str] = [extra_query] if extra_query else []
+ extra_query_params: dict[str, Any] = {}
+ extra_query_parts: list[str] = []
extra_join_parts: list[str] = []
if search and " - " in search:
# handle combined artist + title search
"AND artists.search_name LIKE :search_artist"
)
extra_query_params["search_artist"] = f"%{artist_str}%"
- result = await self._get_library_items_by_query(
+ result = await self.get_library_items_by_query(
favorite=favorite,
search=search,
limit=limit,
)
extra_query_params["search_artist"] = f"%{artist_search_str}%"
existing_uris = {item.uri for item in result}
- for _track in await self._get_library_items_by_query(
+ for _track in await self.get_library_items_by_query(
favorite=favorite,
search=None,
limit=limit,
item_id: str | int,
) -> list[Album]:
"""Return all in-library albums for a track."""
+ db_id = int(item_id) # ensure integer
subquery = (
f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} "
- f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = {item_id}"
+ f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id"
)
query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
- return await self.mass.music.albums._get_library_items_by_query(extra_query_parts=[query])
+ return await self.mass.music.albums.get_library_items_by_query(
+ extra_query_parts=[query],
+ extra_query_params={"track_id": db_id},
+ )
async def match_provider(
self,
f"AND (json_extract({DB_TABLE_ARTISTS}.metadata,'$.images') ISNULL "
f"OR json_extract({DB_TABLE_ARTISTS}.metadata,'$.images') = '[]')"
)
- for artist in await self.mass.music.artists.library_items(
- limit=5, order_by="random", extra_query=query
+ for artist in await self.mass.music.artists.get_library_items_by_query(
+ limit=5, order_by="random", extra_query_parts=[query]
):
if artist.uri:
self.schedule_update_metadata(artist.uri)
f"json_extract({DB_TABLE_PLAYLISTS}.metadata,'$.last_refresh') ISNULL "
f"OR json_extract({DB_TABLE_PLAYLISTS}.metadata,'$.last_refresh') < {timestamp}"
)
- for playlist in await self.mass.music.playlists.library_items(
- limit=5, order_by="random", extra_query=query
+ for playlist in await self.mass.music.playlists.get_library_items_by_query(
+ limit=5, order_by="random", extra_query_parts=[query]
):
if playlist.uri:
self.schedule_update_metadata(playlist.uri)
@use_cache(expiration=3600, category=CACHE_CATEGORY_PLAYLISTS)
async def _get_builtin_playlist_random_album(self) -> list[Track]:
- for random_album in await self.mass.music.albums.library_items(
- limit=1, order_by="random", extra_query="album_type != 'single'"
+ for random_album in await self.mass.music.albums.get_library_items_by_query(
+ limit=1,
+ order_by="random",
+ extra_query_parts=["album_type != :excluded_album_type"],
+ extra_query_params={"excluded_album_type": "single"},
):
tracks = await self.mass.music.albums.tracks(
random_album.item_id, random_album.provider
# searching the filesystem is slow and unreliable,
# so instead we just query the db...
if media_types is None or MediaType.TRACK in media_types:
- result.tracks = await self.mass.music.tracks._get_library_items_by_query(
+ result.tracks = await self.mass.music.tracks.get_library_items_by_query(
search=search_query, provider_filter=[self.instance_id], limit=limit
)
if media_types is None or MediaType.ALBUM in media_types:
- result.albums = await self.mass.music.albums._get_library_items_by_query(
+ result.albums = await self.mass.music.albums.get_library_items_by_query(
search=search_query,
provider_filter=[self.instance_id],
limit=limit,
)
if media_types is None or MediaType.ARTIST in media_types:
- result.artists = await self.mass.music.artists._get_library_items_by_query(
+ result.artists = await self.mass.music.artists.get_library_items_by_query(
search=search_query,
provider_filter=[self.instance_id],
limit=limit,
)
if media_types is None or MediaType.PLAYLIST in media_types:
- result.playlists = await self.mass.music.playlists._get_library_items_by_query(
+ result.playlists = await self.mass.music.playlists.get_library_items_by_query(
search=search_query,
provider_filter=[self.instance_id],
limit=limit,
)
if media_types is None or MediaType.AUDIOBOOK in media_types:
- result.audiobooks = await self.mass.music.audiobooks._get_library_items_by_query(
+ result.audiobooks = await self.mass.music.audiobooks.get_library_items_by_query(
search=search_query,
provider_filter=[self.instance_id],
limit=limit,
)
if media_types is None or MediaType.PODCAST in media_types:
- result.podcasts = await self.mass.music.podcasts._get_library_items_by_query(
+ result.podcasts = await self.mass.music.podcasts.get_library_items_by_query(
search=search_query,
provider_filter=[self.instance_id],
limit=limit,
)
async def _get_or_create_artist_by_name(self, artist_name: str) -> Artist | ItemMapping:
- if library_items := await self.mass.music.artists._get_library_items_by_query(
+ if library_items := await self.mass.music.artists.get_library_items_by_query(
search=artist_name, provider_filter=[self.instance_id]
):
return ItemMapping.from_item(library_items[0])
projects / music-assistant-server.git / commitdiff