Merge pull request #541 from k4black/feat-async-migration

refactor: convert backend to async (engine, repos, services, routers, httpx)
This commit is contained in:
Maximilian Dorninger
2026-07-05 19:02:15 +02:00
committed by GitHub
37 changed files with 947 additions and 820 deletions

View File

@@ -7,11 +7,10 @@ from fastapi_users.db import (
SQLAlchemyUserDatabase,
)
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column, relationship
from media_manager.config import MediaManagerConfig
from media_manager.database import Base, build_db_url
from media_manager.database import Base, get_async_session
class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
@@ -27,17 +26,6 @@ class User(SQLAlchemyBaseUserTableUUID, Base):
)
engine = create_async_engine(
build_db_url(**MediaManagerConfig().database.model_dump()), echo=False
)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
async def get_async_session() -> AsyncGenerator[AsyncSession]:
async with async_session_maker() as session:
yield session
async def get_user_db(
session: AsyncSession = Depends(get_async_session),
) -> AsyncGenerator[SQLAlchemyUserDatabase]:

View File

@@ -67,9 +67,9 @@ openid_config = MediaManagerConfig().auth.openid_connect
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_superuser)],
)
def get_all_users(db: DbSessionDependency) -> list[UserRead]:
async def get_all_users(db: DbSessionDependency) -> list[UserRead]:
stmt = select(User)
result = db.execute(stmt).scalars().unique()
result = (await db.execute(stmt)).scalars().unique()
return [UserRead.model_validate(user) for user in result]

View File

