extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
extra_join_parts=extra_join_parts,
- in_library_only=True,
)
# Calculate how many more items we need to reach the original limit
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=True,
)
async def tracks(
provider_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=True,
)
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=True,
)
return result
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=True,
)
async def iter_library_items(
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_filter=self._ensure_provider_filter(provider),
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
- in_library_only=True,
)
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=True,
)
return result
extra_query_parts=extra_query_parts,
extra_query_params=extra_query_params,
extra_join_parts=extra_join_parts,
- in_library_only=True,
)
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=True,
):
# prevent duplicates (when artist is also in the title)
if _track.uri not in existing_uris:
await self._database.execute(
f"UPDATE {DB_TABLE_PROVIDER_MAPPINGS} SET in_library = 1 "
"WHERE provider_domain IN "
- "('filesystem_local', 'filesystem_smb', 'plex', 'jellyfin', 'opensubsonic', 'builtin');" # noqa: E501
+ "('filesystem_local', 'filesystem_smb', 'plex', "
+ "'jellyfin', 'opensubsonic', 'builtin');"
)
# save changes