provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
album_types: list[AlbumType] | None = None,
- **kwargs: Any,
) -> list[Album]:
"""Get in-database albums.
:param extra_query: Additional SQL query string.
:param extra_query_params: Additional query parameters.
:param album_types: Filter by album types.
- :param library_items_only: If True, only return items that are
- marked as 'in_library' on any provider mapping.
"""
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
extra_join_parts=extra_join_parts,
- in_library_only=library_items_only,
)
# Calculate how many more items we need to reach the original limit
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
extra_join_parts=extra_join_parts,
- in_library_only=library_items_only,
):
# prevent duplicates (when artist is also in the title)
if album.uri not in existing_uris:
provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
album_artists_only: bool = False,
- **kwargs: Any,
) -> list[Artist]:
"""Get in-database (album) artists.
:param extra_query: Additional SQL query string.
:param extra_query_params: Additional query parameters.
:param album_artists_only: Only return artists that have albums.
- :param library_items_only: If True, only return items that are
- marked as 'in_library' on any provider mapping.
"""
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
async def tracks(
provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
- **kwargs: Any,
) -> list[Audiobook]:
"""Get in-database audiobooks.
: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 library_items_only: If True, only return items that are
- marked as 'in_library' on any provider mapping.
"""
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
if search and len(result) < 25 and not offset:
# append author items to result
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
return result
provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
- **kwargs: Any,
) -> list[ItemCls]:
- """
- Get in-database albums.
-
- :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).
- :param extra_query: Additional SQL query string.
- :param extra_query_params: Additional query parameters.
- :param library_items_only: If True, only return items that are
- marked as 'in_library' on any provider mapping.
- """
+ """Get in-database items."""
return await self._get_library_items_by_query(
favorite=favorite,
search=search,
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=[extra_query] if extra_query else None,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
async def iter_library_items(
provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
) -> AsyncGenerator[ItemCls, None]:
"""Iterate all in-database items."""
limit: int = 500
provider_filter=provider_filter,
extra_query_parts=[extra_query] if extra_query else None,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
for item in next_items:
yield item
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_parts=[extra_query], in_library_only=False
+ extra_query_parts=[extra_query],
):
return db_item
msg = f"{self.media_type.value} not found in library: {db_id}"
offset=offset,
extra_query_parts=[query],
extra_query_params=query_params,
- in_library_only=False,
)
@final
extra_query_parts: list[str] | None = None,
extra_query_params: dict[str, Any] | None = None,
extra_join_parts: list[str] | None = None,
- in_library_only: bool = False,
) -> list[ItemCls]:
"""Fetch MediaItem records from database by building the query."""
query_params = extra_query_params or {}
favorite=favorite,
search=search,
provider_filter=provider_filter,
- in_library_only=in_library_only,
limit=limit,
)
else:
favorite=favorite,
search=search,
provider_filter=provider_filter,
- in_library_only=in_library_only,
)
# build and execute final query
sql_query = self._build_final_query(query_parts, join_parts, order_by)
favorite: bool | None,
search: str | None,
provider_filter: list[str] | None,
- in_library_only: bool,
limit: int,
) -> None:
"""Build a fast random subquery with all filters applied."""
favorite=favorite,
search=search,
provider_filter=provider_filter,
- in_library_only=in_library_only,
)
# Build the subquery
favorite: bool | None,
search: str | None,
provider_filter: list[str] | None,
- in_library_only: bool,
) -> None:
"""Apply search, favorite, and provider filters."""
# handle search
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 = {in_library_only} "
+ f"AND provider_mappings.in_library = 1 "
f"AND ({' OR '.join(provider_conditions)})"
)
- elif in_library_only:
- 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 = {in_library_only} "
- )
@final
def _build_final_query(
provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
- **kwargs: Any,
) -> list[Podcast]:
"""Get in-database podcasts.
: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 library_items_only: If True, only return items that are
- marked as 'in_library' on any provider mapping.
"""
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
if search and len(result) < 25 and not offset:
# append publisher items to result
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=library_items_only,
)
return result
provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
- library_items_only: bool = True,
- **kwargs: Any,
) -> list[Track]:
"""Get in-database tracks.
: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 library_items_only: If True, only return items that are
- marked as 'in_library' on any provider mapping.
"""
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
extra_join_parts=extra_join_parts,
- in_library_only=library_items_only,
)
if search and len(result) < 25 and not offset:
# append artist items to result
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
extra_join_parts=extra_join_parts,
- in_library_only=library_items_only,
):
# prevent duplicates (when artist is also in the title)
if _track.uri not in existing_uris:
f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = {item_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], in_library_only=True
- )
+ return await self.mass.music.albums._get_library_items_by_query(extra_query_parts=[query])
async def match_provider(
self,
self.audiobooks,
self.podcasts,
):
- async for db_item in ctrl.iter_library_items(
- provider=list(multi_instance_providers), library_items_only=False
- ):
+ async for db_item in ctrl.iter_library_items(provider=list(multi_instance_providers)):
if self.match_provider_instances(db_item):
await ctrl.update_item_in_library(db_item.item_id, db_item)
# prevent overwhelming the event loop
projects / music-assistant-server.git / commitdiff