@@ -4,7 +4,7 @@ from uuid import UUID
from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from media_manager.exceptions import ConflictError, NotFoundError
@@ -20,51 +20,51 @@ class BaseRepository[T, S]:
Base repository providing common CRUD operations for media models.
"""
def __init__(self, db: Session, model: type[T], schema: type[S]) -> None:
def __init__(self, db: AsyncSession, model: type[T], schema: type[S]) -> None:
self.db = db
self.model = model
self.schema = schema
def get_by_id(self, entity_id: EntityId) -> S:
result = self.db.get(self.model, entity_id)
async def get_by_id(self, entity_id: EntityId) -> S:
result = await self.db.get(self.model, entity_id)
if not result:
msg = f"{self.model.__name__} with id {entity_id} not found."
raise NotFoundError(msg)
return self.schema.model_validate(result)
def get_by_external_id(self, external_id: int, metadata_provider: str) -> S:
async def get_by_external_id(self, external_id: int, metadata_provider: str) -> S:
stmt = select(self.model).where(
self.model.external_id == external_id,
self.model.metadata_provider == metadata_provider,
)
result = self.db.execute(stmt).scalar_one_or_none()
result = (await self.db.execute(stmt)).scalar_one_or_none()
if not result:
msg = f"{self.model.__name__} with external_id {external_id} and provider {metadata_provider} not found."
raise NotFoundError(msg)
return self.schema.model_validate(result)
def get_all(self) -> list[S]:
async def get_all(self) -> list[S]:
stmt = select(self.model)
results = self.db.execute(stmt).scalars().unique().all()
results = (await self.db.execute(stmt)).scalars().unique().all()
return [self.schema.model_validate(r) for r in results]
def delete(self, entity_id: EntityId) -> None:
obj = self.db.get(self.model, entity_id)
async def delete(self, entity_id: EntityId) -> None:
obj = await self.db.get(self.model, entity_id)
if not obj:
msg = f"{self.model.__name__} with id {entity_id} not found."
raise NotFoundError(msg)
self.db.delete(obj)
self.db.commit()
await self.db.delete(obj)
await self.db.commit()
def set_library(self, entity_id: EntityId, library: str) -> None:
obj = self.db.get(self.model, entity_id)
async def set_library(self, entity_id: EntityId, library: str) -> None:
obj = await self.db.get(self.model, entity_id)
if not obj:
msg = f"{self.model.__name__} with id {entity_id} not found."
raise NotFoundError(msg)
obj.library = library
self.db.commit()
await self.db.commit()
def save_media_base(
async def save_media_base(
self,
media_schema: S,
model_class: type[T],
@@ -76,35 +76,34 @@ class BaseRepository[T, S]:
if exclude is None:
exclude = set()
db_obj = self.db.get(model_class, media_schema.id) if media_schema.id else None
db_obj = (
await self.db.get(model_class, media_schema.id) if media_schema.id else None
)
if db_obj:
# Update existing
# Always exclude "id" from updates
update_exclude = exclude | {"id"}
for key, value in media_schema.model_dump(exclude=update_exclude).items():
if hasattr(db_obj, key):
setattr(db_obj, key, value)
else:
# Insert new
db_obj = model_class(**media_schema.model_dump(exclude=exclude))
self.db.add(db_obj)
try:
self.db.commit()
self.db.refresh(db_obj)
await self.db.commit()
await self.db.refresh(db_obj)
except IntegrityError as e:
self.db.rollback()
await self.db.rollback()
msg = f"Integrity error while saving {model_class.__name__}: {e.orig}"
raise ConflictError(msg) from e
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
log.exception(f"Database error while saving {model_class.__name__}")
raise
else:
return self.schema.model_validate(db_obj)
def update_media_attributes_base(
async def update_media_attributes_base(
self,
media_id: EntityId,
model_class: type[T],
@@ -113,7 +112,7 @@ class BaseRepository[T, S]:
"""
Generic update method for media attributes.
"""
db_obj = self.db.get(model_class, media_id)
db_obj = await self.db.get(model_class, media_id)
if not db_obj:
msg = f"{model_class.__name__} with id {media_id} not found."
raise NotFoundError(msg)
@@ -130,15 +129,15 @@ class BaseRepository[T, S]:
if updated:
try:
self.db.commit()
self.db.refresh(db_obj)
await self.db.commit()
await self.db.refresh(db_obj)
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
return self.schema.model_validate(db_obj)
def add_media_file_base(
async def add_media_file_base(
self, file_schema: S, model_class: type[T], schema_class: type[S]
) -> S:
"""
@@ -147,18 +146,18 @@ class BaseRepository[T, S]:
db_model = model_class(**file_schema.model_dump())
try:
self.db.add(db_model)
self.db.commit()
self.db.refresh(db_model)
await self.db.commit()
await self.db.refresh(db_model)
except IntegrityError:
self.db.rollback()
await self.db.rollback()
raise
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
else:
return schema_class.model_validate(db_model)
def remove_files_by_torrent_id_base(
async def remove_files_by_torrent_id_base(
self, torrent_id: EntityId, model_class: type[T]
) -> int:
"""
@@ -166,10 +165,10 @@ class BaseRepository[T, S]:
"""
try:
stmt = delete(model_class).where(model_class.torrent_id == torrent_id)
result = self.db.execute(stmt)
self.db.commit()
result = await self.db.execute(stmt)
await self.db.commit()
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
else:
return result.rowcount

View File

@@ -1,5 +1,5 @@
import logging
from collections.abc import Callable
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any, TypeVar
@@ -41,8 +41,8 @@ class BaseMediaService[T, S]:
self.indexer_service = indexer_service
self.notification_service = notification_service
def get_all_media(self) -> list[S]:
return self.repository.get_all()
async def get_all_media(self) -> list[S]:
return await self.repository.get_all()
def get_root_directory(
self, media: S, default_dir: Path, libraries: list[Any]
@@ -64,36 +64,36 @@ class BaseMediaService[T, S]:
"""
raise NotImplementedError
def notify_import_success(self, media_name: str, media_type: str) -> None:
async def notify_import_success(self, media_name: str, media_type: str) -> None:
if self.notification_service:
self.notification_service.send_notification_to_all_providers(
await self.notification_service.send_notification_to_all_providers(
title=f"{media_type.capitalize()} Downloaded",
message=f"{media_type.capitalize()} {media_name} has been successfully downloaded and imported.",
)
def notify_import_failure(
async def notify_import_failure(
self, media_name: str, media_type: str, error_msg: str = ""
) -> None:
if self.notification_service:
msg = f"Failed to import files for {media_type} {media_name}."
if error_msg:
msg += f" Error: {error_msg}"
self.notification_service.send_notification_to_all_providers(
await self.notification_service.send_notification_to_all_providers(
title="Import Failed",
message=msg,
)
def get_import_candidates(
async def get_import_candidates(
self,
directory: Path,
metadata_provider: AbstractMetadataProvider,
search_func: Callable[
[str, AbstractMetadataProvider], list[MetaDataProviderSearchResult]
[str, AbstractMetadataProvider],
Awaitable[list[MetaDataProviderSearchResult]],
],
) -> MediaImportSuggestion:
# Implementation from previous turn
name, _ = self._extract_name_and_year(directory.name)
candidates = search_func(name, metadata_provider)
candidates = await search_func(name, metadata_provider)
return MediaImportSuggestion(
directory=str(directory),
candidates=candidates,
@@ -107,47 +107,47 @@ class BaseMediaService[T, S]:
return match.group(1), int(match.group(2))
return directory_name, None
def get_importable_media(
async def get_importable_media(
self,
root_path: Path,
metadata_provider: AbstractMetadataProvider,
get_candidates_func: Callable[
[Path, AbstractMetadataProvider], MediaImportSuggestion
[Path, AbstractMetadataProvider], Awaitable[MediaImportSuggestion]
],
) -> list[MediaImportSuggestion]:
importable_dirs = get_importable_media_directories(root_path)
return [
get_candidates_func(directory, metadata_provider)
await get_candidates_func(directory, metadata_provider)
for directory in importable_dirs
]
def import_existing_media(
async def import_existing_media(
self,
media: S,
source_directory: Path,
import_func: Callable[[S, Path, Callable[[Any], None]], bool],
add_file_record_func: Callable[[Any], Any],
import_func: Callable[[S, Path, Callable[[Any], Any]], Awaitable[bool]],
add_file_record_func: Callable[[Any], Awaitable[Any]],
) -> bool:
success = import_func(media, source_directory, add_file_record_func)
success = await import_func(media, source_directory, add_file_record_func)
if success:
log.info(f"Successfully imported {media.name} from {source_directory}")
return success
def import_all_torrents_base(
async def import_all_torrents_base(
self,
get_media_func: Callable[[Any], S],
import_torrent_func: Callable[[Any, S], None],
get_media_func: Callable[[Any], Awaitable[S | None]],
import_torrent_func: Callable[[Any, S], Awaitable[None]],
media_type_name: str,
) -> None:
log.info(f"Importing all torrents for {media_type_name}")
torrents = self.torrent_service.get_completed_torrents()
torrents = await self.torrent_service.get_completed_torrents()
for t in torrents:
if t.imported:
continue
try:
media = get_media_func(t)
media = await get_media_func(t)
if media:
import_torrent_func(t, media)
await import_torrent_func(t, media)
except Exception:
log.exception(f"Error importing torrent {t.title}")
log.info(f"Finished importing all torrents for {media_type_name}")
@@ -161,9 +161,9 @@ class BaseMetadataService[T, S]:
def __init__(self, repository: BaseRepository[T, S]) -> None:
self.repository = repository
def check_if_exists(self, external_id: int, metadata_provider: str) -> bool:
async def check_if_exists(self, external_id: int, metadata_provider: str) -> bool:
try:
self.repository.get_by_external_id(
await self.repository.get_by_external_id(
external_id=external_id, metadata_provider=metadata_provider
)
except NotFoundError:
@@ -171,38 +171,40 @@ class BaseMetadataService[T, S]:
else:
return True
def add_media_base(
async def add_media_base(
self,
external_id: int,
metadata_provider: AbstractMetadataProvider, # noqa: ARG002
get_metadata_func: Callable[..., S],
save_func: Callable[[S], S],
download_poster_func: Callable[[S], bool],
get_metadata_func: Callable[..., Awaitable[S]],
save_func: Callable[[S], Awaitable[S]],
download_poster_func: Callable[[S], Awaitable[bool]],
language: str | None = None,
) -> S:
media_with_metadata = get_metadata_func(external_id, language=language)
media_with_metadata = await get_metadata_func(external_id, language=language)
if not media_with_metadata:
raise NotFoundError
saved_media = save_func(media_with_metadata)
download_poster_func(saved_media)
saved_media = await save_func(media_with_metadata)
await download_poster_func(saved_media)
return saved_media
def search_for_media_base(
async def search_for_media_base(
self,
query: str,
metadata_provider: AbstractMetadataProvider,
search_func: Callable[[str | None], list[MetaDataProviderSearchResult]],
get_by_external_id_func: Callable[..., S],
search_func: Callable[
[str | None], Awaitable[list[MetaDataProviderSearchResult]]
],
get_by_external_id_func: Callable[..., Awaitable[S]],
) -> list[MetaDataProviderSearchResult]:
results = search_func(query)
results = await search_func(query)
for result in results:
if self.check_if_exists(
if await self.check_if_exists(
external_id=result.external_id, metadata_provider=metadata_provider.name
):
result.added = True
try:
media = get_by_external_id_func(
media = await get_by_external_id_func(
external_id=result.external_id,
metadata_provider=metadata_provider.name,
)
@@ -213,30 +215,35 @@ class BaseMetadataService[T, S]:
)
return results
def get_popular_media_base(
async def get_popular_media_base(
self,
metadata_provider: AbstractMetadataProvider,
search_func: Callable[[str | None], list[MetaDataProviderSearchResult]],
search_func: Callable[
[str | None], Awaitable[list[MetaDataProviderSearchResult]]
],
) -> list[MetaDataProviderSearchResult]:
results = search_func(None)
results = await search_func(None)
return [
result
for result in results
if not self.check_if_exists(
external_id=result.external_id, metadata_provider=metadata_provider.name
r
for r in results
if not await self.check_if_exists(
external_id=r.external_id,
metadata_provider=metadata_provider.name,
)
]
def update_all_metadata_base(
async def update_all_metadata_base(
self,
get_all_to_update_func: Callable[[], list[S]],
update_single_func: Callable[[S, AbstractMetadataProvider], S | None],
get_all_to_update_func: Callable[[], Awaitable[list[S]]],
update_single_func: Callable[
[S, AbstractMetadataProvider], Awaitable[S | None]
],
tmdb_provider_class: Callable[[], AbstractMetadataProvider],
tvdb_provider_class: Callable[[], AbstractMetadataProvider],
media_type_name: str,
) -> None:
log.info(f"Updating metadata for all {media_type_name}")
media_list = get_all_to_update_func()
media_list = await get_all_to_update_func()
log.info(f"Found {len(media_list)} {media_type_name} to update")
for item in media_list:
try:
@@ -249,7 +256,7 @@ class BaseMetadataService[T, S]:
f"Unsupported provider {item.metadata_provider} for {item.name}"
)
continue
update_single_func(item, provider)
await update_single_func(item, provider)
except InvalidConfigError:
log.exception(f"Config error for {item.name}")
except Exception:

View File

@@ -1,14 +1,17 @@
import logging
import os
from collections.abc import Generator
from contextvars import ContextVar
from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import Session, declarative_base, sessionmaker
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import declarative_base
from media_manager.database.config import DbConfig
@@ -16,8 +19,8 @@ log = logging.getLogger(__name__)
Base = declarative_base()
engine: Engine | None = None
SessionLocal: sessionmaker | None = None
engine: AsyncEngine | None = None
SessionLocal: async_sessionmaker[AsyncSession] | None = None
def build_db_url(
@@ -40,7 +43,7 @@ def build_db_url(
def init_engine(
db_config: DbConfig | None = None,
url: str | URL | None = None,
) -> Engine:
) -> AsyncEngine:
"""
Initialize the global SQLAlchemy engine and session factory.
Pass either a DbConfig-like object or a full URL. Only initializes once.
@@ -64,7 +67,7 @@ def init_engine(
db_config.dbname,
)
engine = create_engine(
engine = create_async_engine(
url,
echo=False,
pool_size=10,
@@ -72,33 +75,23 @@ def init_engine(
pool_timeout=30,
pool_recycle=1800,
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
log.debug("SQLAlchemy engine initialized")
return engine
def get_engine() -> Engine:
if engine is None:
msg = "Engine not initialized. Call init_engine(...) first."
raise RuntimeError(msg)
return engine
def get_session() -> Generator[Session]:
async def get_async_session() -> AsyncIterator[AsyncSession]:
if SessionLocal is None:
msg = "Session factory not initialized. Call init_engine(...) first."
raise RuntimeError(msg)
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
log.critical("", exc_info=True)
raise
finally:
db.close()
async with SessionLocal() as db:
try:
yield db
await db.commit()
except Exception:
await db.rollback()
log.critical("", exc_info=True)
raise
db_session: ContextVar[Session] = ContextVar("db_session")
DbSessionDependency = Annotated[Session, Depends(get_session)]
DbSessionDependency = Annotated[AsyncSession, Depends(get_async_session)]

View File

@@ -1,6 +1,6 @@
import logging
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from media_manager.indexer.models import IndexerQueryResult
from media_manager.indexer.schemas import (
@@ -14,20 +14,24 @@ log = logging.getLogger(__name__)
class IndexerRepository:
def __init__(self, db: Session) -> None:
def __init__(self, db: AsyncSession) -> None:
self.db = db
def get_result(self, result_id: IndexerQueryResultId) -> IndexerQueryResultSchema:
async def get_result(
self, result_id: IndexerQueryResultId
) -> IndexerQueryResultSchema:
return IndexerQueryResultSchema.model_validate(
self.db.get(IndexerQueryResult, result_id)
await self.db.get(IndexerQueryResult, result_id)
)
def save_result(self, result: IndexerQueryResultSchema) -> IndexerQueryResultSchema:
async def save_result(
self, result: IndexerQueryResultSchema
) -> IndexerQueryResultSchema:
result_data = result.model_dump()
result_data["download_url"] = str(
result.download_url
) # this is the needful, because sqlalchemy is too dumb to handle the HttpUrl type
self.db.add(IndexerQueryResult(**result_data))
self.db.commit()
await self.db.commit()
return result

View File

@@ -1,3 +1,4 @@
import asyncio
import logging
from media_manager.config import MediaManagerConfig
@@ -24,10 +25,10 @@ class IndexerService:
if config.indexers.jackett.enabled:
self.indexers.append(Jackett())
def get_result(self, result_id: IndexerQueryResultId) -> IndexerQueryResult:
return self.repository.get_result(result_id=result_id)
async def get_result(self, result_id: IndexerQueryResultId) -> IndexerQueryResult:
return await self.repository.get_result(result_id=result_id)
def search(self, query: str, is_tv: bool) -> list[IndexerQueryResult]:
async def search(self, query: str, is_tv: bool) -> list[IndexerQueryResult]:
"""
Search for results using the indexers based on a query.
@@ -40,7 +41,9 @@ class IndexerService:
for indexer in self.indexers:
try:
indexer_results = indexer.search(query, is_tv=is_tv)
indexer_results = await asyncio.to_thread(
indexer.search, query, is_tv=is_tv
)
results.extend(indexer_results)
log.debug(
f"Indexer {indexer.__class__.__name__} returned {len(indexer_results)} results for query: {query}"
@@ -51,18 +54,20 @@ class IndexerService:
)
for result in results:
self.repository.save_result(result=result)
await self.repository.save_result(result=result)
return results
def search_movie(self, movie: Movie) -> list[IndexerQueryResult]:
async def search_movie(self, movie: Movie) -> list[IndexerQueryResult]:
query = f"{movie.name} {movie.year}"
query = remove_special_chars_and_parentheses(query)
results = []
for indexer in self.indexers:
try:
indexer_results = indexer.search_movie(query=query, movie=movie)
indexer_results = await asyncio.to_thread(
indexer.search_movie, query=query, movie=movie
)
if indexer_results:
results.extend(indexer_results)
except Exception:
@@ -71,19 +76,24 @@ class IndexerService:
)
for result in results:
self.repository.save_result(result=result)
await self.repository.save_result(result=result)
return results
def search_season(self, show: Show, season_number: int) -> list[IndexerQueryResult]:
async def search_season(
self, show: Show, season_number: int
) -> list[IndexerQueryResult]:
query = f"{show.name} {show.year} S{season_number:02d}"
query = remove_special_chars_and_parentheses(query)
results = []
for indexer in self.indexers:
try:
indexer_results = indexer.search_season(
query=query, show=show, season_number=season_number
indexer_results = await asyncio.to_thread(
indexer.search_season,
query=query,
show=show,
season_number=season_number,
)
if indexer_results:
results.extend(indexer_results)
@@ -93,6 +103,6 @@ class IndexerService:
)
for result in results:
self.repository.save_result(result=result)
await self.repository.save_result(result=result)
return results

View File

@@ -18,27 +18,31 @@ class AbstractMetadataProvider(ABC):
pass
@abstractmethod
def get_show_metadata(self, show_id: int, language: str | None = None) -> Show:
async def get_show_metadata(
self, show_id: int, language: str | None = None
) -> Show:
raise NotImplementedError()
@abstractmethod
def get_movie_metadata(self, movie_id: int, language: str | None = None) -> Movie:
async def get_movie_metadata(
self, movie_id: int, language: str | None = None
) -> Movie:
raise NotImplementedError()
@abstractmethod
def search_show(
async def search_show(
self, query: str | None = None
) -> list[MetaDataProviderSearchResult]:
raise NotImplementedError()
@abstractmethod
def search_movie(
async def search_movie(
self, query: str | None = None
) -> list[MetaDataProviderSearchResult]:
raise NotImplementedError()
@abstractmethod
def download_show_poster_image(self, show: Show) -> bool:
async def download_show_poster_image(self, show: Show) -> bool:
"""
Downloads the poster image for a show.
:param show: The show to download the poster image for.
@@ -47,7 +51,7 @@ class AbstractMetadataProvider(ABC):
raise NotImplementedError()
@abstractmethod
def download_movie_poster_image(self, movie: Movie) -> bool:
async def download_movie_poster_image(self, movie: Movie) -> bool:
"""
Downloads the poster image for a show.
:param movie: The show to download the poster image for.

View File

@@ -1,7 +1,7 @@
import logging
from typing import override
import requests
import httpx
import media_manager.metadataProvider.utils
from media_manager.config import MediaManagerConfig
@@ -17,6 +17,8 @@ ENDED_STATUS = {"Ended", "Canceled"}
log = logging.getLogger(__name__)
_client = httpx.AsyncClient(timeout=30.0)
class TmdbMetadataProvider(AbstractMetadataProvider):
name = "tmdb"
@@ -39,70 +41,72 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return original_language
return self.default_language
def __get_show_metadata(self, show_id: int, language: str | None = None) -> dict:
async def __get_show_metadata(
self, show_id: int, language: str | None = None
) -> dict:
if language is None:
language = self.default_language
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/tv/shows/{show_id}",
params={"language": language},
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(f"TMDB API error getting show metadata for ID {show_id}")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch show metadata for ID {show_id} from TMDB. Error: {e}",
)
raise
def __get_show_external_ids(self, show_id: int) -> dict:
async def __get_show_external_ids(self, show_id: int) -> dict:
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/tv/shows/{show_id}/external_ids",
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(f"TMDB API error getting show external IDs for ID {show_id}")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch show external IDs for ID {show_id} from TMDB. Error: {e}",
)
raise
def __get_season_metadata(
async def __get_season_metadata(
self, show_id: int, season_number: int, language: str | None = None
) -> dict:
if language is None:
language = self.default_language
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/tv/shows/{show_id}/{season_number}",
params={"language": language},
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(
f"TMDB API error getting season {season_number} metadata for show ID {show_id}"
)
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch season {season_number} metadata for show ID {show_id} from TMDB. Error: {e}",
)
raise
def __search_tv(self, query: str, page: int) -> dict:
async def __search_tv(self, query: str, page: int) -> dict:
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/tv/search",
params={
"query": query,
@@ -112,74 +116,76 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(f"TMDB API error searching TV shows with query '{query}'")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to search TV shows with query '{query}' on TMDB. Error: {e}",
)
raise
def __get_trending_tv(self) -> dict:
async def __get_trending_tv(self) -> dict:
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/tv/trending",
params={"language": self.default_language},
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception("TMDB API error getting trending TV")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch trending TV shows from TMDB. Error: {e}",
)
raise
def __get_movie_metadata(self, movie_id: int, language: str | None = None) -> dict:
async def __get_movie_metadata(
self, movie_id: int, language: str | None = None
) -> dict:
if language is None:
language = self.default_language
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/movies/{movie_id}",
params={"language": language},
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(f"TMDB API error getting movie metadata for ID {movie_id}")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch movie metadata for ID {movie_id} from TMDB. Error: {e}",
)
raise
def __get_movie_external_ids(self, movie_id: int) -> dict:
async def __get_movie_external_ids(self, movie_id: int) -> dict:
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/movies/{movie_id}/external_ids", timeout=60
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(
f"TMDB API error getting movie external IDs for ID {movie_id}"
)
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch movie external IDs for ID {movie_id} from TMDB. Error: {e}",
)
raise
def __search_movie(self, query: str, page: int) -> dict:
async def __search_movie(self, query: str, page: int) -> dict:
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/movies/search",
params={
"query": query,
@@ -189,40 +195,42 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception(f"TMDB API error searching movies with query '{query}'")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to search movies with query '{query}' on TMDB. Error: {e}",
)
raise
def __get_trending_movies(self) -> dict:
async def __get_trending_movies(self) -> dict:
try:
response = requests.get(
response = await _client.get(
url=f"{self.url}/movies/trending",
params={"language": self.default_language},
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
except httpx.HTTPError as e:
log.exception("TMDB API error getting trending movies")
if notification_manager.is_configured():
notification_manager.send_notification(
await notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch trending movies from TMDB. Error: {e}",
)
raise
@override
def download_show_poster_image(self, show: Show) -> bool:
async def download_show_poster_image(self, show: Show) -> bool:
# Determine which language to use based on show's original_language
language = self.__get_language_param(show.original_language)
# Fetch metadata in the appropriate language to get localized poster
show_metadata = self.__get_show_metadata(show.external_id, language=language)
show_metadata = await self.__get_show_metadata(
show.external_id, language=language
)
# downloading the poster
# all pictures from TMDB should already be jpeg, so no need to convert
@@ -230,7 +238,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
poster_url = (
"https://image.tmdb.org/t/p/original" + show_metadata["poster_path"]
)
if media_manager.metadataProvider.utils.download_poster_image(
if await media_manager.metadataProvider.utils.download_poster_image(
storage_path=self.storage_path, poster_url=poster_url, uuid=show.id
):
log.info("Successfully downloaded poster image for show " + show.name)
@@ -243,7 +251,9 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return True
@override
def get_show_metadata(self, show_id: int, language: str | None = None) -> Show:
async def get_show_metadata(
self, show_id: int, language: str | None = None
) -> Show:
"""
:param show_id: the external id of the show
@@ -255,45 +265,44 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"""
# If language not provided, fetch once to determine original language
if language is None:
show_metadata = self.__get_show_metadata(show_id)
show_metadata = await self.__get_show_metadata(show_id)
language = show_metadata.get("original_language")
# Determine which language to use for metadata
language = self.__get_language_param(language)
# Fetch show metadata in the appropriate language
show_metadata = self.__get_show_metadata(show_id, language=language)
show_metadata = await self.__get_show_metadata(show_id, language=language)
# get imdb id
external_ids = self.__get_show_external_ids(show_id=show_id)
external_ids = await self.__get_show_external_ids(show_id=show_id)
imdb_id = external_ids.get("imdb_id")
season_list = []
# inserting all the metadata into the objects
for season in show_metadata["seasons"]:
season_metadata = self.__get_season_metadata(
season_metadata_list = [
await self.__get_season_metadata(
show_id=show_metadata["id"],
season_number=season["season_number"],
language=language,
)
episode_list = [
Episode(
external_id=int(episode["id"]),
title=episode["name"],
number=EpisodeNumber(episode["episode_number"]),
)
for episode in season_metadata["episodes"]
]
season_list.append(
Season(
external_id=int(season_metadata["id"]),
name=season_metadata["name"],
overview=season_metadata["overview"],
number=SeasonNumber(season_metadata["season_number"]),
episodes=episode_list,
)
for season in show_metadata["seasons"]
]
season_list = [
Season(
external_id=int(season_metadata["id"]),
name=season_metadata["name"],
overview=season_metadata["overview"],
number=SeasonNumber(season_metadata["season_number"]),
episodes=[
Episode(
external_id=int(episode["id"]),
title=episode["name"],
number=EpisodeNumber(episode["episode_number"]),
)
for episode in season_metadata["episodes"]
],
)
for season_metadata in season_metadata_list
]
year = media_manager.metadataProvider.utils.get_year_from_date(
show_metadata["first_air_date"]
@@ -312,7 +321,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
@override
def search_show(
async def search_show(
self, query: str | None = None, max_pages: int = 5
) -> list[MetaDataProviderSearchResult]:
"""
@@ -321,10 +330,10 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"""
results = []
if query is None:
results = self.__get_trending_tv()["results"]
results = (await self.__get_trending_tv())["results"]
else:
for page_number in range(1, max_pages + 1):
result_page = self.__search_tv(query=query, page=page_number)
result_page = await self.__search_tv(query=query, page=page_number)
if not result_page["results"]:
break
@@ -371,7 +380,9 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return formatted_results
@override
def get_movie_metadata(self, movie_id: int, language: str | None = None) -> Movie:
async def get_movie_metadata(
self, movie_id: int, language: str | None = None
) -> Movie:
"""
Get movie metadata with language-aware fetching.
@@ -384,17 +395,19 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"""
# If language not provided, fetch once to determine original language
if language is None:
movie_metadata = self.__get_movie_metadata(movie_id=movie_id)
movie_metadata = await self.__get_movie_metadata(movie_id=movie_id)
language = movie_metadata.get("original_language")
# Determine which language to use for metadata
language = self.__get_language_param(language)
# Fetch movie metadata in the appropriate language
movie_metadata = self.__get_movie_metadata(movie_id=movie_id, language=language)
movie_metadata = await self.__get_movie_metadata(
movie_id=movie_id, language=language
)
# get imdb id
external_ids = self.__get_movie_external_ids(movie_id=movie_id)
external_ids = await self.__get_movie_external_ids(movie_id=movie_id)
imdb_id = external_ids.get("imdb_id")
year = media_manager.metadataProvider.utils.get_year_from_date(
@@ -412,7 +425,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
@override
def search_movie(
async def search_movie(
self, query: str | None = None, max_pages: int = 5
) -> list[MetaDataProviderSearchResult]:
"""
@@ -421,10 +434,10 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"""
results = []
if query is None:
results = self.__get_trending_movies()["results"]
results = (await self.__get_trending_movies())["results"]
else:
for page_number in range(1, max_pages + 1):
result_page = self.__search_movie(query=query, page=page_number)
result_page = await self.__search_movie(query=query, page=page_number)
if not result_page["results"]:
break
@@ -471,12 +484,12 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return formatted_results
@override
def download_movie_poster_image(self, movie: Movie) -> bool:
async def download_movie_poster_image(self, movie: Movie) -> bool:
# Determine which language to use based on movie's original_language
language = self.__get_language_param(movie.original_language)
# Fetch metadata in the appropriate language to get localized poster
movie_metadata = self.__get_movie_metadata(
movie_metadata = await self.__get_movie_metadata(
movie_id=movie.external_id, language=language
)
@@ -486,7 +499,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
poster_url = (
"https://image.tmdb.org/t/p/original" + movie_metadata["poster_path"]
)
if media_manager.metadataProvider.utils.download_poster_image(
if await media_manager.metadataProvider.utils.download_poster_image(
storage_path=self.storage_path, poster_url=poster_url, uuid=movie.id
):
log.info("Successfully downloaded poster image for movie " + movie.name)

View File

@@ -1,7 +1,7 @@
import logging
from typing import override
import requests
import httpx
import media_manager.metadataProvider.utils
from media_manager.config import MediaManagerConfig
@@ -14,6 +14,8 @@ from media_manager.tv.schemas import Episode, Season, SeasonNumber, Show
log = logging.getLogger(__name__)
_client = httpx.AsyncClient(timeout=30.0)
class TvdbMetadataProvider(AbstractMetadataProvider):
name = "tvdb"
@@ -22,37 +24,46 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
config = MediaManagerConfig().metadata.tvdb
self.url = config.tvdb_relay_url
def __get_show(self, show_id: int) -> dict:
return requests.get(url=f"{self.url}/tv/shows/{show_id}", timeout=60).json()
async def __get_show(self, show_id: int) -> dict:
response = await _client.get(url=f"{self.url}/tv/shows/{show_id}", timeout=60)
return response.json()
def __get_season(self, show_id: int) -> dict:
return requests.get(url=f"{self.url}/tv/seasons/{show_id}", timeout=60).json()
async def __get_season(self, show_id: int) -> dict:
response = await _client.get(
url=f"{self.url}/tv/seasons/{show_id}", timeout=60
)
return response.json()
def __search_tv(self, query: str) -> dict:
return requests.get(
async def __search_tv(self, query: str) -> dict:
response = await _client.get(
url=f"{self.url}/tv/search", params={"query": query}, timeout=60
).json()
)
return response.json()
def __get_trending_tv(self) -> dict:
return requests.get(url=f"{self.url}/tv/trending", timeout=60).json()
async def __get_trending_tv(self) -> dict:
response = await _client.get(url=f"{self.url}/tv/trending", timeout=60)
return response.json()
def __get_movie(self, movie_id: int) -> dict:
return requests.get(url=f"{self.url}/movies/{movie_id}", timeout=60).json()
async def __get_movie(self, movie_id: int) -> dict:
response = await _client.get(url=f"{self.url}/movies/{movie_id}", timeout=60)
return response.json()
def __search_movie(self, query: str) -> dict:
return requests.get(
async def __search_movie(self, query: str) -> dict:
response = await _client.get(
url=f"{self.url}/movies/search", params={"query": query}, timeout=60
).json()
)
return response.json()
def __get_trending_movies(self) -> dict:
return requests.get(url=f"{self.url}/movies/trending", timeout=60).json()
async def __get_trending_movies(self) -> dict:
response = await _client.get(url=f"{self.url}/movies/trending", timeout=60)
return response.json()
@override
def download_show_poster_image(self, show: Show) -> bool:
show_metadata = self.__get_show(show_id=show.external_id)
async def download_show_poster_image(self, show: Show) -> bool:
show_metadata = await self.__get_show(show_id=show.external_id)
if show_metadata["image"] is not None:
media_manager.metadataProvider.utils.download_poster_image(
await media_manager.metadataProvider.utils.download_poster_image(
storage_path=self.storage_path,
poster_url=show_metadata["image"],
uuid=show.id,
@@ -63,13 +74,15 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return False
@override
def get_show_metadata(self, show_id: int, language: str | None = None) -> Show:
async def get_show_metadata(
self, show_id: int, language: str | None = None
) -> Show:
"""
:param show_id: The external id of the show
:param language: does nothing, TVDB does not support multiple languages
"""
series = self.__get_show(show_id)
series = await self.__get_show(show_id)
seasons = []
seasons_ids = [season["id"] for season in series["seasons"]]
@@ -81,11 +94,10 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
if remote_id.get("type") == 2:
imdb_id = remote_id.get("id")
for season in seasons_ids:
s = self.__get_season(show_id=season)
# the seasons need to be filtered to a certain type,
# otherwise the same season will be imported in aired and dvd order,
# which causes duplicate season number + show ids which in turn violates a unique constraint of the season table
season_payloads = [await self.__get_season(show_id=sid) for sid in seasons_ids]
for s in season_payloads:
# Filter to "aired order" only; mixing aired/dvd orders duplicates
# (show_id, season_number) and violates the seasons unique constraint.
if s["type"]["id"] != 1:
log.info(
f"Season {s['type']['id']} will not be downloaded because it is not a 'aired order' season"
@@ -122,11 +134,11 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
)
@override
def search_show(
async def search_show(
self, query: str | None = None
) -> list[MetaDataProviderSearchResult]:
if query:
results = self.__search_tv(query=query)
results = await self.__search_tv(query=query)
formatted_results = []
for result in results:
try:
@@ -151,7 +163,7 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
except Exception:
log.warning("Error processing search result", exc_info=True)
return formatted_results
results = self.__get_trending_tv()
results = await self.__get_trending_tv()
formatted_results = []
for result in results:
try:
@@ -181,20 +193,16 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return formatted_results
@override
def search_movie(
async def search_movie(
self, query: str | None = None
) -> list[MetaDataProviderSearchResult]:
if query:
results = self.__search_movie(query=query)
results = results[0:20]
results = await self.__search_movie(query=query)
results = [r for r in results[0:20] if r["type"] == "movie"]
log.debug(f"got {len(results)} results from TVDB search")
movie_payloads = [await self.__get_movie(r["tvdb_id"]) for r in results]
formatted_results = []
for result in results:
if result["type"] != "movie":
continue
result = self.__get_movie(result["tvdb_id"])
for result in movie_payloads:
try:
try:
year = result["year"]
@@ -216,12 +224,12 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
except Exception:
log.warning("Error processing search result", exc_info=True)
return formatted_results
results = self.__get_trending_movies()
results = await self.__get_trending_movies()
results = results[0:20]
log.debug(f"got {len(results)} results from TVDB search")
movie_payloads = [await self.__get_movie(r["id"]) for r in results]
formatted_results = []
for result in results:
result = self.__get_movie(result["id"])
for result in movie_payloads:
try:
try:
year = result["year"]
@@ -252,11 +260,11 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return formatted_results
@override
def download_movie_poster_image(self, movie: Movie) -> bool:
movie_metadata = self.__get_movie(movie.external_id)
async def download_movie_poster_image(self, movie: Movie) -> bool:
movie_metadata = await self.__get_movie(movie.external_id)
if movie_metadata["image"] is not None:
media_manager.metadataProvider.utils.download_poster_image(
await media_manager.metadataProvider.utils.download_poster_image(
storage_path=self.storage_path,
poster_url=movie_metadata["image"],
uuid=movie.id,
@@ -267,14 +275,16 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return False
@override
def get_movie_metadata(self, movie_id: int, language: str | None = None) -> Movie:
async def get_movie_metadata(
self, movie_id: int, language: str | None = None
) -> Movie:
"""
:param movie_id: the external id of the movie
:param language: does nothing, TVDB does not support multiple languages
:return: returns a Movie object
"""
movie = self.__get_movie(movie_id=movie_id)
movie = await self.__get_movie(movie_id=movie_id)
# get imdb id from remote ids
imdb_id = None

View File

@@ -1,7 +1,8 @@
import asyncio
from pathlib import Path
from uuid import UUID
import requests
import httpx
from PIL import Image
@@ -11,15 +12,22 @@ def get_year_from_date(first_air_date: str | None) -> int | None:
return None
def download_poster_image(storage_path: Path, poster_url: str, uuid: UUID) -> bool:
res = requests.get(poster_url, stream=True, timeout=60)
def _process_image(image_file_path: Path, content: bytes) -> None:
image_file_path.write_bytes(content)
original_image = Image.open(image_file_path)
original_image.save(image_file_path.with_suffix(".avif"), quality=50)
original_image.save(image_file_path.with_suffix(".webp"), quality=50)
async def download_poster_image(
storage_path: Path, poster_url: str, uuid: UUID
) -> bool:
async with httpx.AsyncClient(timeout=60.0) as client:
res = await client.get(poster_url)
if res.status_code == 200:
image_file_path = storage_path.joinpath(str(uuid)).with_suffix(".jpg")
image_file_path.write_bytes(res.content)
original_image = Image.open(image_file_path)
original_image.save(image_file_path.with_suffix(".avif"), quality=50)
original_image.save(image_file_path.with_suffix(".webp"), quality=50)
await asyncio.to_thread(_process_image, image_file_path, res.content)
return True
return False

View File

@@ -72,12 +72,12 @@ def get_movie_service(
movie_service_dep = Annotated[MovieService, Depends(get_movie_service)]
def get_movie_by_id(
async def get_movie_by_id(
movie_service: movie_service_dep,
movie_id: MovieId = Path(..., description="The ID of the movie"),
) -> Movie:
try:
movie = movie_service.get_movie_by_id(movie_id)
movie = await movie_service.get_movie_by_id(movie_id)
except NotFoundError:
raise HTTPException(
status_code=404,

View File

@@ -1,6 +1,7 @@
import asyncio
import logging
import re
from collections.abc import Callable
from collections.abc import Awaitable, Callable
from pathlib import Path
from media_manager.common.service import BaseMediaService
@@ -49,7 +50,7 @@ class MovieImportService(BaseMediaService[Movie, Movie]):
libraries=misc_config.movie_libraries,
)
def import_movie(
async def import_movie(
self,
movie: Movie,
video_files: list[Path],
@@ -72,7 +73,9 @@ class MovieImportService(BaseMediaService[Movie, Movie]):
target_video_file = (
movie_root_path / f"{movie_file_name}{video_files[0].suffix}"
)
import_file(target_file=target_video_file, source_file=video_files[0])
await asyncio.to_thread(
import_file, target_file=target_video_file, source_file=video_files[0]
)
imported_any = True
for subtitle_file in subtitle_files:
@@ -82,7 +85,9 @@ class MovieImportService(BaseMediaService[Movie, Movie]):
if match:
lang = match.group(1)
target = movie_root_path / f"{movie_file_name}.{lang}.srt"
import_file(target_file=target, source_file=subtitle_file)
await asyncio.to_thread(
import_file, target_file=target, source_file=subtitle_file
)
imported_any = True
except Exception:
log.exception(f"Failed to import movie {movie.name}")
@@ -90,50 +95,55 @@ class MovieImportService(BaseMediaService[Movie, Movie]):
else:
return imported_any
def import_torrent_files(self, torrent: Torrent, movie: Movie) -> None:
video_files, subtitle_files, _ = get_files_for_import(torrent=torrent)
async def import_torrent_files(self, torrent: Torrent, movie: Movie) -> None:
# Filesystem scan + archive extraction; offload off the event loop.
video_files, subtitle_files, _ = await asyncio.to_thread(
get_files_for_import, torrent=torrent
)
if len(video_files) != 1:
self.notify_import_failure(
await self.notify_import_failure(
movie.name,
"movie",
"Multiple video files found. Manual import required.",
)
return
movie_files = self.torrent_service.get_movie_files_of_torrent(torrent=torrent)
movie_files = await self.torrent_service.get_movie_files_of_torrent(torrent=torrent)
if not movie_files:
torrent.imported = False
self.torrent_service.torrent_repository.save_torrent(torrent=torrent)
self.notify_import_failure(movie.name, "movie")
await self.torrent_service.torrent_repository.save_torrent(torrent=torrent)
await self.notify_import_failure(movie.name, "movie")
return
success = [
self.import_movie(movie, video_files, subtitle_files, mf.file_path_suffix)
await self.import_movie(movie, video_files, subtitle_files, mf.file_path_suffix)
for mf in movie_files
]
if all(success):
torrent.imported = True
self.torrent_service.torrent_repository.save_torrent(torrent=torrent)
self.notify_import_success(movie.name, "movie")
await self.torrent_service.torrent_repository.save_torrent(torrent=torrent)
await self.notify_import_success(movie.name, "movie")
else:
self.notify_import_failure(movie.name, "movie")
await self.notify_import_failure(movie.name, "movie")
def get_import_candidates(
async def get_import_candidates(
self, movie_path: Path, metadata_provider: AbstractMetadataProvider
) -> MediaImportSuggestion:
return super().get_import_candidates(
return await super().get_import_candidates(
directory=movie_path,
metadata_provider=metadata_provider,
search_func=self.movie_metadata_service.search_for_movie,
)
def import_existing_movie(self, movie: Movie, source_directory: Path) -> bool:
def _logic(m: Movie, path: Path, add_cb: Callable[[MovieFile], None]) -> bool:
v, s, _ = get_files_for_import(directory=path)
res = self.import_movie(m, v, s, "IMPORTED")
async def import_existing_movie(self, movie: Movie, source_directory: Path) -> bool:
async def _logic(
m: Movie, path: Path, add_cb: Callable[[MovieFile], Awaitable[None]]
) -> bool:
v, s, _ = await asyncio.to_thread(get_files_for_import, directory=path)
res = await self.import_movie(m, v, s, "IMPORTED")
if res:
add_cb(
await add_cb(
MovieFile(
movie_id=m.id,
file_path_suffix="IMPORTED",
@@ -143,24 +153,24 @@ class MovieImportService(BaseMediaService[Movie, Movie]):
)
return res
return self.import_existing_media(
return await self.import_existing_media(
media=movie,
source_directory=source_directory,
import_func=_logic,
add_file_record_func=self.movie_repository.add_movie_file,
)
def get_importable_movies(
async def get_importable_movies(
self, metadata_provider: AbstractMetadataProvider
) -> list[MediaImportSuggestion]:
return self.get_importable_media(
return await self.get_importable_media(
root_path=MediaManagerConfig().misc.movie_directory,
metadata_provider=metadata_provider,
get_candidates_func=self.get_import_candidates,
)
def import_all_torrents(self) -> None:
self.import_all_torrents_base(
async def import_all_torrents(self) -> None:
await self.import_all_torrents_base(
get_media_func=self.torrent_service.get_movie_of_torrent,
import_torrent_func=self.import_torrent_files,
media_type_name="movie",

View File

@@ -18,13 +18,13 @@ class MovieMetadataService(BaseMetadataService[Movie, Movie]):
super().__init__(repository=movie_repository)
self.movie_repository = movie_repository
def add_movie(
async def add_movie(
self,
external_id: int,
metadata_provider: AbstractMetadataProvider,
language: str | None = None,
) -> Movie:
return self.add_media_base(
return await self.add_media_base(
external_id=external_id,
metadata_provider=metadata_provider,
get_metadata_func=metadata_provider.get_movie_metadata,
@@ -33,51 +33,52 @@ class MovieMetadataService(BaseMetadataService[Movie, Movie]):
language=language,
)
def search_for_movie(
async def search_for_movie(
self, query: str, metadata_provider: AbstractMetadataProvider
) -> list[MetaDataProviderSearchResult]:
return self.search_for_media_base(
return await self.search_for_media_base(
query=query,
metadata_provider=metadata_provider,
search_func=metadata_provider.search_movie,
get_by_external_id_func=self.movie_repository.get_movie_by_external_id,
)
def get_popular_movies(
async def get_popular_movies(
self, metadata_provider: AbstractMetadataProvider
) -> list[MetaDataProviderSearchResult]:
return self.get_popular_media_base(
return await self.get_popular_media_base(
metadata_provider=metadata_provider,
search_func=metadata_provider.search_movie,
)
def update_movie_metadata(
async def update_movie_metadata(
self, db_movie: Movie, metadata_provider: AbstractMetadataProvider
) -> Movie | None:
"""
Updates the metadata of a movie.
"""
log.debug(f"Found movie: {db_movie.name} for metadata update.")
fresh_movie_data = metadata_provider.get_movie_metadata(
movie_id=db_movie.external_id, language=db_movie.original_language
fresh_movie_data = await metadata_provider.get_movie_metadata(
movie_id=db_movie.external_id,
language=db_movie.original_language,
)
if not fresh_movie_data:
log.warning(f"Could not fetch fresh metadata for movie: {db_movie.name}")
return None
self.movie_repository.update_movie_attributes(
await self.movie_repository.update_movie_attributes(
movie_id=db_movie.id,
name=fresh_movie_data.name,
overview=fresh_movie_data.overview,
year=fresh_movie_data.year,
imdb_id=fresh_movie_data.imdb_id,
)
updated_movie = self.movie_repository.get_movie_by_id(db_movie.id)
metadata_provider.download_movie_poster_image(movie=updated_movie)
updated_movie = await self.movie_repository.get_movie_by_id(db_movie.id)
await metadata_provider.download_movie_poster_image(movie=updated_movie)
return updated_movie
def update_all_metadata(self) -> None:
self.update_all_metadata_base(
async def update_all_metadata(self) -> None:
await self.update_all_metadata_base(
get_all_to_update_func=self.movie_repository.get_movies,
update_single_func=self.update_movie_metadata,
tmdb_provider_class=TmdbMetadataProvider,

View File

@@ -2,7 +2,7 @@ import logging
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from media_manager.common.repository import BaseRepository
from media_manager.exceptions import NotFoundError
@@ -31,45 +31,47 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
Provides methods to retrieve, save, and delete movies.
"""
def __init__(self, db: Session) -> None:
def __init__(self, db: AsyncSession) -> None:
super().__init__(db, Movie, MovieSchema)
def get_movie_by_id(self, movie_id: MovieId) -> MovieSchema:
return self.get_by_id(entity_id=movie_id)
async def get_movie_by_id(self, movie_id: MovieId) -> MovieSchema:
return await self.get_by_id(entity_id=movie_id)
def get_movie_by_external_id(
async def get_movie_by_external_id(
self, external_id: int, metadata_provider: str
) -> MovieSchema:
return self.get_by_external_id(
return await self.get_by_external_id(
external_id=external_id, metadata_provider=metadata_provider
)
def get_movies(self) -> list[MovieSchema]:
return self.get_all()
async def get_movies(self) -> list[MovieSchema]:
return await self.get_all()
def delete_movie(self, movie_id: MovieId) -> None:
self.delete(entity_id=movie_id)
async def delete_movie(self, movie_id: MovieId) -> None:
await self.delete(entity_id=movie_id)
def set_movie_library(self, movie_id: MovieId, library: str) -> None:
self.set_library(entity_id=movie_id, library=library)
async def set_movie_library(self, movie_id: MovieId, library: str) -> None:
await self.set_library(entity_id=movie_id, library=library)
def save_movie(self, movie: MovieSchema) -> MovieSchema:
return self.save_media_base(media_schema=movie, model_class=Movie)
async def save_movie(self, movie: MovieSchema) -> MovieSchema:
return await self.save_media_base(media_schema=movie, model_class=Movie)
def add_movie_file(self, movie_file: MovieFileSchema) -> MovieFileSchema:
return self.add_media_file_base(
async def add_movie_file(self, movie_file: MovieFileSchema) -> MovieFileSchema:
return await self.add_media_file_base(
file_schema=movie_file, model_class=MovieFile, schema_class=MovieFileSchema
)
def remove_movie_files_by_torrent_id(self, torrent_id: TorrentId) -> int:
return self.remove_files_by_torrent_id_base(
async def remove_movie_files_by_torrent_id(self, torrent_id: TorrentId) -> int:
return await self.remove_files_by_torrent_id_base(
torrent_id=torrent_id, model_class=MovieFile
)
def get_movie_files_by_movie_id(self, movie_id: MovieId) -> list[MovieFileSchema]:
async def get_movie_files_by_movie_id(
self, movie_id: MovieId
) -> list[MovieFileSchema]:
try:
stmt = select(MovieFile).where(MovieFile.movie_id == movie_id)
results = self.db.execute(stmt).scalars().all()
results = (await self.db.execute(stmt)).scalars().all()
return [MovieFileSchema.model_validate(sf) for sf in results]
except SQLAlchemyError:
log.exception(
@@ -77,7 +79,9 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
)
raise
def get_torrents_by_movie_id(self, movie_id: MovieId) -> list[MovieTorrentSchema]:
async def get_torrents_by_movie_id(
self, movie_id: MovieId
) -> list[MovieTorrentSchema]:
try:
stmt = (
select(Torrent, MovieFile.file_path_suffix)
@@ -85,7 +89,7 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
.join(MovieFile, MovieFile.torrent_id == Torrent.id)
.where(MovieFile.movie_id == movie_id)
)
results = self.db.execute(stmt).all()
results = (await self.db.execute(stmt)).all()
formatted_results = []
for torrent, file_path_suffix in results:
movie_torrent = MovieTorrentSchema(
@@ -104,7 +108,7 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
else:
return formatted_results
def get_all_movies_with_torrents(self) -> list[MovieSchema]:
async def get_all_movies_with_torrents(self) -> list[MovieSchema]:
try:
stmt = (
select(Movie)
@@ -113,20 +117,20 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
.join(Torrent, MovieFile.torrent_id == Torrent.id)
.order_by(Movie.name)
)
results = self.db.execute(stmt).scalars().unique().all()
results = (await self.db.execute(stmt)).scalars().unique().all()
return [MovieSchema.model_validate(movie) for movie in results]
except SQLAlchemyError:
log.exception("Database error retrieving all movies with torrents")
raise
def get_movie_by_torrent_id(self, torrent_id: TorrentId) -> MovieSchema:
async def get_movie_by_torrent_id(self, torrent_id: TorrentId) -> MovieSchema:
try:
stmt = (
select(Movie)
.join(MovieFile, Movie.id == MovieFile.movie_id)
.where(MovieFile.torrent_id == torrent_id)
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if not result:
msg = f"Movie for torrent_id {torrent_id} not found."
raise NotFoundError(msg)
@@ -136,7 +140,7 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
else:
return MovieSchema.model_validate(result)
def update_movie_attributes(
async def update_movie_attributes(
self,
movie_id: MovieId,
name: str | None = None,
@@ -144,7 +148,7 @@ class MovieRepository(BaseRepository[Movie, MovieSchema]):
year: int | None = None,
imdb_id: str | None = None,
) -> MovieSchema:
return self.update_media_attributes_base(
return await self.update_media_attributes_base(
media_id=movie_id,
model_class=Movie,
name=name,

View File

@@ -38,7 +38,7 @@ router = APIRouter()
"/search",
dependencies=[Depends(current_active_user)],
)
def search_for_movie(
async def search_for_movie(
query: str,
movie_metadata_service: movie_metadata_service_dep,
metadata_provider: metadata_provider_dep,
@@ -46,7 +46,7 @@ def search_for_movie(
"""
Search for a movie on the configured metadata provider.
"""
return movie_metadata_service.search_for_movie(
return await movie_metadata_service.search_for_movie(
query=query, metadata_provider=metadata_provider
)
@@ -55,14 +55,14 @@ def search_for_movie(
"/recommended",
dependencies=[Depends(current_active_user)],
)
def get_popular_movies(
async def get_popular_movies(
movie_metadata_service: movie_metadata_service_dep,
metadata_provider: metadata_provider_dep,
) -> list[MetaDataProviderSearchResult]:
"""
Get a list of recommended/popular movies from the metadata provider.
"""
return movie_metadata_service.get_popular_movies(
return await movie_metadata_service.get_popular_movies(
metadata_provider=metadata_provider
)
@@ -77,14 +77,14 @@ def get_popular_movies(
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_superuser)],
)
def get_all_importable_movies(
async def get_all_importable_movies(
movie_import_service: movie_import_service_dep,
metadata_provider: metadata_provider_dep,
) -> list[MediaImportSuggestion]:
"""
Get a list of unknown movies that were detected in the movie directory and are importable.
"""
return movie_import_service.get_importable_movies(
return await movie_import_service.get_importable_movies(
metadata_provider=metadata_provider
)
@@ -94,7 +94,7 @@ def get_all_importable_movies(
dependencies=[Depends(current_superuser)],
status_code=status.HTTP_204_NO_CONTENT,
)
def import_detected_movie(
async def import_detected_movie(
movie_import_service: movie_import_service_dep, movie: movie_dep, directory: str
) -> None:
"""
@@ -105,7 +105,7 @@ def import_detected_movie(
MediaManagerConfig().misc.movie_directory
):
raise HTTPException(status.HTTP_400_BAD_REQUEST, "No such directory")
success = movie_import_service.import_existing_movie(
success = await movie_import_service.import_existing_movie(
movie=movie, source_directory=source_directory
)
if not success:
@@ -121,11 +121,11 @@ def import_detected_movie(
"",
dependencies=[Depends(current_active_user)],
)
def get_all_movies(movie_service: movie_service_dep) -> list[Movie]:
async def get_all_movies(movie_service: movie_service_dep) -> list[Movie]:
"""
Get all movies in the library.
"""
return movie_service.get_all_movies()
return await movie_service.get_all_movies()
@router.post(
@@ -139,7 +139,7 @@ def get_all_movies(movie_service: movie_service_dep) -> list[Movie]:
}
},
)
def add_a_movie(
async def add_a_movie(
movie_metadata_service: movie_metadata_service_dep,
metadata_provider: metadata_provider_dep,
movie_id: int,
@@ -149,13 +149,13 @@ def add_a_movie(
Add a new movie to the library.
"""
try:
movie = movie_metadata_service.add_movie(
movie = await movie_metadata_service.add_movie(
external_id=movie_id,
metadata_provider=metadata_provider,
language=language,
)
except ConflictError:
movie = movie_metadata_service.movie_repository.get_movie_by_external_id(
movie = await movie_metadata_service.movie_repository.get_movie_by_external_id(
external_id=movie_id, metadata_provider=metadata_provider.name
)
if not movie:
@@ -167,13 +167,13 @@ def add_a_movie(
"/torrents",
dependencies=[Depends(current_active_user)],
)
def get_all_movies_with_torrents(
async def get_all_movies_with_torrents(
movie_service: movie_service_dep,
) -> list[RichMovieTorrent]:
"""
Get all movies that are associated with torrents.
"""
return movie_service.get_all_movies_with_torrents()
return await movie_service.get_all_movies_with_torrents()
@router.get(
@@ -196,11 +196,13 @@ def get_available_libraries() -> list[LibraryItem]:
"/{movie_id}",
dependencies=[Depends(current_active_user)],
)
def get_movie_by_id(movie_service: movie_service_dep, movie: movie_dep) -> PublicMovie:
async def get_movie_by_id(
movie_service: movie_service_dep, movie: movie_dep
) -> PublicMovie:
"""
Get details for a specific movie.
"""
return movie_service.get_public_movie_by_id(movie=movie)
return await movie_service.get_public_movie_by_id(movie=movie)
@router.delete(
@@ -208,7 +210,7 @@ def get_movie_by_id(movie_service: movie_service_dep, movie: movie_dep) -> Publi
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(current_superuser)],
)
def delete_a_movie(
async def delete_a_movie(
movie_service: movie_service_dep,
movie: movie_dep,
delete_files_on_disk: bool = False,
@@ -217,7 +219,7 @@ def delete_a_movie(
"""
Delete a movie from the library.
"""
movie_service.delete_movie(
await movie_service.delete_movie(
movie=movie,
delete_files_on_disk=delete_files_on_disk,
delete_torrents=delete_torrents,
@@ -229,7 +231,7 @@ def delete_a_movie(
dependencies=[Depends(current_superuser)],
status_code=status.HTTP_204_NO_CONTENT,
)
def set_library(
async def set_library(
movie: movie_dep,
movie_service: movie_service_dep,
library: str,
@@ -237,7 +239,7 @@ def set_library(
"""
Set the library path for a Movie.
"""
movie_service.set_movie_library(movie=movie, library=library)
await movie_service.set_movie_library(movie=movie, library=library)
return
@@ -245,20 +247,20 @@ def set_library(
"/{movie_id}/files",
dependencies=[Depends(current_active_user)],
)
def get_movie_files_by_movie_id(
async def get_movie_files_by_movie_id(
movie_service: movie_service_dep, movie: movie_dep
) -> list[PublicMovieFile]:
"""
Get files associated with a specific movie.
"""
return movie_service.get_public_movie_files(movie=movie)
return await movie_service.get_public_movie_files(movie=movie)
@router.get(
"/{movie_id}/torrents",
dependencies=[Depends(current_active_user)],
)
def search_for_torrents_for_movie(
async def search_for_torrents_for_movie(
movie_service: movie_service_dep,
movie: movie_dep,
search_query_override: str | None = None,
@@ -266,7 +268,7 @@ def search_for_torrents_for_movie(
"""
Search for torrents for a specific movie.
"""
return movie_service.get_all_available_torrents_for_movie(
return await movie_service.get_all_available_torrents_for_movie(
movie=movie, search_query_override=search_query_override
)
@@ -276,7 +278,7 @@ def search_for_torrents_for_movie(
status_code=status.HTTP_201_CREATED,
dependencies=[Depends(current_active_user)],
)
def download_torrent_for_movie(
async def download_torrent_for_movie(
movie_service: movie_service_dep,
movie: movie_dep,
public_indexer_result_id: IndexerQueryResultId,
@@ -285,7 +287,7 @@ def download_torrent_for_movie(
"""
Trigger a download for a specific torrent for a movie.
"""
return movie_service.download_torrent(
return await movie_service.download_torrent(
public_indexer_result_id=public_indexer_result_id,
movie=movie,
override_movie_file_path_suffix=override_file_path_suffix,

View File

@@ -1,3 +1,4 @@
import asyncio
import logging
import shutil
from pathlib import Path
@@ -49,7 +50,7 @@ class MovieService(BaseMediaService[Movie, Movie]):
self.movie_import_service = movie_import_service
self.movie_metadata_service = movie_metadata_service
def delete_movie(
async def delete_movie(
self,
movie: Movie,
delete_files_on_disk: bool = False,
@@ -69,23 +70,23 @@ class MovieService(BaseMediaService[Movie, Movie]):
if movie_dir.exists() and movie_dir.is_dir():
try:
shutil.rmtree(movie_dir)
await asyncio.to_thread(shutil.rmtree, movie_dir)
log.info(f"Deleted movie directory: {movie_dir}")
except OSError:
log.exception(f"Deleting movie directory: {movie_dir}")
if delete_torrents:
# Get all torrents associated with this movie
movie_torrents = self.movie_repository.get_torrents_by_movie_id(
movie_torrents = await self.movie_repository.get_torrents_by_movie_id(
movie_id=movie.id
)
for movie_torrent in movie_torrents:
torrent = self.torrent_service.get_torrent_by_id(
torrent = await self.torrent_service.get_torrent_by_id(
torrent_id=movie_torrent.torrent_id
)
try:
self.torrent_service.cancel_download(
await self.torrent_service.cancel_download(
torrent=torrent, delete_files=True
)
log.info(f"Deleted torrent: {torrent.torrent_title}")
@@ -93,26 +94,26 @@ class MovieService(BaseMediaService[Movie, Movie]):
log.exception(f"Failed to delete torrent {torrent.hash}")
# Delete from database
self.movie_repository.delete_movie(movie.id)
await self.movie_repository.delete_movie(movie.id)
def get_public_movie_files(self, movie: Movie) -> list[PublicMovieFile]:
async def get_public_movie_files(self, movie: Movie) -> list[PublicMovieFile]:
"""
Get all public movie files for a given movie.
:param movie: The movie object.
:return: A list of public movie files.
"""
movie_files = self.movie_repository.get_movie_files_by_movie_id(
movie_files = await self.movie_repository.get_movie_files_by_movie_id(
movie_id=movie.id
)
public_movie_files = [PublicMovieFile.model_validate(x) for x in movie_files]
result = []
for movie_file in public_movie_files:
movie_file.imported = self.movie_file_exists_on_file(movie_file=movie_file)
movie_file.imported = await self.movie_file_exists_on_file(movie_file=movie_file)
result.append(movie_file)
return result
def get_all_available_torrents_for_movie(
async def get_all_available_torrents_for_movie(
self, movie: Movie, search_query_override: str | None = None
) -> list[IndexerQueryResult]:
"""
@@ -123,52 +124,52 @@ class MovieService(BaseMediaService[Movie, Movie]):
:return: A list of indexer query results.
"""
if search_query_override:
return self.indexer_service.search(query=search_query_override, is_tv=False)
return await self.indexer_service.search(query=search_query_override, is_tv=False)
torrents = self.indexer_service.search_movie(movie=movie)
torrents = await self.indexer_service.search_movie(movie=movie)
return evaluate_indexer_query_results(
is_tv=False, query_results=torrents, media=movie
)
def get_public_movie_by_id(self, movie: Movie) -> PublicMovie:
async def get_public_movie_by_id(self, movie: Movie) -> PublicMovie:
"""
Get a public movie from a Movie object.
:param movie: The movie object.
:return: A public movie.
"""
torrents = self.get_torrents_for_movie(movie=movie).torrents
torrents = (await self.get_torrents_for_movie(movie=movie)).torrents
public_movie = PublicMovie.model_validate(movie)
public_movie.downloaded = self.is_movie_downloaded(movie=movie)
public_movie.downloaded = await self.is_movie_downloaded(movie=movie)
public_movie.torrents = torrents
return public_movie
def get_movie_by_id(self, movie_id: MovieId) -> Movie:
async def get_movie_by_id(self, movie_id: MovieId) -> Movie:
"""
Get a movie by its ID.
:param movie_id: The ID of the movie.
:return: The movie.
"""
return self.movie_repository.get_movie_by_id(movie_id)
return await self.movie_repository.get_movie_by_id(movie_id)
def is_movie_downloaded(self, movie: Movie) -> bool:
async def is_movie_downloaded(self, movie: Movie) -> bool:
"""
Check if a movie is downloaded.
:param movie: The movie object.
:return: True if the movie is downloaded, False otherwise.
"""
movie_files = self.movie_repository.get_movie_files_by_movie_id(
movie_files = await self.movie_repository.get_movie_files_by_movie_id(
movie_id=movie.id
)
for movie_file in movie_files:
if self.movie_file_exists_on_file(movie_file=movie_file):
if await self.movie_file_exists_on_file(movie_file=movie_file):
return True
return False
def movie_file_exists_on_file(self, movie_file: MovieFile) -> bool:
async def movie_file_exists_on_file(self, movie_file: MovieFile) -> bool:
"""
Check if a movie file exists on the filesystem.
@@ -177,12 +178,12 @@ class MovieService(BaseMediaService[Movie, Movie]):
"""
if movie_file.torrent_id is None:
return True
torrent_file = self.torrent_service.get_torrent_by_id(
torrent_file = await self.torrent_service.get_torrent_by_id(
torrent_id=movie_file.torrent_id
)
return bool(torrent_file.imported)
def get_movie_by_external_id(
async def get_movie_by_external_id(
self, external_id: int, metadata_provider: str
) -> Movie | None:
"""
@@ -192,27 +193,27 @@ class MovieService(BaseMediaService[Movie, Movie]):
:param metadata_provider: The metadata provider.
:return: The movie or None if not found.
"""
return self.movie_repository.get_movie_by_external_id(
return await self.movie_repository.get_movie_by_external_id(
external_id=external_id, metadata_provider=metadata_provider
)
def set_movie_library(self, movie: Movie, library: str) -> None:
self.movie_repository.set_movie_library(movie.id, library)
async def set_movie_library(self, movie: Movie, library: str) -> None:
await self.movie_repository.set_movie_library(movie.id, library)
def get_all_movies(self) -> list[Movie]:
async def get_all_movies(self) -> list[Movie]:
"""
Get all movies in the library.
"""
return self.get_all_media()
return await self.get_all_media()
def get_torrents_for_movie(self, movie: Movie) -> RichMovieTorrent:
async def get_torrents_for_movie(self, movie: Movie) -> RichMovieTorrent:
"""
Get torrents for a given movie.
:param movie: The movie.
:return: A rich movie torrent.
"""
movie_torrents = self.movie_repository.get_torrents_by_movie_id(
movie_torrents = await self.movie_repository.get_torrents_by_movie_id(
movie_id=movie.id
)
return RichMovieTorrent(
@@ -223,16 +224,16 @@ class MovieService(BaseMediaService[Movie, Movie]):
torrents=movie_torrents,
)
def get_all_movies_with_torrents(self) -> list[RichMovieTorrent]:
async def get_all_movies_with_torrents(self) -> list[RichMovieTorrent]:
"""
Get all movies with torrents.
:return: A list of rich movie torrents.
"""
movies = self.movie_repository.get_all_movies_with_torrents()
return [self.get_torrents_for_movie(movie=movie) for movie in movies]
movies = await self.movie_repository.get_all_movies_with_torrents()
return [await self.get_torrents_for_movie(movie=movie) for movie in movies]
def download_torrent(
async def download_torrent(
self,
public_indexer_result_id: IndexerQueryResultId,
movie: Movie,
@@ -246,11 +247,11 @@ class MovieService(BaseMediaService[Movie, Movie]):
:param override_movie_file_path_suffix: Optional override for the file path suffix.
:return: The downloaded torrent.
"""
indexer_result = self.indexer_service.get_result(
indexer_result = await self.indexer_service.get_result(
result_id=public_indexer_result_id
)
movie_torrent = self.torrent_service.download(indexer_result=indexer_result)
self.torrent_service.pause_download(torrent=movie_torrent)
movie_torrent = await self.torrent_service.download(indexer_result=indexer_result)
await self.torrent_service.pause_download(torrent=movie_torrent)
movie_file = MovieFile(
movie_id=movie.id,
quality=indexer_result.quality,
@@ -258,12 +259,12 @@ class MovieService(BaseMediaService[Movie, Movie]):
file_path_suffix=override_movie_file_path_suffix,
)
try:
self.movie_repository.add_movie_file(movie_file=movie_file)
await self.movie_repository.add_movie_file(movie_file=movie_file)
except IntegrityError:
log.warning(
f"Movie file for movie {movie.name} and torrent {movie_torrent.title} already exists"
)
self.torrent_service.cancel_download(
await self.torrent_service.cancel_download(
torrent=movie_torrent, delete_files=True
)
raise
@@ -271,7 +272,7 @@ class MovieService(BaseMediaService[Movie, Movie]):
log.info(
f"Added movie file for movie {movie.name} and torrent {movie_torrent.title}"
)
self.torrent_service.resume_download(torrent=movie_torrent)
await self.torrent_service.resume_download(torrent=movie_torrent)
return movie_torrent
def get_movie_root_path(self, movie: Movie) -> Path:
@@ -282,14 +283,14 @@ class MovieService(BaseMediaService[Movie, Movie]):
libraries=misc_config.movie_libraries,
)
def import_all_torrents(self) -> None:
async def import_all_torrents(self) -> None:
"""
Delegate to MovieImportService.
"""
self.movie_import_service.import_all_torrents()
await self.movie_import_service.import_all_torrents()
def update_all_metadata(self) -> None:
async def update_all_metadata(self) -> None:
"""
Delegate to MovieMetadataService.
"""
self.movie_metadata_service.update_all_metadata()
await self.movie_metadata_service.update_all_metadata()

View File

@@ -70,7 +70,7 @@ class NotificationManager:
logger.info(f"Initialized {len(self.providers)} notification providers")
def send_notification(self, title: str, message: str) -> None:
async def send_notification(self, title: str, message: str) -> None:
if not self.providers:
logger.warning("No notification providers configured")
@@ -79,7 +79,7 @@ class NotificationManager:
for provider in self.providers:
provider_name = provider.__class__.__name__
try:
success = provider.send_notification(notification)
success = await provider.send_notification(notification)
if success:
logger.info(f"Notification sent successfully via {provider_name}")
else:

View File

@@ -5,7 +5,7 @@ from sqlalchemy.exc import (
IntegrityError,
SQLAlchemyError,
)
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.expression import false
from media_manager.exceptions import ConflictError, NotFoundError
@@ -21,11 +21,11 @@ log = logging.getLogger(__name__)
class NotificationRepository:
def __init__(self, db: Session) -> None:
def __init__(self, db: AsyncSession) -> None:
self.db = db
def get_notification(self, nid: NotificationId) -> NotificationSchema:
result = self.db.get(Notification, nid)
async def get_notification(self, nid: NotificationId) -> NotificationSchema:
result = await self.db.get(Notification, nid)
if not result:
msg = f"Notification with id {nid} not found."
@@ -33,14 +33,14 @@ class NotificationRepository:
return NotificationSchema.model_validate(result)
def get_unread_notifications(self) -> list[NotificationSchema]:
async def get_unread_notifications(self) -> list[NotificationSchema]:
try:
stmt = (
select(Notification)
.where(Notification.read == false())
.order_by(Notification.timestamp.desc())
)
results = self.db.execute(stmt).scalars().all()
results = (await self.db.execute(stmt)).scalars().all()
return [
NotificationSchema.model_validate(notification)
for notification in results
@@ -49,10 +49,10 @@ class NotificationRepository:
log.exception("Database error while retrieving unread notifications")
raise
def get_all_notifications(self) -> list[NotificationSchema]:
async def get_all_notifications(self) -> list[NotificationSchema]:
try:
stmt = select(Notification).order_by(Notification.timestamp.desc())
results = self.db.execute(stmt).scalars().all()
results = (await self.db.execute(stmt)).scalars().all()
return [
NotificationSchema.model_validate(notification)
for notification in results
@@ -61,7 +61,7 @@ class NotificationRepository:
log.exception("Database error while retrieving notifications")
raise
def save_notification(self, notification: NotificationSchema) -> None:
async def save_notification(self, notification: NotificationSchema) -> None:
try:
self.db.add(
Notification(
@@ -71,28 +71,31 @@ class NotificationRepository:
message=notification.message,
)
)
self.db.commit()
await self.db.commit()
except IntegrityError:
# AsyncSession leaves the txn in invalid state on IntegrityError;
# without rollback the request-end commit raises PendingRollbackError.
await self.db.rollback()
log.exception("Could not save notification")
msg = f"Notification with id {notification.id} already exists."
raise ConflictError(msg) from None
return
def mark_notification_as_read(self, nid: NotificationId) -> None:
async def mark_notification_as_read(self, nid: NotificationId) -> None:
stmt = update(Notification).where(Notification.id == nid).values(read=True)
self.db.execute(stmt)
await self.db.execute(stmt)
return
def mark_notification_as_unread(self, nid: NotificationId) -> None:
async def mark_notification_as_unread(self, nid: NotificationId) -> None:
stmt = update(Notification).where(Notification.id == nid).values(read=False)
self.db.execute(stmt)
await self.db.execute(stmt)
return
def delete_notification(self, nid: NotificationId) -> None:
async def delete_notification(self, nid: NotificationId) -> None:
stmt = delete(Notification).where(Notification.id == nid)
result = self.db.execute(stmt)
result = await self.db.execute(stmt)
if result.rowcount == 0:
msg = f"Notification with id {nid} not found."
raise NotFoundError(msg)
self.db.commit()
await self.db.commit()
return

View File

@@ -16,26 +16,26 @@ router = APIRouter()
"",
dependencies=[Depends(current_active_user)],
)
def get_all_notifications(
async def get_all_notifications(
notification_service: notification_service_dep,
) -> list[Notification]:
"""
Get all notifications.
"""
return notification_service.get_all_notifications()
return await notification_service.get_all_notifications()
@router.get(
"/unread",
dependencies=[Depends(current_active_user)],
)
def get_unread_notifications(
async def get_unread_notifications(
notification_service: notification_service_dep,
) -> list[Notification]:
"""
Get all unread notifications.
"""
return notification_service.get_unread_notifications()
return await notification_service.get_unread_notifications()
@router.get(
@@ -45,13 +45,13 @@ def get_unread_notifications(
status.HTTP_404_NOT_FOUND: {"description": "Notification not found"},
},
)
def get_notification(
async def get_notification(
notification_id: NotificationId, notification_service: notification_service_dep
) -> Notification:
"""
Get a specific notification by ID.
"""
return notification_service.get_notification(nid=notification_id)
return await notification_service.get_notification(nid=notification_id)
# --------------------------------
@@ -67,13 +67,13 @@ def get_notification(
status.HTTP_404_NOT_FOUND: {"description": "Notification not found"},
},
)
def mark_notification_as_read(
async def mark_notification_as_read(
notification_id: NotificationId, notification_service: notification_service_dep
) -> None:
"""
Mark a notification as read.
"""
notification_service.mark_notification_as_read(nid=notification_id)
await notification_service.mark_notification_as_read(nid=notification_id)
@router.patch(
@@ -84,13 +84,13 @@ def mark_notification_as_read(
status.HTTP_404_NOT_FOUND: {"description": "Notification not found"},
},
)
def mark_notification_as_unread(
async def mark_notification_as_unread(
notification_id: NotificationId, notification_service: notification_service_dep
) -> None:
"""
Mark a notification as unread.
"""
notification_service.mark_notification_as_unread(nid=notification_id)
await notification_service.mark_notification_as_unread(nid=notification_id)
@router.delete(
@@ -101,10 +101,10 @@ def mark_notification_as_unread(
status.HTTP_404_NOT_FOUND: {"description": "Notification not found"},
},
)
def delete_notification(
async def delete_notification(
notification_id: NotificationId, notification_service: notification_service_dep
) -> None:
"""
Delete a notification.
"""
notification_service.delete_notification(nid=notification_id)
await notification_service.delete_notification(nid=notification_id)

View File

@@ -11,30 +11,30 @@ class NotificationService:
self.notification_repository = notification_repository
self.notification_manager = notification_manager
def get_notification(self, nid: NotificationId) -> Notification:
return self.notification_repository.get_notification(nid=nid)
async def get_notification(self, nid: NotificationId) -> Notification:
return await self.notification_repository.get_notification(nid=nid)
def get_unread_notifications(self) -> list[Notification]:
return self.notification_repository.get_unread_notifications()
async def get_unread_notifications(self) -> list[Notification]:
return await self.notification_repository.get_unread_notifications()
def get_all_notifications(self) -> list[Notification]:
return self.notification_repository.get_all_notifications()
async def get_all_notifications(self) -> list[Notification]:
return await self.notification_repository.get_all_notifications()
def save_notification(self, notification: Notification) -> None:
return self.notification_repository.save_notification(notification)
async def save_notification(self, notification: Notification) -> None:
return await self.notification_repository.save_notification(notification)
def mark_notification_as_read(self, nid: NotificationId) -> None:
return self.notification_repository.mark_notification_as_read(nid=nid)
async def mark_notification_as_read(self, nid: NotificationId) -> None:
return await self.notification_repository.mark_notification_as_read(nid=nid)
def mark_notification_as_unread(self, nid: NotificationId) -> None:
return self.notification_repository.mark_notification_as_unread(nid=nid)
async def mark_notification_as_unread(self, nid: NotificationId) -> None:
return await self.notification_repository.mark_notification_as_unread(nid=nid)
def delete_notification(self, nid: NotificationId) -> None:
return self.notification_repository.delete_notification(nid=nid)
async def delete_notification(self, nid: NotificationId) -> None:
return await self.notification_repository.delete_notification(nid=nid)
def send_notification_to_all_providers(self, title: str, message: str) -> None:
self.notification_manager.send_notification(title, message)
async def send_notification_to_all_providers(self, title: str, message: str) -> None:
await self.notification_manager.send_notification(title, message)
internal_notification = Notification(message=f"{title}: {message}", read=False)
self.save_notification(internal_notification)
await self.save_notification(internal_notification)
return

View File

@@ -5,7 +5,7 @@ from media_manager.notification.schemas import MessageNotification
class AbstractNotificationServiceProvider(ABC):
@abstractmethod
def send_notification(self, message: MessageNotification) -> bool:
async def send_notification(self, message: MessageNotification) -> bool:
"""
Sends a notification with the given message.

View File

@@ -1,3 +1,5 @@
import asyncio
import media_manager.notification.utils
from media_manager.config import MediaManagerConfig
from media_manager.notification.schemas import MessageNotification
@@ -10,7 +12,7 @@ class EmailNotificationServiceProvider(AbstractNotificationServiceProvider):
def __init__(self) -> None:
self.config = MediaManagerConfig().notifications.email_notifications
def send_notification(self, message: MessageNotification) -> bool:
async def send_notification(self, message: MessageNotification) -> bool:
subject = "MediaManager - " + message.title
html = f"""\
<html>
@@ -25,8 +27,11 @@ class EmailNotificationServiceProvider(AbstractNotificationServiceProvider):
"""
for email in self.config.emails:
media_manager.notification.utils.send_email(
subject=subject, html=html, addressee=email
await asyncio.to_thread(
media_manager.notification.utils.send_email,
subject=subject,
html=html,
addressee=email,
)
return True

View File

@@ -1,4 +1,4 @@
import requests
import httpx
from media_manager.config import MediaManagerConfig
from media_manager.notification.schemas import MessageNotification
@@ -15,15 +15,15 @@ class GotifyNotificationServiceProvider(AbstractNotificationServiceProvider):
def __init__(self) -> None:
self.config = MediaManagerConfig().notifications.gotify
def send_notification(self, message: MessageNotification) -> bool:
response = requests.post(
url=f"{self.config.url}/message?token={self.config.api_key}",
json={
"message": message.message,
"title": message.title,
},
timeout=60,
)
async def send_notification(self, message: MessageNotification) -> bool:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
url=f"{self.config.url}/message?token={self.config.api_key}",
json={
"message": message.message,
"title": message.title,
},
)
if response.status_code not in range(200, 300):
return False
return True

View File

@@ -1,4 +1,4 @@
import requests
import httpx
from media_manager.config import MediaManagerConfig
from media_manager.notification.schemas import MessageNotification
@@ -15,15 +15,15 @@ class NtfyNotificationServiceProvider(AbstractNotificationServiceProvider):
def __init__(self) -> None:
self.config = MediaManagerConfig().notifications.ntfy
def send_notification(self, message: MessageNotification) -> bool:
response = requests.post(
url=self.config.url,
data=message.message.encode(encoding="utf-8"),
headers={
"Title": "MediaManager - " + message.title,
},
timeout=60,
)
async def send_notification(self, message: MessageNotification) -> bool:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
url=self.config.url,
content=message.message.encode(encoding="utf-8"),
headers={
"Title": "MediaManager - " + message.title,
},
)
if response.status_code not in range(200, 300):
return False
return True

View File

@@ -1,4 +1,4 @@
import requests
import httpx
from media_manager.config import MediaManagerConfig
from media_manager.notification.schemas import MessageNotification
@@ -11,17 +11,17 @@ class PushoverNotificationServiceProvider(AbstractNotificationServiceProvider):
def __init__(self) -> None:
self.config = MediaManagerConfig().notifications.pushover
def send_notification(self, message: MessageNotification) -> bool:
response = requests.post(
url="https://api.pushover.net/1/messages.json",
params={
"token": self.config.api_key,
"user": self.config.user,
"message": message.message,
"title": "MediaManager - " + message.title,
},
timeout=60,
)
async def send_notification(self, message: MessageNotification) -> bool:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
url="https://api.pushover.net/1/messages.json",
params={
"token": self.config.api_key,
"user": self.config.user,
"message": message.message,
"title": "MediaManager - " + message.title,
},
)
if response.status_code not in range(200, 300):
return False
return True

View File

@@ -1,4 +1,3 @@
import asyncio
import logging
from urllib.parse import quote
@@ -44,7 +43,7 @@ async def import_all_movie_torrents_task(
movie_service: MovieService = TaskiqDepends(get_movie_service),
) -> None:
log.info("Importing all Movie torrents")
await asyncio.to_thread(movie_service.import_all_torrents)
await movie_service.import_all_torrents()
@broker.task
@@ -52,21 +51,21 @@ async def import_all_show_torrents_task(
tv_service: TvService = TaskiqDepends(get_tv_service),
) -> None:
log.info("Importing all Show torrents")
await asyncio.to_thread(tv_service.import_all_torrents)
await tv_service.import_all_torrents()
@broker.task
async def update_all_movies_metadata_task(
movie_service: MovieService = TaskiqDepends(get_movie_service),
) -> None:
await asyncio.to_thread(movie_service.update_all_metadata)
await movie_service.update_all_metadata()
@broker.task
async def update_all_non_ended_shows_metadata_task(
tv_service: TvService = TaskiqDepends(get_tv_service),
) -> None:
await asyncio.to_thread(tv_service.update_all_non_ended_shows_metadata)
await tv_service.update_all_non_ended_shows_metadata()
# Maps each task to its cron schedule so PostgresqlSchedulerSource can seed

View File

@@ -24,7 +24,7 @@ def get_torrent_service(torrent_repository: torrent_repository_dep) -> TorrentSe
torrent_service_dep = Annotated[TorrentService, Depends(get_torrent_service)]
def get_torrent_by_id(
async def get_torrent_by_id(
torrent_service: torrent_service_dep, torrent_id: TorrentId
) -> Torrent:
"""
@@ -35,7 +35,7 @@ def get_torrent_by_id(
:return: The TorrentService instance with the specified torrent.
"""
try:
torrent = torrent_service.get_torrent_by_id(torrent_id=torrent_id)
torrent = await torrent_service.get_torrent_by_id(torrent_id=torrent_id)
except NotFoundError:
raise HTTPException(
status_code=404, detail=f"Torrent with ID {torrent_id} not found"

View File

@@ -1,4 +1,5 @@
from sqlalchemy import delete, select
from sqlalchemy.orm import selectinload
from media_manager.database import DbSessionDependency
from media_manager.exceptions import NotFoundError
@@ -17,78 +18,83 @@ class TorrentRepository:
def __init__(self, db: DbSessionDependency) -> None:
self.db = db
def get_episode_files_of_torrent(
async def get_episode_files_of_torrent(
self, torrent_id: TorrentId
) -> list[EpisodeFileSchema]:
stmt = select(EpisodeFile).where(EpisodeFile.torrent_id == torrent_id)
result = self.db.execute(stmt).scalars().all()
result = (await self.db.execute(stmt)).scalars().all()
return [
EpisodeFileSchema.model_validate(episode_file) for episode_file in result
]
def get_show_of_torrent(self, torrent_id: TorrentId) -> ShowSchema | None:
async def get_show_of_torrent(self, torrent_id: TorrentId) -> ShowSchema | None:
# Eager-load the show tree; ShowSchema requires seasons -> episodes and
# AsyncSession can't satisfy implicit lazy loads during validation.
stmt = (
select(Show)
.join(Show.seasons)
.join(Season.episodes)
.join(Episode.episode_files)
.where(EpisodeFile.torrent_id == torrent_id)
.options(selectinload(Show.seasons).selectinload(Season.episodes))
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if result is None:
return None
return ShowSchema.model_validate(result)
def save_torrent(self, torrent: TorrentSchema) -> TorrentSchema:
self.db.merge(Torrent(**torrent.model_dump()))
self.db.commit()
async def save_torrent(self, torrent: TorrentSchema) -> TorrentSchema:
await self.db.merge(Torrent(**torrent.model_dump()))
await self.db.commit()
return TorrentSchema.model_validate(torrent)
def get_all_torrents(self) -> list[TorrentSchema]:
async def get_all_torrents(self) -> list[TorrentSchema]:
stmt = select(Torrent)
result = self.db.execute(stmt).scalars().all()
result = (await self.db.execute(stmt)).scalars().all()
return [
TorrentSchema.model_validate(torrent_schema) for torrent_schema in result
]
def get_torrent_by_id(self, torrent_id: TorrentId) -> TorrentSchema:
result = self.db.get(Torrent, torrent_id)
async def get_torrent_by_id(self, torrent_id: TorrentId) -> TorrentSchema:
result = await self.db.get(Torrent, torrent_id)
if result is None:
msg = f"Torrent with ID {torrent_id} not found."
raise NotFoundError(msg)
return TorrentSchema.model_validate(result)
def delete_torrent(
async def delete_torrent(
self, torrent_id: TorrentId, delete_associated_media_files: bool = False
) -> None:
if delete_associated_media_files:
movie_files_stmt = delete(MovieFile).where(
MovieFile.torrent_id == torrent_id
)
self.db.execute(movie_files_stmt)
await self.db.execute(movie_files_stmt)
episode_files_stmt = delete(EpisodeFile).where(
EpisodeFile.torrent_id == torrent_id
)
self.db.execute(episode_files_stmt)
await self.db.execute(episode_files_stmt)
self.db.delete(self.db.get(Torrent, torrent_id))
obj = await self.db.get(Torrent, torrent_id)
if obj is not None:
await self.db.delete(obj)
def get_movie_of_torrent(self, torrent_id: TorrentId) -> MovieSchema | None:
async def get_movie_of_torrent(self, torrent_id: TorrentId) -> MovieSchema | None:
stmt = (
select(Movie)
.join(MovieFile, Movie.id == MovieFile.movie_id)
.where(MovieFile.torrent_id == torrent_id)
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if result is None:
return None
return MovieSchema.model_validate(result)
def get_movie_files_of_torrent(
async def get_movie_files_of_torrent(
self, torrent_id: TorrentId
) -> list[MovieFileSchema]:
stmt = select(MovieFile).where(MovieFile.torrent_id == torrent_id)
result = self.db.execute(stmt).scalars().all()
result = (await self.db.execute(stmt)).scalars().all()
return [MovieFileSchema.model_validate(movie_file) for movie_file in result]

View File

@@ -18,13 +18,13 @@ router = APIRouter()
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_active_user)],
)
def get_all_torrents(service: torrent_service_dep) -> list[Torrent]:
return service.get_all_torrents()
async def get_all_torrents(service: torrent_service_dep) -> list[Torrent]:
return await service.get_all_torrents()
@router.get("/{torrent_id}", status_code=status.HTTP_200_OK)
def get_torrent(service: torrent_service_dep, torrent: torrent_dep) -> Torrent:
return service.get_torrent_by_id(torrent_id=torrent.id)
async def get_torrent(service: torrent_service_dep, torrent: torrent_dep) -> Torrent:
return await service.get_torrent_by_id(torrent_id=torrent.id)
@router.delete(
@@ -32,18 +32,18 @@ def get_torrent(service: torrent_service_dep, torrent: torrent_dep) -> Torrent:
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(current_superuser)],
)
def delete_torrent(
async def delete_torrent(
service: torrent_service_dep,
torrent: torrent_dep,
delete_files: bool = False,
) -> None:
if delete_files:
try:
service.cancel_download(torrent=torrent, delete_files=False)
await service.cancel_download(torrent=torrent, delete_files=False)
except RuntimeError:
pass
service.delete_torrent(torrent_id=torrent.id)
await service.delete_torrent(torrent_id=torrent.id)
@router.post(
@@ -51,12 +51,12 @@ def delete_torrent(
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(current_superuser)],
)
def retry_torrent_download(
async def retry_torrent_download(
service: torrent_service_dep,
torrent: torrent_dep,
) -> None:
service.pause_download(torrent=torrent)
service.resume_download(torrent=torrent)
await service.pause_download(torrent=torrent)
await service.resume_download(torrent=torrent)
@router.patch(
@@ -64,7 +64,7 @@ def retry_torrent_download(
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_superuser)],
)
def update_torrent_status(
async def update_torrent_status(
rep: torrent_repository_dep,
torrent: torrent_dep,
state: TorrentStatus | None = None,
@@ -80,5 +80,5 @@ def update_torrent_status(
detail="No status or imported value provided",
)
rep.save_torrent(torrent=torrent)
await rep.save_torrent(torrent=torrent)
return torrent

View File

@@ -1,3 +1,4 @@
import asyncio
import logging
from media_manager.indexer.schemas import IndexerQueryResult
@@ -19,44 +20,50 @@ class TorrentService:
self.torrent_repository = torrent_repository
self.download_manager = download_manager or DownloadManager()
def get_episode_files_of_torrent(self, torrent: Torrent) -> list[EpisodeFile]:
async def get_episode_files_of_torrent(
self, torrent: Torrent
) -> list[EpisodeFile]:
"""
Returns all episode files of a torrent
:param torrent: the torrent to get the episode files of
:return: list of episode files
"""
return self.torrent_repository.get_episode_files_of_torrent(
return await self.torrent_repository.get_episode_files_of_torrent(
torrent_id=torrent.id
)
def get_show_of_torrent(self, torrent: Torrent) -> Show | None:
async def get_show_of_torrent(self, torrent: Torrent) -> Show | None:
"""
Returns the show of a torrent
:param torrent: the torrent to get the show of
:return: the show of the torrent
"""
return self.torrent_repository.get_show_of_torrent(torrent_id=torrent.id)
return await self.torrent_repository.get_show_of_torrent(torrent_id=torrent.id)
def get_movie_of_torrent(self, torrent: Torrent) -> Movie | None:
async def get_movie_of_torrent(self, torrent: Torrent) -> Movie | None:
"""
Returns the movie of a torrent
:param torrent: the torrent to get the movie of
:return: the movie of the torrent
"""
return self.torrent_repository.get_movie_of_torrent(torrent_id=torrent.id)
return await self.torrent_repository.get_movie_of_torrent(torrent_id=torrent.id)
def download(self, indexer_result: IndexerQueryResult) -> Torrent:
async def download(self, indexer_result: IndexerQueryResult) -> Torrent:
log.info(f"Starting download for torrent: {indexer_result.title}")
torrent = self.download_manager.download(indexer_result)
torrent = await asyncio.to_thread(self.download_manager.download, indexer_result)
return self.torrent_repository.save_torrent(torrent=torrent)
return await self.torrent_repository.save_torrent(torrent=torrent)
def get_torrent_status(self, torrent: Torrent) -> Torrent:
torrent.status = self.download_manager.get_torrent_status(torrent)
self.torrent_repository.save_torrent(torrent=torrent)
async def get_torrent_status(self, torrent: Torrent) -> Torrent:
torrent.status = await asyncio.to_thread(
self.download_manager.get_torrent_status, torrent
)
await self.torrent_repository.save_torrent(torrent=torrent)
return torrent
def cancel_download(self, torrent: Torrent, delete_files: bool = False) -> Torrent:
async def cancel_download(
self, torrent: Torrent, delete_files: bool = False
) -> Torrent:
"""
cancels download of a torrent
@@ -64,57 +71,62 @@ class TorrentService:
:param torrent: the torrent to cancel
"""
log.info(f"Cancelling download for torrent: {torrent.title}")
self.download_manager.remove_torrent(torrent, delete_data=delete_files)
return self.get_torrent_status(torrent=torrent)
await asyncio.to_thread(
self.download_manager.remove_torrent, torrent, delete_data=delete_files
)
return await self.get_torrent_status(torrent=torrent)
def pause_download(self, torrent: Torrent) -> Torrent:
async def pause_download(self, torrent: Torrent) -> Torrent:
"""
pauses download of a torrent
:param torrent: the torrent to pause
"""
log.info(f"Pausing download for torrent: {torrent.title}")
self.download_manager.pause_torrent(torrent)
return self.get_torrent_status(torrent=torrent)
await asyncio.to_thread(self.download_manager.pause_torrent, torrent)
return await self.get_torrent_status(torrent=torrent)
def resume_download(self, torrent: Torrent) -> Torrent:
async def resume_download(self, torrent: Torrent) -> Torrent:
"""
resumes download of a torrent
:param torrent: the torrent to resume
"""
log.info(f"Resuming download for torrent: {torrent.title}")
self.download_manager.resume_torrent(torrent)
return self.get_torrent_status(torrent=torrent)
await asyncio.to_thread(self.download_manager.resume_torrent, torrent)
return await self.get_torrent_status(torrent=torrent)
def get_all_torrents(self) -> list[Torrent]:
torrents = []
for x in self.torrent_repository.get_all_torrents():
async def get_all_torrents(self) -> list[Torrent]:
all_torrents = await self.torrent_repository.get_all_torrents()
torrents: list[Torrent] = []
for t in all_torrents:
try:
torrents.append(self.get_torrent_status(x))
except RuntimeError:
log.exception(f"Error fetching status for torrent {x.title}")
torrents.append(await self.get_torrent_status(t))
except Exception:
log.exception(f"Error fetching status for torrent {t.title}")
return torrents
def get_completed_torrents(self) -> list[Torrent]:
async def get_completed_torrents(self) -> list[Torrent]:
return [
t
for t in self.get_all_torrents()
for t in await self.get_all_torrents()
if t.status == TorrentStatus.finished and not t.imported
]
def get_torrent_by_id(self, torrent_id: TorrentId) -> Torrent:
return self.get_torrent_status(
self.torrent_repository.get_torrent_by_id(torrent_id=torrent_id)
async def get_torrent_by_id(self, torrent_id: TorrentId) -> Torrent:
return await self.get_torrent_status(
await self.torrent_repository.get_torrent_by_id(torrent_id=torrent_id)
)
def delete_torrent(self, torrent_id: TorrentId) -> None:
async def delete_torrent(self, torrent_id: TorrentId) -> None:
log.info(f"Deleting torrent with ID: {torrent_id}")
t = self.torrent_repository.get_torrent_by_id(torrent_id=torrent_id)
t = await self.torrent_repository.get_torrent_by_id(torrent_id=torrent_id)
delete_media_files = not t.imported
self.torrent_repository.delete_torrent(
await self.torrent_repository.delete_torrent(
torrent_id=torrent_id, delete_associated_media_files=delete_media_files
)
def get_movie_files_of_torrent(self, torrent: Torrent) -> list[MovieFile]:
return self.torrent_repository.get_movie_files_of_torrent(torrent_id=torrent.id)
async def get_movie_files_of_torrent(self, torrent: Torrent) -> list[MovieFile]:
return await self.torrent_repository.get_movie_files_of_torrent(
torrent_id=torrent.id
)

View File

@@ -68,12 +68,12 @@ def get_tv_service(
tv_service_dep = Annotated[TvService, Depends(get_tv_service)]
def get_show_by_id(
async def get_show_by_id(
tv_service: tv_service_dep,
show_id: ShowId = Path(..., description="The ID of the show"),
) -> Show:
try:
show = tv_service.get_show_by_id(show_id)
show = await tv_service.get_show_by_id(show_id)
except NotFoundError:
raise HTTPException(
status_code=404,
@@ -85,12 +85,12 @@ def get_show_by_id(
show_dep = Annotated[Show, Depends(get_show_by_id)]
def get_season_by_id(
async def get_season_by_id(
tv_service: tv_service_dep,
season_id: SeasonId = Path(..., description="The ID of the season"),
) -> Season:
try:
season = tv_service.get_season(season_id=season_id)
season = await tv_service.get_season(season_id=season_id)
except NotFoundError:
raise HTTPException(
status_code=404,

View File

@@ -1,3 +1,4 @@
import asyncio
import logging
import re
from collections.abc import Callable
@@ -51,7 +52,7 @@ class TvImportService(BaseMediaService[Show, Show]):
libraries=misc_config.tv_libraries,
)
def import_tv_show(
async def import_tv_show(
self,
show: Show,
source_directory: Path,
@@ -59,7 +60,10 @@ class TvImportService(BaseMediaService[Show, Show]):
torrent_id: str | None = None,
file_path_suffix: str = "",
) -> bool:
video_files, _, _ = get_files_for_import(directory=source_directory)
# Filesystem scan + archive extraction; offload off the event loop.
video_files, _, _ = await asyncio.to_thread(
get_files_for_import, directory=source_directory
)
if not video_files:
return False
@@ -79,17 +83,19 @@ class TvImportService(BaseMediaService[Show, Show]):
target_name += f" - {file_path_suffix}"
target_file = season_dir / f"{target_name}{video_file.suffix}"
import_file(target_file=target_file, source_file=video_file)
await asyncio.to_thread(
import_file, target_file=target_file, source_file=video_file
)
any_imported = True
# Update DB
try:
season = self.tv_repository.get_season_by_number(s_num, show.id)
season = await self.tv_repository.get_season_by_number(s_num, show.id)
episode = next(
(e for e in season.episodes if e.number == e_num), None
)
if episode:
self.tv_repository.add_episode_file(
await self.tv_repository.add_episode_file(
EpisodeFile(
episode_id=episode.id,
quality=quality,
@@ -101,8 +107,8 @@ class TvImportService(BaseMediaService[Show, Show]):
log.exception(f"Could not update DB for {video_file.name}")
return any_imported
def import_torrent_files(self, torrent: Torrent, show: Show) -> None:
success = self.import_tv_show(
async def import_torrent_files(self, torrent: Torrent, show: Show) -> None:
success = await self.import_tv_show(
show=show,
source_directory=get_torrent_filepath(torrent),
quality=torrent.quality,
@@ -110,44 +116,45 @@ class TvImportService(BaseMediaService[Show, Show]):
)
if success:
torrent.imported = True
self.torrent_service.torrent_repository.save_torrent(torrent=torrent)
self.notify_import_success(show.name, "TV show")
await self.torrent_service.torrent_repository.save_torrent(torrent=torrent)
await self.notify_import_success(show.name, "TV show")
else:
self.notify_import_failure(show.name, "TV show")
await self.notify_import_failure(show.name, "TV show")
def get_import_candidates(
async def get_import_candidates(
self, tv_path: Path, metadata_provider: AbstractMetadataProvider
) -> MediaImportSuggestion:
return super().get_import_candidates(
return await super().get_import_candidates(
directory=tv_path,
metadata_provider=metadata_provider,
search_func=self.tv_metadata_service.search_for_show,
)
def import_existing_tv_show(self, tv_show: Show, source_directory: Path) -> bool:
def _logic(s: Show, path: Path, _: Callable[[Any], Any]) -> bool:
return self.import_tv_show(s, path, file_path_suffix="IMPORTED")
async def import_existing_tv_show(self, tv_show: Show, source_directory: Path) -> bool:
async def _logic(s: Show, path: Path, _: Callable[[Any], Any]) -> bool:
return await self.import_tv_show(s, path, file_path_suffix="IMPORTED")
return self.import_existing_media(
async def _noop(_: object) -> None:
return None
return await self.import_existing_media(
media=tv_show,
source_directory=source_directory,
import_func=_logic,
add_file_record_func=lambda _: (
None
), # Handled inside import_tv_show for now
add_file_record_func=_noop,
)
def get_importable_tv_shows(
async def get_importable_tv_shows(
self, metadata_provider: AbstractMetadataProvider
) -> list[MediaImportSuggestion]:
return self.get_importable_media(
return await self.get_importable_media(
root_path=MediaManagerConfig().misc.tv_directory,
metadata_provider=metadata_provider,
get_candidates_func=self.get_import_candidates,
)
def import_all_torrents(self) -> None:
self.import_all_torrents_base(
async def import_all_torrents(self) -> None:
await self.import_all_torrents_base(
get_media_func=self.torrent_service.get_show_of_torrent,
import_torrent_func=self.import_torrent_files,
media_type_name="tv show",

View File

@@ -18,13 +18,13 @@ class TvMetadataService(BaseMetadataService[Show, Show]):
super().__init__(repository=tv_repository)
self.tv_repository = tv_repository
def add_show(
async def add_show(
self,
external_id: int,
metadata_provider: AbstractMetadataProvider,
language: str | None = None,
) -> Show:
return self.add_media_base(
return await self.add_media_base(
external_id=external_id,
metadata_provider=metadata_provider,
get_metadata_func=metadata_provider.get_show_metadata,
@@ -33,39 +33,39 @@ class TvMetadataService(BaseMetadataService[Show, Show]):
language=language,
)
def search_for_show(
async def search_for_show(
self, query: str, metadata_provider: AbstractMetadataProvider
) -> list[MetaDataProviderSearchResult]:
return self.search_for_media_base(
return await self.search_for_media_base(
query=query,
metadata_provider=metadata_provider,
search_func=metadata_provider.search_show,
get_by_external_id_func=self.tv_repository.get_show_by_external_id,
)
def get_popular_shows(
async def get_popular_shows(
self, metadata_provider: AbstractMetadataProvider
) -> list[MetaDataProviderSearchResult]:
return self.get_popular_media_base(
return await self.get_popular_media_base(
metadata_provider=metadata_provider,
search_func=metadata_provider.search_show,
)
def update_show_metadata(
async def update_show_metadata(
self, db_show: Show, metadata_provider: AbstractMetadataProvider
) -> Show | None:
"""
Updates the metadata of a show.
"""
log.debug(f"Found show: {db_show.name} for metadata update.")
fresh_show_data = metadata_provider.get_show_metadata(
fresh_show_data = await metadata_provider.get_show_metadata(
show_id=db_show.external_id, language=db_show.original_language
)
if not fresh_show_data:
log.warning(f"Could not fetch fresh metadata for show {db_show.name}")
return db_show
self.tv_repository.update_show_attributes(
await self.tv_repository.update_show_attributes(
show_id=db_show.id,
name=fresh_show_data.name,
overview=fresh_show_data.overview,
@@ -83,7 +83,7 @@ class TvMetadataService(BaseMetadataService[Show, Show]):
existing_season = existing_season_external_ids[
fresh_season_data.external_id
]
self.tv_repository.update_season_attributes(
await self.tv_repository.update_season_attributes(
season_id=existing_season.id,
name=fresh_season_data.name,
overview=fresh_season_data.overview,
@@ -96,7 +96,7 @@ class TvMetadataService(BaseMetadataService[Show, Show]):
existing_episode = existing_episode_external_ids[
fresh_episode_data.external_id
]
self.tv_repository.update_episode_attributes(
await self.tv_repository.update_episode_attributes(
episode_id=existing_episode.id,
title=fresh_episode_data.title,
overview=fresh_episode_data.overview,
@@ -109,7 +109,7 @@ class TvMetadataService(BaseMetadataService[Show, Show]):
title=fresh_episode_data.title,
overview=fresh_episode_data.overview,
)
self.tv_repository.add_episode_to_season(
await self.tv_repository.add_episode_to_season(
season_id=existing_season.id, episode_data=episode_schema
)
else:
@@ -130,19 +130,19 @@ class TvMetadataService(BaseMetadataService[Show, Show]):
for ep_data in fresh_season_data.episodes
],
)
self.tv_repository.add_season_to_show(
await self.tv_repository.add_season_to_show(
show_id=db_show.id, season_data=season_schema
)
updated_show = self.tv_repository.get_show_by_id(show_id=db_show.id)
metadata_provider.download_show_poster_image(show=updated_show)
updated_show = await self.tv_repository.get_show_by_id(show_id=db_show.id)
await metadata_provider.download_show_poster_image(show=updated_show)
return updated_show
def update_all_non_ended_shows_metadata(self) -> None:
def get_non_ended_shows() -> list[Show]:
return [show for show in self.tv_repository.get_shows() if not show.ended]
async def update_all_non_ended_shows_metadata(self) -> None:
async def get_non_ended_shows() -> list[Show]:
return [show for show in await self.tv_repository.get_shows() if not show.ended]
self.update_all_metadata_base(
await self.update_all_metadata_base(
get_all_to_update_func=get_non_ended_shows,
update_single_func=self.update_show_metadata,
tmdb_provider_class=TmdbMetadataProvider,

View File

@@ -1,8 +1,9 @@
from sqlalchemy import distinct, func, select
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.orm import Session, joinedload
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload, selectinload
from media_manager.common.repository import BaseRepository
from media_manager.common.repository import BaseRepository, EntityId
from media_manager.exceptions import ConflictError, NotFoundError
from media_manager.torrent.models import Torrent as TorrentModel
from media_manager.torrent.schemas import Torrent as TorrentSchema
@@ -22,23 +23,27 @@ from media_manager.tv.schemas import Season as SeasonSchema
from media_manager.tv.schemas import Show as ShowSchema
def _load_show_tree(): # noqa: ANN202
return selectinload(Show.seasons).selectinload(Season.episodes)
class TvRepository(BaseRepository[Show, ShowSchema]):
"""
Repository for managing TV shows, seasons, and episodes in the database.
Provides methods to retrieve, save, and delete shows and seasons.
"""
def __init__(self, db: Session) -> None:
def __init__(self, db: AsyncSession) -> None:
super().__init__(db, Show, ShowSchema)
def get_show_by_id(self, show_id: ShowId) -> ShowSchema:
async def get_show_by_id(self, show_id: ShowId) -> ShowSchema:
try:
stmt = (
select(Show)
.where(Show.id == show_id)
.options(joinedload(Show.seasons).joinedload(Season.episodes))
.options(_load_show_tree())
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if not result:
msg = f"Show with id {show_id} not found."
raise NotFoundError(msg)
@@ -48,7 +53,7 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
else:
return ShowSchema.model_validate(result)
def get_show_by_external_id(
async def get_show_by_external_id(
self, external_id: int, metadata_provider: str
) -> ShowSchema:
try:
@@ -56,9 +61,9 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
select(Show)
.where(Show.external_id == external_id)
.where(Show.metadata_provider == metadata_provider)
.options(joinedload(Show.seasons).joinedload(Season.episodes))
.options(_load_show_tree())
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if not result:
msg = f"Show with external_id {external_id} and provider {metadata_provider} not found."
raise NotFoundError(msg)
@@ -70,42 +75,46 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
else:
return ShowSchema.model_validate(result)
def get_shows(self) -> list[ShowSchema]:
async def get_shows(self) -> list[ShowSchema]:
try:
stmt = select(Show).options(
joinedload(Show.seasons).joinedload(Season.episodes)
)
results = self.db.execute(stmt).scalars().unique().all()
stmt = select(Show).options(_load_show_tree())
results = (await self.db.execute(stmt)).scalars().unique().all()
except SQLAlchemyError:
log.exception("Database error while retrieving all shows")
raise
else:
return [ShowSchema.model_validate(show) for show in results]
delete_show = BaseRepository.delete
set_show_library = BaseRepository.set_library
async def delete_show(self, entity_id: EntityId) -> None:
await self.delete(entity_id)
def get_total_downloaded_episodes_count(self) -> int:
async def set_show_library(self, entity_id: EntityId, library: str) -> None:
await self.set_library(entity_id, library)
async def get_total_downloaded_episodes_count(self) -> int:
try:
stmt = (
select(func.count(distinct(Episode.id)))
.select_from(Episode)
.join(EpisodeFile)
)
result = self.db.execute(stmt).scalar_one_or_none()
result = (await self.db.execute(stmt)).scalar_one_or_none()
except SQLAlchemyError:
log.exception("Database error while calculating downloaded episodes count")
raise
else:
return result or 0
def save_show(self, show: ShowSchema) -> ShowSchema:
db_show = self.db.get(Show, show.id) if show.id else None
async def save_show(self, show: ShowSchema) -> ShowSchema:
db_show = await self.db.get(Show, show.id) if show.id else None
if db_show: # Use base for update
return self.save_media_base(
await self.save_media_base(
media_schema=show, model_class=Show, exclude={"seasons", "episodes"}
)
# save_media_base returns a non-eager-loaded schema; reload with
# selectinload so ShowSchema.seasons/episodes don't lazy-load.
return await self.get_show_by_id(db_show.id)
# Custom insertion for nested seasons/episodes
db_show = Show(
@@ -145,80 +154,92 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
)
self.db.add(db_show)
try:
self.db.commit()
self.db.refresh(db_show)
await self.db.commit()
await self.db.refresh(db_show, ["seasons"])
except IntegrityError as e:
self.db.rollback()
await self.db.rollback()
msg = f"Integrity error: {e.orig}"
raise ConflictError(msg) from e
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
else:
return ShowSchema.model_validate(db_show)
# AsyncSession forbids implicit lazy loads after commit; reload eagerly.
return await self.get_show_by_id(db_show.id)
def get_season(self, season_id: SeasonId) -> SeasonSchema:
season = self.db.get(Season, season_id)
async def get_season(self, season_id: SeasonId) -> SeasonSchema:
season = await self.db.get(
Season, season_id, options=[selectinload(Season.episodes)]
)
if not season:
msg = f"Season {season_id} not found"
raise NotFoundError(msg)
return SeasonSchema.model_validate(season)
def get_episode(self, episode_id: EpisodeId) -> EpisodeSchema:
episode = self.db.get(Episode, episode_id)
async def get_episode(self, episode_id: EpisodeId) -> EpisodeSchema:
episode = await self.db.get(Episode, episode_id)
if not episode:
msg = f"Episode {episode_id} not found"
raise NotFoundError(msg)
return EpisodeSchema.model_validate(episode)
def get_season_by_episode(self, episode_id: EpisodeId) -> SeasonSchema:
stmt = select(Season).join(Season.episodes).where(Episode.id == episode_id)
season = self.db.scalar(stmt)
async def get_season_by_episode(self, episode_id: EpisodeId) -> SeasonSchema:
stmt = (
select(Season)
.join(Season.episodes)
.where(Episode.id == episode_id)
.options(selectinload(Season.episodes))
)
season = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if not season:
msg = f"Season for episode {episode_id} not found"
raise NotFoundError(msg)
return SeasonSchema.model_validate(season)
def get_season_by_number(self, season_number: int, show_id: ShowId) -> SeasonSchema:
async def get_season_by_number(
self, season_number: int, show_id: ShowId
) -> SeasonSchema:
stmt = (
select(Season)
.where(Season.show_id == show_id)
.where(Season.number == season_number)
.options(joinedload(Season.episodes), joinedload(Season.show))
.options(selectinload(Season.episodes), joinedload(Season.show))
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if not result:
msg = f"Season {season_number} for show {show_id} not found"
raise NotFoundError(msg)
return SeasonSchema.model_validate(result)
def add_episode_file(self, episode_file: EpisodeFileSchema) -> EpisodeFileSchema:
return self.add_media_file_base(
async def add_episode_file(
self, episode_file: EpisodeFileSchema
) -> EpisodeFileSchema:
return await self.add_media_file_base(
file_schema=episode_file,
model_class=EpisodeFile,
schema_class=EpisodeFileSchema,
)
def remove_episode_files_by_torrent_id(self, torrent_id: TorrentId) -> int:
return self.remove_files_by_torrent_id_base(
async def remove_episode_files_by_torrent_id(self, torrent_id: TorrentId) -> int:
return await self.remove_files_by_torrent_id_base(
torrent_id=torrent_id, model_class=EpisodeFile
)
def get_episode_files_by_season_id(
async def get_episode_files_by_season_id(
self, season_id: SeasonId
) -> list[EpisodeFileSchema]:
stmt = select(EpisodeFile).join(Episode).where(Episode.season_id == season_id)
results = self.db.execute(stmt).scalars().all()
results = (await self.db.execute(stmt)).scalars().all()
return [EpisodeFileSchema.model_validate(ef) for ef in results]
def get_episode_files_by_episode_id(
async def get_episode_files_by_episode_id(
self, episode_id: EpisodeId
) -> list[EpisodeFileSchema]:
stmt = select(EpisodeFile).where(EpisodeFile.episode_id == episode_id)
results = self.db.execute(stmt).scalars().all()
results = (await self.db.execute(stmt)).scalars().all()
return [EpisodeFileSchema.model_validate(sf) for sf in results]
def get_torrents_by_show_id(self, show_id: ShowId) -> list[TorrentSchema]:
async def get_torrents_by_show_id(self, show_id: ShowId) -> list[TorrentSchema]:
stmt = (
select(TorrentModel)
.distinct()
@@ -227,10 +248,10 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
.join(Season, Season.id == Episode.season_id)
.where(Season.show_id == show_id)
)
results = self.db.execute(stmt).scalars().unique().all()
results = (await self.db.execute(stmt)).scalars().unique().all()
return [TorrentSchema.model_validate(t) for t in results]
def get_all_shows_with_torrents(self) -> list[ShowSchema]:
async def get_all_shows_with_torrents(self) -> list[ShowSchema]:
stmt = (
select(Show)
.distinct()
@@ -238,13 +259,15 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
.join(Episode, Season.id == Episode.season_id)
.join(EpisodeFile, Episode.id == EpisodeFile.episode_id)
.join(TorrentModel, EpisodeFile.torrent_id == TorrentModel.id)
.options(joinedload(Show.seasons).joinedload(Season.episodes))
.options(_load_show_tree())
.order_by(Show.name)
)
results = self.db.execute(stmt).scalars().unique().all()
results = (await self.db.execute(stmt)).scalars().unique().all()
return [ShowSchema.model_validate(show) for show in results]
def get_seasons_by_torrent_id(self, torrent_id: TorrentId) -> list[SeasonNumber]:
async def get_seasons_by_torrent_id(
self, torrent_id: TorrentId
) -> list[SeasonNumber]:
stmt = (
select(Season.number)
.distinct()
@@ -252,10 +275,12 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
.join(EpisodeFile, EpisodeFile.episode_id == Episode.id)
.where(EpisodeFile.torrent_id == torrent_id)
)
results = self.db.execute(stmt).scalars().unique().all()
results = (await self.db.execute(stmt)).scalars().unique().all()
return [SeasonNumber(x) for x in results]
def get_episodes_by_torrent_id(self, torrent_id: TorrentId) -> list[EpisodeNumber]:
async def get_episodes_by_torrent_id(
self, torrent_id: TorrentId
) -> list[EpisodeNumber]:
stmt = (
select(Episode.number)
.distinct()
@@ -263,33 +288,35 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
.where(EpisodeFile.torrent_id == torrent_id)
.order_by(Episode.number)
)
episode_numbers = self.db.execute(stmt).scalars().all()
episode_numbers = (await self.db.execute(stmt)).scalars().all()
return [EpisodeNumber(n) for n in episode_numbers]
def get_show_by_season_id(self, season_id: SeasonId) -> ShowSchema:
async def get_show_by_season_id(self, season_id: SeasonId) -> ShowSchema:
stmt = (
select(Show)
.join(Season, Show.id == Season.show_id)
.where(Season.id == season_id)
.options(joinedload(Show.seasons).joinedload(Season.episodes))
.options(_load_show_tree())
)
result = self.db.execute(stmt).unique().scalar_one_or_none()
result = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if not result:
msg = f"Show for season {season_id} not found"
raise NotFoundError(msg)
return ShowSchema.model_validate(result)
def add_season_to_show(
async def add_season_to_show(
self, show_id: ShowId, season_data: SeasonSchema
) -> SeasonSchema:
db_show = self.db.get(Show, show_id)
db_show = await self.db.get(Show, show_id)
if not db_show:
msg = f"Show {show_id} not found"
raise NotFoundError(msg)
stmt = select(Season).where(
Season.show_id == show_id, Season.number == season_data.number
stmt = (
select(Season)
.where(Season.show_id == show_id, Season.number == season_data.number)
.options(selectinload(Season.episodes))
)
existing = self.db.execute(stmt).scalar_one_or_none()
existing = (await self.db.execute(stmt)).unique().scalar_one_or_none()
if existing:
return SeasonSchema.model_validate(existing)
db_season = Season(
@@ -312,28 +339,28 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
)
self.db.add(db_season)
try:
self.db.commit()
self.db.refresh(db_season)
await self.db.commit()
await self.db.refresh(db_season, ["episodes"])
except IntegrityError as e:
self.db.rollback()
await self.db.rollback()
msg = f"Integrity error: {e.orig}"
raise ConflictError(msg) from e
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
return SeasonSchema.model_validate(db_season)
def add_episode_to_season(
async def add_episode_to_season(
self, season_id: SeasonId, episode_data: EpisodeSchema
) -> EpisodeSchema:
db_season = self.db.get(Season, season_id)
db_season = await self.db.get(Season, season_id)
if not db_season:
msg = f"Season {season_id} not found"
raise NotFoundError(msg)
stmt = select(Episode).where(
Episode.season_id == season_id, Episode.number == episode_data.number
)
existing = self.db.execute(stmt).scalar_one_or_none()
existing = (await self.db.execute(stmt)).scalar_one_or_none()
if existing:
return EpisodeSchema.model_validate(existing)
db_episode = Episode(
@@ -346,18 +373,18 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
)
self.db.add(db_episode)
try:
self.db.commit()
self.db.refresh(db_episode)
await self.db.commit()
await self.db.refresh(db_episode)
except IntegrityError as e:
self.db.rollback()
await self.db.rollback()
msg = f"Integrity error: {e.orig}"
raise ConflictError(msg) from e
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
return EpisodeSchema.model_validate(db_episode)
def update_show_attributes(
async def update_show_attributes(
self,
show_id: ShowId,
name: str | None = None,
@@ -367,7 +394,7 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
continuous_download: bool | None = None,
imdb_id: str | None = None,
) -> ShowSchema:
return self.update_media_attributes_base(
await self.update_media_attributes_base(
media_id=show_id,
model_class=Show,
name=name,
@@ -377,11 +404,18 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
continuous_download=continuous_download,
imdb_id=imdb_id,
)
# Reload with seasons/episodes eagerly; the base returns a schema built
# from a non-eager db.get() which can't lazy-load under AsyncSession.
return await self.get_show_by_id(show_id)
def update_season_attributes(
async def update_season_attributes(
self, season_id: SeasonId, name: str | None = None, overview: str | None = None
) -> SeasonSchema:
db_season = self.db.get(Season, season_id)
# selectinload episodes so SeasonSchema.model_validate doesn't trip
# an implicit lazy load under AsyncSession.
db_season = await self.db.get(
Season, season_id, options=[selectinload(Season.episodes)]
)
if not db_season:
msg = f"Season {season_id} not found"
raise NotFoundError(msg)
@@ -394,20 +428,20 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
updated = True
if updated:
try:
self.db.commit()
self.db.refresh(db_season)
await self.db.commit()
await self.db.refresh(db_season, ["episodes"])
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
return SeasonSchema.model_validate(db_season)
def update_episode_attributes(
async def update_episode_attributes(
self,
episode_id: EpisodeId,
title: str | None = None,
overview: str | None = None,
) -> EpisodeSchema:
db_episode = self.db.get(Episode, episode_id)
db_episode = await self.db.get(Episode, episode_id)
if not db_episode:
msg = f"Episode {episode_id} not found"
raise NotFoundError(msg)
@@ -420,9 +454,9 @@ class TvRepository(BaseRepository[Show, ShowSchema]):
updated = True
if updated:
try:
self.db.commit()
self.db.refresh(db_episode)
await self.db.commit()
await self.db.refresh(db_episode)
except SQLAlchemyError:
self.db.rollback()
await self.db.rollback()
raise
return EpisodeSchema.model_validate(db_episode)

View File

@@ -41,7 +41,7 @@ router = APIRouter()
"/search",
dependencies=[Depends(current_active_user)],
)
def search_metadata_providers_for_a_show(
async def search_metadata_providers_for_a_show(
tv_metadata_service: tv_metadata_service_dep,
query: str,
metadata_provider: metadata_provider_dep,
@@ -49,7 +49,7 @@ def search_metadata_providers_for_a_show(
"""
Search for a show on the configured metadata provider.
"""
return tv_metadata_service.search_for_show(
return await tv_metadata_service.search_for_show(
query=query, metadata_provider=metadata_provider
)
@@ -58,14 +58,16 @@ def search_metadata_providers_for_a_show(
"/recommended",
dependencies=[Depends(current_active_user)],
)
def get_recommended_shows(
async def get_recommended_shows(
tv_metadata_service: tv_metadata_service_dep,
metadata_provider: metadata_provider_dep,
) -> list[MetaDataProviderSearchResult]:
"""
Get a list of recommended/popular shows from the metadata provider.
"""
return tv_metadata_service.get_popular_shows(metadata_provider=metadata_provider)
return await tv_metadata_service.get_popular_shows(
metadata_provider=metadata_provider
)
# -----------------------------------------------------------------------------
@@ -78,13 +80,13 @@ def get_recommended_shows(
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_superuser)],
)
def get_all_importable_shows(
async def get_all_importable_shows(
tv_import_service: tv_import_service_dep, metadata_provider: metadata_provider_dep
) -> list[MediaImportSuggestion]:
"""
Get a list of unknown shows that were detected in the TV directory and are importable.
"""
return tv_import_service.get_importable_tv_shows(
return await tv_import_service.get_importable_tv_shows(
metadata_provider=metadata_provider
)
@@ -94,7 +96,7 @@ def get_all_importable_shows(
dependencies=[Depends(current_superuser)],
status_code=status.HTTP_204_NO_CONTENT,
)
def import_detected_show(
async def import_detected_show(
tv_import_service: tv_import_service_dep, tv_show: show_dep, directory: str
) -> None:
"""
@@ -105,7 +107,7 @@ def import_detected_show(
MediaManagerConfig().misc.tv_directory
):
raise HTTPException(status.HTTP_400_BAD_REQUEST, "No such directory")
tv_import_service.import_existing_tv_show(
await tv_import_service.import_existing_tv_show(
tv_show=tv_show, source_directory=source_directory
)
@@ -119,11 +121,11 @@ def import_detected_show(
"/shows",
dependencies=[Depends(current_active_user)],
)
def get_all_shows(tv_service: tv_service_dep) -> list[Show]:
async def get_all_shows(tv_service: tv_service_dep) -> list[Show]:
"""
Get all shows in the library.
"""
return tv_service.get_all_shows()
return await tv_service.get_all_shows()
@router.post(
@@ -137,7 +139,7 @@ def get_all_shows(tv_service: tv_service_dep) -> list[Show]:
}
},
)
def add_a_show(
async def add_a_show(
tv_metadata_service: tv_metadata_service_dep,
metadata_provider: metadata_provider_dep,
show_id: int,
@@ -147,13 +149,13 @@ def add_a_show(
Add a new show to the library.
"""
try:
show = tv_metadata_service.add_show(
show = await tv_metadata_service.add_show(
external_id=show_id,
metadata_provider=metadata_provider,
language=language,
)
except MediaAlreadyExistsError:
show = tv_metadata_service.tv_repository.get_show_by_external_id(
show = await tv_metadata_service.tv_repository.get_show_by_external_id(
show_id, metadata_provider=metadata_provider.name
)
if not show:
@@ -165,11 +167,11 @@ def add_a_show(
"/shows/torrents",
dependencies=[Depends(current_active_user)],
)
def get_shows_with_torrents(tv_service: tv_service_dep) -> list[RichShowTorrent]:
async def get_shows_with_torrents(tv_service: tv_service_dep) -> list[RichShowTorrent]:
"""
Get all shows that are associated with torrents.
"""
return tv_service.get_all_shows_with_torrents()
return await tv_service.get_all_shows_with_torrents()
@router.get(
@@ -192,11 +194,11 @@ def get_available_libraries() -> list[LibraryItem]:
"/shows/{show_id}",
dependencies=[Depends(current_active_user)],
)
def get_a_show(show: show_dep, tv_service: tv_service_dep) -> PublicShow:
async def get_a_show(show: show_dep, tv_service: tv_service_dep) -> PublicShow:
"""
Get details for a specific show.
"""
return tv_service.get_public_show_by_id(show=show)
return await tv_service.get_public_show_by_id(show=show)
@router.delete(
@@ -204,7 +206,7 @@ def get_a_show(show: show_dep, tv_service: tv_service_dep) -> PublicShow:
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(current_superuser)],
)
def delete_a_show(
async def delete_a_show(
tv_service: tv_service_dep,
show: show_dep,
delete_files_on_disk: bool = False,
@@ -213,7 +215,7 @@ def delete_a_show(
"""
Delete a show from the library.
"""
tv_service.delete_show(
await tv_service.delete_show(
show=show,
delete_files_on_disk=delete_files_on_disk,
delete_torrents=delete_torrents,
@@ -224,7 +226,7 @@ def delete_a_show(
"/shows/{show_id}/metadata",
dependencies=[Depends(current_active_user)],
)
def update_shows_metadata(
async def update_shows_metadata(
show: show_dep,
tv_metadata_service: tv_metadata_service_dep,
tv_service: tv_service_dep,
@@ -233,26 +235,26 @@ def update_shows_metadata(
"""
Update a show's metadata from the provider.
"""
tv_metadata_service.update_show_metadata(
await tv_metadata_service.update_show_metadata(
db_show=show, metadata_provider=metadata_provider
)
return tv_service.get_public_show_by_id(show=show)
return await tv_service.get_public_show_by_id(show=show)
@router.post(
"/shows/{show_id}/continuousDownload",
dependencies=[Depends(current_superuser)],
)
def set_continuous_download(
async def set_continuous_download(
show: show_dep, tv_service: tv_service_dep, continuous_download: bool
) -> PublicShow:
"""
Toggle whether future seasons of a show will be automatically downloaded.
"""
tv_service.set_show_continuous_download(
updated_show = await tv_service.set_show_continuous_download(
show=show, continuous_download=continuous_download
)
return tv_service.get_public_show_by_id(show=show)
return await tv_service.get_public_show_by_id(show=updated_show)
@router.post(
@@ -260,7 +262,7 @@ def set_continuous_download(
dependencies=[Depends(current_superuser)],
status_code=status.HTTP_204_NO_CONTENT,
)
def set_library(
async def set_library(
show: show_dep,
tv_service: tv_service_dep,
library: str,
@@ -268,7 +270,7 @@ def set_library(
"""
Set the library path for a Show.
"""
tv_service.set_show_library(show=show, library=library)
await tv_service.set_show_library(show=show, library=library)
return
@@ -276,11 +278,13 @@ def set_library(
"/shows/{show_id}/torrents",
dependencies=[Depends(current_active_user)],
)
def get_a_shows_torrents(show: show_dep, tv_service: tv_service_dep) -> RichShowTorrent:
async def get_a_shows_torrents(
show: show_dep, tv_service: tv_service_dep
) -> RichShowTorrent:
"""
Get torrents associated with a specific show.
"""
return tv_service.get_torrents_for_show(show=show)
return await tv_service.get_torrents_for_show(show=show)
# -----------------------------------------------------------------------------
@@ -292,7 +296,7 @@ def get_a_shows_torrents(show: show_dep, tv_service: tv_service_dep) -> RichShow
"/seasons/{season_id}",
dependencies=[Depends(current_active_user)],
)
def get_season(season: season_dep) -> Season:
async def get_season(season: season_dep) -> Season:
"""
Get details for a specific season.
"""
@@ -303,13 +307,13 @@ def get_season(season: season_dep) -> Season:
"/seasons/{season_id}/files",
dependencies=[Depends(current_active_user)],
)
def get_episode_files(
async def get_episode_files(
season: season_dep, tv_service: tv_service_dep
) -> list[PublicEpisodeFile]:
"""
Get episode files associated with a specific season.
"""
return tv_service.get_public_episode_files_by_season_id(season=season)
return await tv_service.get_public_episode_files_by_season_id(season=season)
# -----------------------------------------------------------------------------
@@ -322,7 +326,7 @@ def get_episode_files(
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_superuser)],
)
def get_torrents_for_a_season(
async def get_torrents_for_a_season(
tv_service: tv_service_dep,
show_id: ShowId,
season_number: int = 1,
@@ -332,7 +336,7 @@ def get_torrents_for_a_season(
Search for torrents for a specific season of a show.
Default season_number is 1 because it often returns multi-season torrents.
"""
return tv_service.get_all_available_torrents_for_a_season(
return await tv_service.get_all_available_torrents_for_a_season(
season_number=season_number,
show_id=show_id,
search_query_override=search_query_override,
@@ -344,7 +348,7 @@ def get_torrents_for_a_season(
status_code=status.HTTP_200_OK,
dependencies=[Depends(current_superuser)],
)
def download_a_torrent(
async def download_a_torrent(
tv_service: tv_service_dep,
public_indexer_result_id: IndexerQueryResultId,
show_id: ShowId,
@@ -353,7 +357,7 @@ def download_a_torrent(
"""
Trigger a download for a specific torrent.
"""
return tv_service.download_torrent(
return await tv_service.download_torrent(
public_indexer_result_id=public_indexer_result_id,
show_id=show_id,
override_show_file_path_suffix=override_file_path_suffix,
@@ -371,8 +375,8 @@ def download_a_torrent(
description="Total number of episodes downloaded",
dependencies=[Depends(current_active_user)],
)
def get_total_count_of_downloaded_episodes(tv_service: tv_service_dep) -> int:
async def get_total_count_of_downloaded_episodes(tv_service: tv_service_dep) -> int:
"""
Get the total count of downloaded episodes across all shows.
"""
return tv_service.get_total_downloaded_episodes_count()
return await tv_service.get_total_downloaded_episodes_count()

View File

@@ -1,3 +1,4 @@
import asyncio
import shutil
from pathlib import Path
@@ -54,23 +55,25 @@ class TvService(BaseMediaService[Show, Show]):
self.tv_import_service = tv_import_service
self.tv_metadata_service = tv_metadata_service
def get_total_downloaded_episodes_count(self) -> int:
async def get_total_downloaded_episodes_count(self) -> int:
"""
Get total number of downloaded episodes.
"""
return self.tv_repository.get_total_downloaded_episodes_count()
return await self.tv_repository.get_total_downloaded_episodes_count()
def set_show_library(self, show: Show, library: str) -> None:
self.tv_repository.set_show_library(show.id, library)
async def set_show_library(self, show: Show, library: str) -> None:
await self.tv_repository.set_show_library(show.id, library)
def get_all_shows(self) -> list[Show]:
async def get_all_shows(self) -> list[Show]:
"""
Get all shows in the library.
"""
return self.get_all_media()
# Use the TV-specific eager-loaded query so ShowSchema validation
# doesn't trigger lazy loads on seasons/episodes under AsyncSession.
return await self.tv_repository.get_shows()
def delete_show(
async def delete_show(
self,
show: Show,
delete_files_on_disk: bool = False,
@@ -91,24 +94,24 @@ class TvService(BaseMediaService[Show, Show]):
log.debug(f"Attempt to delete show directory: {show_dir}")
if show_dir.exists() and show_dir.is_dir():
shutil.rmtree(show_dir)
await asyncio.to_thread(shutil.rmtree, show_dir)
log.info(f"Deleted show directory: {show_dir}")
if delete_torrents:
torrents = self.tv_repository.get_torrents_by_show_id(show_id=show.id)
torrents = await self.tv_repository.get_torrents_by_show_id(show_id=show.id)
for torrent in torrents:
try:
self.torrent_service.cancel_download(torrent, delete_files=True)
self.torrent_service.delete_torrent(torrent_id=torrent.id)
await self.torrent_service.cancel_download(torrent, delete_files=True)
await self.torrent_service.delete_torrent(torrent_id=torrent.id)
log.info(f"Deleted torrent: {torrent.hash}")
except Exception:
log.warning(
f"Failed to delete torrent {torrent.hash}", exc_info=True
)
self.tv_repository.delete_show(show.id)
await self.tv_repository.delete_show(show.id)
def get_public_episode_files_by_season_id(
async def get_public_episode_files_by_season_id(
self, season: Season
) -> list[PublicEpisodeFile]:
"""
@@ -117,7 +120,7 @@ class TvService(BaseMediaService[Show, Show]):
:param season: The season object.
:return: A list of public episode files.
"""
episode_files = self.tv_repository.get_episode_files_by_season_id(
episode_files = await self.tv_repository.get_episode_files_by_season_id(
season_id=season.id
)
public_episode_files = [
@@ -125,12 +128,12 @@ class TvService(BaseMediaService[Show, Show]):
]
result = []
for episode_file in public_episode_files:
if self.episode_file_exists_on_file(episode_file=episode_file):
if await self.episode_file_exists_on_file(episode_file=episode_file):
episode_file.downloaded = True
result.append(episode_file)
return result
def get_all_available_torrents_for_a_season(
async def get_all_available_torrents_for_a_season(
self,
season_number: int,
show_id: ShowId,
@@ -146,11 +149,11 @@ class TvService(BaseMediaService[Show, Show]):
"""
if search_query_override:
return self.indexer_service.search(query=search_query_override, is_tv=True)
return await self.indexer_service.search(query=search_query_override, is_tv=True)
show = self.tv_repository.get_show_by_id(show_id=show_id)
show = await self.tv_repository.get_show_by_id(show_id=show_id)
torrents = self.indexer_service.search_season(
torrents = await self.indexer_service.search_season(
show=show, season_number=season_number
)
@@ -160,7 +163,7 @@ class TvService(BaseMediaService[Show, Show]):
is_tv=True, query_results=results, media=show
)
def get_public_show_by_id(self, show: Show) -> PublicShow:
async def get_public_show_by_id(self, show: Show) -> PublicShow:
"""
Get a public show from a Show object.
@@ -174,7 +177,7 @@ class TvService(BaseMediaService[Show, Show]):
public_season = PublicSeason.model_validate(season)
for episode in public_season.episodes:
episode.downloaded = self.is_episode_downloaded(
episode.downloaded = await self.is_episode_downloaded(
episode=episode,
season=season,
show=show,
@@ -190,16 +193,16 @@ class TvService(BaseMediaService[Show, Show]):
public_show.seasons = public_seasons
return public_show
def get_show_by_id(self, show_id: ShowId) -> Show:
async def get_show_by_id(self, show_id: ShowId) -> Show:
"""
Get a show by its ID.
:param show_id: The ID of the show.
:return: The show.
"""
return self.tv_repository.get_show_by_id(show_id=show_id)
return await self.tv_repository.get_show_by_id(show_id=show_id)
def is_season_downloaded(self, season: Season, show: Show) -> bool:
async def is_season_downloaded(self, season: Season, show: Show) -> bool:
"""
Check if a season is downloaded.
@@ -213,13 +216,13 @@ class TvService(BaseMediaService[Show, Show]):
return False
for episode in episodes:
if not self.is_episode_downloaded(
if not await self.is_episode_downloaded(
episode=episode, season=season, show=show
):
return False
return True
def is_episode_downloaded(
async def is_episode_downloaded(
self, episode: Episode, season: Season, show: Show
) -> bool:
"""
@@ -234,7 +237,7 @@ class TvService(BaseMediaService[Show, Show]):
:param show: The show object.
:return: True if the episode is downloaded and imported, False otherwise.
"""
episode_files = self.tv_repository.get_episode_files_by_episode_id(
episode_files = await self.tv_repository.get_episode_files_by_episode_id(
episode_id=episode.id
)
@@ -266,7 +269,7 @@ class TvService(BaseMediaService[Show, Show]):
return False
def episode_file_exists_on_file(self, episode_file: EpisodeFile) -> bool:
async def episode_file_exists_on_file(self, episode_file: EpisodeFile) -> bool:
"""
Check if an episode file exists on the filesystem.
@@ -276,7 +279,7 @@ class TvService(BaseMediaService[Show, Show]):
if episode_file.torrent_id is None:
return True
try:
torrent_file = self.torrent_service.get_torrent_by_id(
torrent_file = await self.torrent_service.get_torrent_by_id(
torrent_id=episode_file.torrent_id
)
@@ -287,7 +290,7 @@ class TvService(BaseMediaService[Show, Show]):
return False
def get_show_by_external_id(
async def get_show_by_external_id(
self, external_id: int, metadata_provider: str
) -> Show | None:
"""
@@ -297,54 +300,54 @@ class TvService(BaseMediaService[Show, Show]):
:param metadata_provider: The metadata provider.
:return: The show or None if not found.
"""
return self.tv_repository.get_show_by_external_id(
return await self.tv_repository.get_show_by_external_id(
external_id=external_id, metadata_provider=metadata_provider
)
def get_season(self, season_id: SeasonId) -> Season:
async def get_season(self, season_id: SeasonId) -> Season:
"""
Get a season by its ID.
:param season_id: The ID of the season.
:return: The season.
"""
return self.tv_repository.get_season(season_id=season_id)
return await self.tv_repository.get_season(season_id=season_id)
def get_episode(self, episode_id: EpisodeId) -> Episode:
async def get_episode(self, episode_id: EpisodeId) -> Episode:
"""
Get an episode by its ID.
:param episode_id: The ID of the episode.
:return: The episode.
"""
return self.tv_repository.get_episode(episode_id=episode_id)
return await self.tv_repository.get_episode(episode_id=episode_id)
def get_season_by_episode(self, episode_id: EpisodeId) -> Season:
async def get_season_by_episode(self, episode_id: EpisodeId) -> Season:
"""
Get a season by the episode ID.
:param episode_id: The ID of the episode.
:return: The season.
"""
return self.tv_repository.get_season_by_episode(episode_id=episode_id)
return await self.tv_repository.get_season_by_episode(episode_id=episode_id)
def get_torrents_for_show(self, show: Show) -> RichShowTorrent:
async def get_torrents_for_show(self, show: Show) -> RichShowTorrent:
"""
Get torrents for a given show.
:param show: The show.
:return: A rich show torrent.
"""
show_torrents = self.tv_repository.get_torrents_by_show_id(show_id=show.id)
show_torrents = await self.tv_repository.get_torrents_by_show_id(show_id=show.id)
rich_season_torrents = []
for show_torrent in show_torrents:
seasons = self.tv_repository.get_seasons_by_torrent_id(
seasons = await self.tv_repository.get_seasons_by_torrent_id(
torrent_id=show_torrent.id
)
episodes = self.tv_repository.get_episodes_by_torrent_id(
episodes = await self.tv_repository.get_episodes_by_torrent_id(
torrent_id=show_torrent.id
)
episode_files = self.torrent_service.get_episode_files_of_torrent(
episode_files = await self.torrent_service.get_episode_files_of_torrent(
torrent=show_torrent
)
@@ -372,16 +375,16 @@ class TvService(BaseMediaService[Show, Show]):
torrents=rich_season_torrents,
)
def get_all_shows_with_torrents(self) -> list[RichShowTorrent]:
async def get_all_shows_with_torrents(self) -> list[RichShowTorrent]:
"""
Get all shows with torrents.
:return: A list of rich show torrents.
"""
shows = self.tv_repository.get_all_shows_with_torrents()
return [self.get_torrents_for_show(show=show) for show in shows]
shows = await self.tv_repository.get_all_shows_with_torrents()
return [await self.get_torrents_for_show(show=show) for show in shows]
def download_torrent(
async def download_torrent(
self,
public_indexer_result_id: IndexerQueryResultId,
show_id: ShowId,
@@ -395,15 +398,15 @@ class TvService(BaseMediaService[Show, Show]):
:param override_show_file_path_suffix: Optional override for the file path suffix.
:return: The downloaded torrent.
"""
indexer_result = self.indexer_service.get_result(
indexer_result = await self.indexer_service.get_result(
result_id=public_indexer_result_id
)
show_torrent = self.torrent_service.download(indexer_result=indexer_result)
self.torrent_service.pause_download(torrent=show_torrent)
show_torrent = await self.torrent_service.download(indexer_result=indexer_result)
await self.torrent_service.pause_download(torrent=show_torrent)
try:
for season_number in indexer_result.season:
season = self.tv_repository.get_season_by_number(
season = await self.tv_repository.get_season_by_number(
season_number=season_number, show_id=show_id
)
episodes = {episode.number: episode.id for episode in season.episodes}
@@ -435,14 +438,14 @@ class TvService(BaseMediaService[Show, Show]):
torrent_id=show_torrent.id,
file_path_suffix=override_show_file_path_suffix,
)
self.tv_repository.add_episode_file(episode_file=episode_file)
await self.tv_repository.add_episode_file(episode_file=episode_file)
except IntegrityError:
log.error(
f"Episode file for episode {episode_id} of season {season.id} and quality {indexer_result.quality} already exists, skipping."
)
self.tv_repository.remove_episode_files_by_torrent_id(show_torrent.id)
self.torrent_service.cancel_download(
await self.tv_repository.remove_episode_files_by_torrent_id(show_torrent.id)
await self.torrent_service.cancel_download(
torrent=show_torrent, delete_files=True
)
raise
@@ -450,7 +453,7 @@ class TvService(BaseMediaService[Show, Show]):
log.info(
f"Successfully added episode files for torrent {show_torrent.title} and show ID {show_id}"
)
self.torrent_service.resume_download(torrent=show_torrent)
await self.torrent_service.resume_download(torrent=show_torrent)
return show_torrent
@@ -465,7 +468,7 @@ class TvService(BaseMediaService[Show, Show]):
def get_root_season_directory(self, show: Show, season_number: int) -> Path:
return self.get_root_show_directory(show) / Path(f"Season {season_number}")
def set_show_continuous_download(
async def set_show_continuous_download(
self, show: Show, continuous_download: bool
) -> Show:
"""
@@ -475,18 +478,18 @@ class TvService(BaseMediaService[Show, Show]):
:param continuous_download: True to enable continuous download, False to disable.
:return: The updated Show object.
"""
return self.tv_repository.update_show_attributes(
return await self.tv_repository.update_show_attributes(
show_id=show.id, continuous_download=continuous_download
)
def import_all_torrents(self) -> None:
async def import_all_torrents(self) -> None:
"""
Delegate to TvImportService.
"""
self.tv_import_service.import_all_torrents()
await self.tv_import_service.import_all_torrents()
def update_all_non_ended_shows_metadata(self) -> None:
async def update_all_non_ended_shows_metadata(self) -> None:
"""
Delegate to TvMetadataService.
"""
self.tv_metadata_service.update_all_non_ended_shows_metadata()
await self.tv_metadata_service.update_all_non_ended_shows_metadata()