lukas.pupkalipinski/Aniworld
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix queue error

Browse Source
This commit is contained in:
Lukas 2025-12-10 20:55:09 +01:00
parent 798461a1ea
commit 99f79e4c29
6 changed files with 263 additions and 683 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
data/config.json
View File

@ -17,7 +17,8 @@
"keep_days": 30 "keep_days": 30
}, },
"other": { "other": {
"master_password_hash": "$pbkdf2-sha256$29000$lvLeO.c8xzjHOAeAcM45Zw$NwtHXYLnbZE5oQwAJtlvcxLTZav3LjQhkYOhHiPXwWc" "master_password_hash": "$pbkdf2-sha256$29000$Nyak1Np7j1Gq9V5rLUUoxQ$9/v2NQ9x2YcJ7N8aEgMVET24CO0ND3dWiGythcUgrJs",
"anime_directory": "/home/lukas/Volume/serien/"
}, },
"version": "1.0.0" "version": "1.0.0"
} }

2
src/server/services/anime_service.py
View File

@ -222,7 +222,7 @@ class AnimeService:
loop loop
) )
except Exception as exc: except Exception as exc:
logger.error("Error handling scan status event", error=str(exc)) logger.error("Error handling scan status event: %s", exc)
@lru_cache(maxsize=128) @lru_cache(maxsize=128)
def _cached_list_missing(self) -> list[dict]: def _cached_list_missing(self) -> list[dict]:

173
src/server/services/download_service.py
View File

@ -120,6 +120,9 @@ class DownloadService:
"""Initialize the service by loading queue state from database. """Initialize the service by loading queue state from database.
Should be called after database is initialized during app startup. Should be called after database is initialized during app startup.
Note: With the simplified model, status/priority/progress are now
managed in-memory only. The database stores the queue items
for persistence across restarts.
""" """
if self._db_initialized: if self._db_initialized:
return return
@ -127,44 +130,22 @@ class DownloadService:
try: try:
repository = self._get_repository() repository = self._get_repository()
# Load pending items from database # Load all items from database - they all start as PENDING
pending_items = await repository.get_pending_items() # since status is now managed in-memory only
for item in pending_items: all_items = await repository.get_all_items()
# Reset status if was downloading when saved for item in all_items:
if item.status == DownloadStatus.DOWNLOADING: # All items from database are treated as pending
item.status = DownloadStatus.PENDING item.status = DownloadStatus.PENDING
await repository.update_status(
item.id, DownloadStatus.PENDING
)
self._add_to_pending_queue(item) self._add_to_pending_queue(item)
# Load failed items from database
failed_items = await repository.get_failed_items()
for item in failed_items:
if item.retry_count < self._max_retries:
item.status = DownloadStatus.PENDING
await repository.update_status(
item.id, DownloadStatus.PENDING
)
self._add_to_pending_queue(item)
else:
self._failed_items.append(item)
# Load completed items for history
completed_items = await repository.get_completed_items(limit=100)
for item in completed_items:
self._completed_items.append(item)
self._db_initialized = True self._db_initialized = True
logger.info( logger.info(
"Queue restored from database", "Queue restored from database: pending_count=%d",
pending_count=len(self._pending_queue), len(self._pending_queue),
failed_count=len(self._failed_items),
completed_count=len(self._completed_items),
) )
except Exception as e: except Exception as e:
logger.error("Failed to load queue from database", error=str(e)) logger.error("Failed to load queue from database: %s", e, exc_info=True)
# Continue without persistence - queue will work in memory only # Continue without persistence - queue will work in memory only
self._db_initialized = True self._db_initialized = True
@ -181,59 +162,28 @@ class DownloadService:
repository = self._get_repository() repository = self._get_repository()
return await repository.save_item(item) return await repository.save_item(item)
except Exception as e: except Exception as e:
logger.error("Failed to save item to database", error=str(e)) logger.error("Failed to save item to database: %s", e)
return item return item
async def _update_status_in_database( async def _set_error_in_database(
self, self,
item_id: str, item_id: str,
status: DownloadStatus, error: str,
error: Optional[str] = None,
) -> bool: ) -> bool:
"""Update item status in the database. """Set error message on an item in the database.
Args: Args:
item_id: Download item ID item_id: Download item ID
status: New status error: Error message
error: Optional error message
Returns: Returns:
True if update succeeded True if update succeeded
""" """
try: try:
repository = self._get_repository() repository = self._get_repository()
return await repository.update_status(item_id, status, error) return await repository.set_error(item_id, error)
except Exception as e: except Exception as e:
logger.error("Failed to update status in database", error=str(e)) logger.error("Failed to set error in database: %s", e)
return False
async def _update_progress_in_database(
self,
item_id: str,
progress: float,
downloaded: int,
total: Optional[int],
speed: Optional[float],
) -> bool:
"""Update download progress in the database.
Args:
item_id: Download item ID
progress: Progress percentage
downloaded: Downloaded bytes
total: Total bytes
speed: Download speed in bytes/sec
Returns:
True if update succeeded
"""
try:
repository = self._get_repository()
return await repository.update_progress(
item_id, progress, downloaded, total, speed
)
except Exception as e:
logger.error("Failed to update progress in database", error=str(e))
return False return False
async def _delete_from_database(self, item_id: str) -> bool: async def _delete_from_database(self, item_id: str) -> bool:
@ -249,7 +199,7 @@ class DownloadService:
repository = self._get_repository() repository = self._get_repository()
return await repository.delete_item(item_id) return await repository.delete_item(item_id)
except Exception as e: except Exception as e:
logger.error("Failed to delete from database", error=str(e)) logger.error("Failed to delete from database: %s", e)
return False return False
async def _init_queue_progress(self) -> None: async def _init_queue_progress(self) -> None:
@ -271,7 +221,7 @@ class DownloadService:
) )
self._queue_progress_initialized = True self._queue_progress_initialized = True
except Exception as e: except Exception as e:
logger.error("Failed to initialize queue progress", error=str(e)) logger.error("Failed to initialize queue progress: %s", e)
def _add_to_pending_queue( def _add_to_pending_queue(
self, item: DownloadItem, front: bool = False self, item: DownloadItem, front: bool = False
@ -396,7 +346,7 @@ class DownloadService:
return created_ids return created_ids
except Exception as e: except Exception as e:
logger.error("Failed to add items to queue", error=str(e)) logger.error("Failed to add items to queue: %s", e)
raise DownloadServiceError(f"Failed to add items: {str(e)}") from e raise DownloadServiceError(f"Failed to add items: {str(e)}") from e
async def remove_from_queue(self, item_ids: List[str]) -> List[str]: async def remove_from_queue(self, item_ids: List[str]) -> List[str]:
@ -423,12 +373,10 @@ class DownloadService:
item.completed_at = datetime.now(timezone.utc) item.completed_at = datetime.now(timezone.utc)
self._failed_items.append(item) self._failed_items.append(item)
self._active_download = None self._active_download = None
# Update status in database # Delete cancelled item from database
await self._update_status_in_database( await self._delete_from_database(item_id)
item_id, DownloadStatus.CANCELLED
)
removed_ids.append(item_id) removed_ids.append(item_id)
logger.info("Cancelled active download", item_id=item_id) logger.info("Cancelled active download: item_id=%s", item_id)
continue continue
# Check pending queue - O(1) lookup using helper dict # Check pending queue - O(1) lookup using helper dict
@ -460,7 +408,7 @@ class DownloadService:
return removed_ids return removed_ids
except Exception as e: except Exception as e:
logger.error("Failed to remove items", error=str(e)) logger.error("Failed to remove items: %s", e)
raise DownloadServiceError( raise DownloadServiceError(
f"Failed to remove items: {str(e)}" f"Failed to remove items: {str(e)}"
) from e ) from e
@ -514,7 +462,7 @@ class DownloadService:
logger.info("Queue reordered", reordered_count=len(item_ids)) logger.info("Queue reordered", reordered_count=len(item_ids))
except Exception as e: except Exception as e:
logger.error("Failed to reorder queue", error=str(e)) logger.error("Failed to reorder queue: %s", e)
raise DownloadServiceError( raise DownloadServiceError(
f"Failed to reorder queue: {str(e)}" f"Failed to reorder queue: {str(e)}"
) from e ) from e
@ -558,7 +506,7 @@ class DownloadService:
return "queue_started" return "queue_started"
except Exception as e: except Exception as e:
logger.error("Failed to start queue processing", error=str(e)) logger.error("Failed to start queue processing: %s", e)
raise DownloadServiceError( raise DownloadServiceError(
f"Failed to start queue processing: {str(e)}" f"Failed to start queue processing: {str(e)}"
) from e ) from e
@ -847,15 +795,12 @@ class DownloadService:
self._add_to_pending_queue(item) self._add_to_pending_queue(item)
retried_ids.append(item.id) retried_ids.append(item.id)
# Update status in database # Status is now managed in-memory only
await self._update_status_in_database(
item.id, DownloadStatus.PENDING
)
logger.info( logger.info(
"Retrying failed item", "Retrying failed item: item_id=%s, retry_count=%d",
item_id=item.id, item.id,
retry_count=item.retry_count item.retry_count,
) )
if retried_ids: if retried_ids:
@ -875,7 +820,7 @@ class DownloadService:
return retried_ids return retried_ids
except Exception as e: except Exception as e:
logger.error("Failed to retry items", error=str(e)) logger.error("Failed to retry items: %s", e)
raise DownloadServiceError( raise DownloadServiceError(
f"Failed to retry: {str(e)}" f"Failed to retry: {str(e)}"
) from e ) from e
@ -892,21 +837,17 @@ class DownloadService:
logger.info("Skipping download due to shutdown") logger.info("Skipping download due to shutdown")
return return
# Update status in memory and database # Update status in memory (status is now in-memory only)
item.status = DownloadStatus.DOWNLOADING item.status = DownloadStatus.DOWNLOADING
item.started_at = datetime.now(timezone.utc) item.started_at = datetime.now(timezone.utc)
self._active_download = item self._active_download = item
await self._update_status_in_database(
item.id, DownloadStatus.DOWNLOADING
)
logger.info( logger.info(
"Starting download", "Starting download: item_id=%s, serie_key=%s, S%02dE%02d",
item_id=item.id, item.id,
serie_key=item.serie_id, item.serie_id,
serie_name=item.serie_name, item.episode.season,
season=item.episode.season, item.episode.episode,
episode=item.episode.episode,
) )
# Execute download via anime service # Execute download via anime service
@ -941,13 +882,11 @@ class DownloadService:
self._completed_items.append(item) self._completed_items.append(item)
# Update database # Delete completed item from database (status is in-memory)
await self._update_status_in_database( await self._delete_from_database(item.id)
item.id, DownloadStatus.COMPLETED
)
logger.info( logger.info(
"Download completed successfully", item_id=item.id "Download completed successfully: item_id=%s", item.id
) )
else: else:
raise AnimeServiceError("Download returned False") raise AnimeServiceError("Download returned False")
@ -955,20 +894,18 @@ class DownloadService:
except asyncio.CancelledError: except asyncio.CancelledError:
# Handle task cancellation during shutdown # Handle task cancellation during shutdown
logger.info( logger.info(
"Download cancelled during shutdown", "Download cancelled during shutdown: item_id=%s",
item_id=item.id, item.id,
) )
item.status = DownloadStatus.CANCELLED item.status = DownloadStatus.CANCELLED
item.completed_at = datetime.now(timezone.utc) item.completed_at = datetime.now(timezone.utc)
await self._update_status_in_database( # Delete cancelled item from database
item.id, DownloadStatus.CANCELLED await self._delete_from_database(item.id)
)
# Return item to pending queue if not shutting down # Return item to pending queue if not shutting down
if not self._is_shutting_down: if not self._is_shutting_down:
self._add_to_pending_queue(item, front=True) self._add_to_pending_queue(item, front=True)
await self._update_status_in_database( # Re-save to database as pending
item.id, DownloadStatus.PENDING await self._save_to_database(item)
)
raise # Re-raise to properly cancel the task raise # Re-raise to properly cancel the task
except Exception as e: except Exception as e:
@ -978,16 +915,14 @@ class DownloadService:
item.error = str(e) item.error = str(e)
self._failed_items.append(item) self._failed_items.append(item)
# Update database with error # Set error in database
await self._update_status_in_database( await self._set_error_in_database(item.id, str(e))
item.id, DownloadStatus.FAILED, str(e)
)
logger.error( logger.error(
"Download failed", "Download failed: item_id=%s, error=%s, retry_count=%d",
item_id=item.id, item.id,
error=str(e), str(e),
retry_count=item.retry_count, item.retry_count,
) )
# Note: Failure is already broadcast by AnimeService # Note: Failure is already broadcast by AnimeService
# via ProgressService when SeriesApp fires failed event # via ProgressService when SeriesApp fires failed event

657
src/server/services/queue_repository.py
View File

@ -3,9 +3,9 @@
This module provides a repository adapter that wraps the DownloadQueueService This module provides a repository adapter that wraps the DownloadQueueService
and provides the interface needed by DownloadService for queue persistence. and provides the interface needed by DownloadService for queue persistence.
The repository pattern abstracts the database operations from the business logic, The repository pattern abstracts the database operations from the business
allowing the DownloadService to work with domain models (DownloadItem) while logic, allowing the DownloadService to work with domain models (DownloadItem)
the repository handles conversion to/from database models (DownloadQueueItem). while the repository handles conversion to/from database models.
""" """
from __future__ import annotations from __future__ import annotations
@ -15,15 +15,15 @@ from typing import Callable, List, Optional
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from src.server.database.models import AnimeSeries
from src.server.database.models import DownloadPriority as DBDownloadPriority
from src.server.database.models import DownloadQueueItem as DBDownloadQueueItem from src.server.database.models import DownloadQueueItem as DBDownloadQueueItem
from src.server.database.models import DownloadStatus as DBDownloadStatus from src.server.database.service import (
from src.server.database.service import AnimeSeriesService, DownloadQueueService AnimeSeriesService,
DownloadQueueService,
EpisodeService,
)
from src.server.models.download import ( from src.server.models.download import (
DownloadItem, DownloadItem,
DownloadPriority, DownloadPriority,
DownloadProgress,
DownloadStatus, DownloadStatus,
EpisodeIdentifier, EpisodeIdentifier,
) )
@ -37,194 +37,110 @@ class QueueRepositoryError(Exception):
class QueueRepository: class QueueRepository:
"""Repository adapter for database-backed download queue operations. """Repository adapter for database-backed download queue operations.
Provides clean interface for queue operations while handling Provides clean interface for queue operations while handling
model conversion between Pydantic (DownloadItem) and SQLAlchemy model conversion between Pydantic (DownloadItem) and SQLAlchemy
(DownloadQueueItem) models. (DownloadQueueItem) models.
Note: The database model (DownloadQueueItem) is simplified and only
stores episode_id as a foreign key. Status, priority, progress, and
retry_count are managed in-memory by the DownloadService.
Attributes: Attributes:
_db_session_factory: Factory function to create database sessions _db_session_factory: Factory function to create database sessions
""" """
def __init__( def __init__(
self, self,
db_session_factory: Callable[[], AsyncSession], db_session_factory: Callable[[], AsyncSession],
) -> None: ) -> None:
"""Initialize the queue repository. """Initialize the queue repository.
Args: Args:
db_session_factory: Factory function that returns AsyncSession instances db_session_factory: Factory function that returns AsyncSession
""" """
self._db_session_factory = db_session_factory self._db_session_factory = db_session_factory
logger.info("QueueRepository initialized") logger.info("QueueRepository initialized")
# ========================================================================= # =========================================================================
# Model Conversion Methods # Model Conversion Methods
# ========================================================================= # =========================================================================
def _status_to_db(self, status: DownloadStatus) -> DBDownloadStatus:
"""Convert Pydantic DownloadStatus to SQLAlchemy DownloadStatus.
Args:
status: Pydantic status enum
Returns:
SQLAlchemy status enum
"""
return DBDownloadStatus(status.value)
def _status_from_db(self, status: DBDownloadStatus) -> DownloadStatus:
"""Convert SQLAlchemy DownloadStatus to Pydantic DownloadStatus.
Args:
status: SQLAlchemy status enum
Returns:
Pydantic status enum
"""
return DownloadStatus(status.value)
def _priority_to_db(self, priority: DownloadPriority) -> DBDownloadPriority:
"""Convert Pydantic DownloadPriority to SQLAlchemy DownloadPriority.
Args:
priority: Pydantic priority enum
Returns:
SQLAlchemy priority enum
"""
# Handle case differences (Pydantic uses uppercase, DB uses lowercase)
return DBDownloadPriority(priority.value.lower())
def _priority_from_db(self, priority: DBDownloadPriority) -> DownloadPriority:
"""Convert SQLAlchemy DownloadPriority to Pydantic DownloadPriority.
Args:
priority: SQLAlchemy priority enum
Returns:
Pydantic priority enum
"""
# Handle case differences (DB uses lowercase, Pydantic uses uppercase)
return DownloadPriority(priority.value.upper())
def _to_db_model(
self,
item: DownloadItem,
series_id: int,
) -> DBDownloadQueueItem:
"""Convert DownloadItem to database model.
Args:
item: Pydantic download item
series_id: Database series ID (foreign key)
Returns:
SQLAlchemy download queue item model
"""
return DBDownloadQueueItem(
series_id=series_id,
season=item.episode.season,
episode_number=item.episode.episode,
status=self._status_to_db(item.status),
priority=self._priority_to_db(item.priority),
progress_percent=item.progress.percent if item.progress else 0.0,
downloaded_bytes=int(
item.progress.downloaded_mb * 1024 * 1024
) if item.progress else 0,
total_bytes=int(
item.progress.total_mb * 1024 * 1024
) if item.progress and item.progress.total_mb else None,
download_speed=(
item.progress.speed_mbps * 1024 * 1024
) if item.progress and item.progress.speed_mbps else None,
error_message=item.error,
retry_count=item.retry_count,
download_url=str(item.source_url) if item.source_url else None,
started_at=item.started_at,
completed_at=item.completed_at,
)
def _from_db_model( def _from_db_model(
self, self,
db_item: DBDownloadQueueItem, db_item: DBDownloadQueueItem,
item_id: Optional[str] = None, item_id: Optional[str] = None,
) -> DownloadItem: ) -> DownloadItem:
"""Convert database model to DownloadItem. """Convert database model to DownloadItem.
Note: Since the database model is simplified, status, priority,
progress, and retry_count default to initial values.
Args: Args:
db_item: SQLAlchemy download queue item db_item: SQLAlchemy download queue item
item_id: Optional override for item ID (uses db ID if not provided) item_id: Optional override for item ID
Returns: Returns:
Pydantic download item Pydantic download item with default status/priority
""" """
# Build progress object if there's progress data # Get episode info from the related Episode object
progress = None episode = db_item.episode
if db_item.progress_percent > 0 or db_item.downloaded_bytes > 0: series = db_item.series
progress = DownloadProgress(
percent=db_item.progress_percent, episode_identifier = EpisodeIdentifier(
downloaded_mb=db_item.downloaded_bytes / (1024 * 1024), season=episode.season if episode else 1,
total_mb=( episode=episode.episode_number if episode else 1,
db_item.total_bytes / (1024 * 1024) title=episode.title if episode else None,
if db_item.total_bytes else None )
),
speed_mbps=(
db_item.download_speed / (1024 * 1024)
if db_item.download_speed else None
),
)
return DownloadItem( return DownloadItem(
id=item_id or str(db_item.id), id=item_id or str(db_item.id),
serie_id=db_item.series.key if db_item.series else "", serie_id=series.key if series else "",
serie_folder=db_item.series.folder if db_item.series else "", serie_folder=series.folder if series else "",
serie_name=db_item.series.name if db_item.series else "", serie_name=series.name if series else "",
episode=EpisodeIdentifier( episode=episode_identifier,
season=db_item.season, status=DownloadStatus.PENDING, # Default - managed in-memory
episode=db_item.episode_number, priority=DownloadPriority.NORMAL, # Default - managed in-memory
),
status=self._status_from_db(db_item.status),
priority=self._priority_from_db(db_item.priority),
added_at=db_item.created_at or datetime.now(timezone.utc), added_at=db_item.created_at or datetime.now(timezone.utc),
started_at=db_item.started_at, started_at=db_item.started_at,
completed_at=db_item.completed_at, completed_at=db_item.completed_at,
progress=progress, progress=None, # Managed in-memory
error=db_item.error_message, error=db_item.error_message,
retry_count=db_item.retry_count, retry_count=0, # Managed in-memory
source_url=db_item.download_url, source_url=db_item.download_url,
) )
# ========================================================================= # =========================================================================
# CRUD Operations # CRUD Operations
# ========================================================================= # =========================================================================
async def save_item( async def save_item(
self, self,
item: DownloadItem, item: DownloadItem,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> DownloadItem: ) -> DownloadItem:
"""Save a download item to the database. """Save a download item to the database.
Creates a new record if the item doesn't exist in the database. Creates a new record if the item doesn't exist in the database.
Note: Status, priority, progress, and retry_count are NOT persisted.
Args: Args:
item: Download item to save item: Download item to save
db: Optional existing database session db: Optional existing database session
Returns: Returns:
Saved download item with database ID Saved download item with database ID
Raises: Raises:
QueueRepositoryError: If save operation fails QueueRepositoryError: If save operation fails
""" """
session = db or self._db_session_factory() session = db or self._db_session_factory()
manage_session = db is None manage_session = db is None
try: try:
# Find series by key # Find series by key
series = await AnimeSeriesService.get_by_key(session, item.serie_id) series = await AnimeSeriesService.get_by_key(session, item.serie_id)
if not series: if not series:
# Create series if it doesn't exist # Create series if it doesn't exist
series = await AnimeSeriesService.create( series = await AnimeSeriesService.create(
@ -235,490 +151,272 @@ class QueueRepository:
folder=item.serie_folder, folder=item.serie_folder,
) )
logger.info( logger.info(
"Created new series for queue item", "Created new series for queue item: key=%s, name=%s",
key=item.serie_id, item.serie_id,
name=item.serie_name, item.serie_name,
) )
# Find or create episode
episode = await EpisodeService.get_by_episode(
session,
series.id,
item.episode.season,
item.episode.episode,
)
if not episode:
# Create episode if it doesn't exist
episode = await EpisodeService.create(
db=session,
series_id=series.id,
season=item.episode.season,
episode_number=item.episode.episode,
title=item.episode.title,
)
logger.info(
"Created new episode for queue item: S%02dE%02d",
item.episode.season,
item.episode.episode,
)
# Create queue item # Create queue item
db_item = await DownloadQueueService.create( db_item = await DownloadQueueService.create(
db=session, db=session,
series_id=series.id, series_id=series.id,
season=item.episode.season, episode_id=episode.id,
episode_number=item.episode.episode,
priority=self._priority_to_db(item.priority),
download_url=str(item.source_url) if item.source_url else None, download_url=str(item.source_url) if item.source_url else None,
) )
if manage_session: if manage_session:
await session.commit() await session.commit()
# Update the item ID with the database ID # Update the item ID with the database ID
item.id = str(db_item.id) item.id = str(db_item.id)
logger.debug( logger.debug(
"Saved queue item to database", "Saved queue item to database: item_id=%s, serie_key=%s",
item_id=item.id, item.id,
serie_key=item.serie_id, item.serie_id,
) )
return item return item
except Exception as e: except Exception as e:
if manage_session: if manage_session:
await session.rollback() await session.rollback()
logger.error("Failed to save queue item", error=str(e)) logger.error("Failed to save queue item: %s", e)
raise QueueRepositoryError(f"Failed to save item: {str(e)}") from e raise QueueRepositoryError(f"Failed to save item: {e}") from e
finally: finally:
if manage_session: if manage_session:
await session.close() await session.close()
async def get_item( async def get_item(
self, self,
item_id: str, item_id: str,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> Optional[DownloadItem]: ) -> Optional[DownloadItem]:
"""Get a download item by ID. """Get a download item by ID.
Args: Args:
item_id: Download item ID (database ID as string) item_id: Download item ID (database ID as string)
db: Optional existing database session db: Optional existing database session
Returns: Returns:
Download item or None if not found Download item or None if not found
Raises: Raises:
QueueRepositoryError: If query fails QueueRepositoryError: If query fails
""" """
session = db or self._db_session_factory() session = db or self._db_session_factory()
manage_session = db is None manage_session = db is None
try: try:
db_item = await DownloadQueueService.get_by_id( db_item = await DownloadQueueService.get_by_id(
session, int(item_id) session, int(item_id)
) )
if not db_item: if not db_item:
return None return None
return self._from_db_model(db_item, item_id) return self._from_db_model(db_item, item_id)
except ValueError: except ValueError:
# Invalid ID format # Invalid ID format
return None return None
except Exception as e: except Exception as e:
logger.error("Failed to get queue item", error=str(e)) logger.error("Failed to get queue item: %s", e)
raise QueueRepositoryError(f"Failed to get item: {str(e)}") from e raise QueueRepositoryError(f"Failed to get item: {e}") from e
finally: finally:
if manage_session: if manage_session:
await session.close() await session.close()
async def get_pending_items( async def get_all_items(
self, self,
limit: Optional[int] = None,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> List[DownloadItem]: ) -> List[DownloadItem]:
"""Get pending download items ordered by priority. """Get all download items regardless of status.
Note: All items are returned with default status (PENDING) since
status is now managed in-memory by the DownloadService.
Args: Args:
limit: Optional maximum number of items to return
db: Optional existing database session db: Optional existing database session
Returns: Returns:
List of pending download items List of all download items
Raises: Raises:
QueueRepositoryError: If query fails QueueRepositoryError: If query fails
""" """
session = db or self._db_session_factory() session = db or self._db_session_factory()
manage_session = db is None manage_session = db is None
try: try:
db_items = await DownloadQueueService.get_pending(session, limit) db_items = await DownloadQueueService.get_all(
return [self._from_db_model(item) for item in db_items] session, with_series=True
except Exception as e:
logger.error("Failed to get pending items", error=str(e))
raise QueueRepositoryError(
f"Failed to get pending items: {str(e)}"
) from e
finally:
if manage_session:
await session.close()
async def get_active_item(
self,
db: Optional[AsyncSession] = None,
) -> Optional[DownloadItem]:
"""Get the currently active (downloading) item.
Args:
db: Optional existing database session
Returns:
Active download item or None if none active
Raises:
QueueRepositoryError: If query fails
"""
session = db or self._db_session_factory()
manage_session = db is None
try:
db_items = await DownloadQueueService.get_active(session)
if not db_items:
return None
# Return first active item (should only be one)
return self._from_db_model(db_items[0])
except Exception as e:
logger.error("Failed to get active item", error=str(e))
raise QueueRepositoryError(
f"Failed to get active item: {str(e)}"
) from e
finally:
if manage_session:
await session.close()
async def get_completed_items(
self,
limit: int = 100,
db: Optional[AsyncSession] = None,
) -> List[DownloadItem]:
"""Get completed download items.
Args:
limit: Maximum number of items to return
db: Optional existing database session
Returns:
List of completed download items
Raises:
QueueRepositoryError: If query fails
"""
session = db or self._db_session_factory()
manage_session = db is None
try:
db_items = await DownloadQueueService.get_by_status(
session, DBDownloadStatus.COMPLETED, limit
) )
return [self._from_db_model(item) for item in db_items] return [self._from_db_model(item) for item in db_items]
except Exception as e: except Exception as e:
logger.error("Failed to get completed items", error=str(e)) logger.error("Failed to get all items: %s", e)
raise QueueRepositoryError( raise QueueRepositoryError(f"Failed to get all items: {e}") from e
f"Failed to get completed items: {str(e)}"
) from e
finally: finally:
if manage_session: if manage_session:
await session.close() await session.close()
async def get_failed_items( async def set_error(
self,
limit: int = 50,
db: Optional[AsyncSession] = None,
) -> List[DownloadItem]:
"""Get failed download items.
Args:
limit: Maximum number of items to return
db: Optional existing database session
Returns:
List of failed download items
Raises:
QueueRepositoryError: If query fails
"""
session = db or self._db_session_factory()
manage_session = db is None
try:
db_items = await DownloadQueueService.get_by_status(
session, DBDownloadStatus.FAILED, limit
)
return [self._from_db_model(item) for item in db_items]
except Exception as e:
logger.error("Failed to get failed items", error=str(e))
raise QueueRepositoryError(
f"Failed to get failed items: {str(e)}"
) from e
finally:
if manage_session:
await session.close()
async def update_status(
self, self,
item_id: str, item_id: str,
status: DownloadStatus, error: str,
error: Optional[str] = None,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> bool: ) -> bool:
"""Update the status of a download item. """Set error message on a download item.
Args: Args:
item_id: Download item ID item_id: Download item ID
status: New download status error: Error message
error: Optional error message for failed status
db: Optional existing database session db: Optional existing database session
Returns: Returns:
True if update succeeded, False if item not found True if update succeeded, False if item not found
Raises: Raises:
QueueRepositoryError: If update fails QueueRepositoryError: If update fails
""" """
session = db or self._db_session_factory() session = db or self._db_session_factory()
manage_session = db is None manage_session = db is None
try: try:
result = await DownloadQueueService.update_status( result = await DownloadQueueService.set_error(
session, session,
int(item_id), int(item_id),
self._status_to_db(status),
error, error,
) )
if manage_session: if manage_session:
await session.commit() await session.commit()
success = result is not None success = result is not None
if success: if success:
logger.debug( logger.debug(
"Updated queue item status", "Set error on queue item: item_id=%s",
item_id=item_id, item_id,
status=status.value,
) )
return success return success
except ValueError: except ValueError:
return False return False
except Exception as e: except Exception as e:
if manage_session: if manage_session:
await session.rollback() await session.rollback()
logger.error("Failed to update status", error=str(e)) logger.error("Failed to set error: %s", e)
raise QueueRepositoryError( raise QueueRepositoryError(f"Failed to set error: {e}") from e
f"Failed to update status: {str(e)}"
) from e
finally: finally:
if manage_session: if manage_session:
await session.close() await session.close()
async def update_progress(
self,
item_id: str,
progress: float,
downloaded: int,
total: Optional[int],
speed: Optional[float],
db: Optional[AsyncSession] = None,
) -> bool:
"""Update download progress for an item.
Args:
item_id: Download item ID
progress: Progress percentage (0-100)
downloaded: Downloaded bytes
total: Total bytes (optional)
speed: Download speed in bytes/second (optional)
db: Optional existing database session
Returns:
True if update succeeded, False if item not found
Raises:
QueueRepositoryError: If update fails
"""
session = db or self._db_session_factory()
manage_session = db is None
try:
result = await DownloadQueueService.update_progress(
session,
int(item_id),
progress,
downloaded,
total,
speed,
)
if manage_session:
await session.commit()
return result is not None
except ValueError:
return False
except Exception as e:
if manage_session:
await session.rollback()
logger.error("Failed to update progress", error=str(e))
raise QueueRepositoryError(
f"Failed to update progress: {str(e)}"
) from e
finally:
if manage_session:
await session.close()
async def delete_item( async def delete_item(
self, self,
item_id: str, item_id: str,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> bool: ) -> bool:
"""Delete a download item from the database. """Delete a download item from the database.
Args: Args:
item_id: Download item ID item_id: Download item ID
db: Optional existing database session db: Optional existing database session
Returns: Returns:
True if item was deleted, False if not found True if item was deleted, False if not found
Raises: Raises:
QueueRepositoryError: If delete fails QueueRepositoryError: If delete fails
""" """
session = db or self._db_session_factory() session = db or self._db_session_factory()
manage_session = db is None manage_session = db is None
try: try:
result = await DownloadQueueService.delete(session, int(item_id)) result = await DownloadQueueService.delete(session, int(item_id))
if manage_session: if manage_session:
await session.commit() await session.commit()
if result: if result:
logger.debug("Deleted queue item", item_id=item_id) logger.debug("Deleted queue item: item_id=%s", item_id)
return result return result
except ValueError: except ValueError:
return False return False
except Exception as e: except Exception as e:
if manage_session: if manage_session:
await session.rollback() await session.rollback()
logger.error("Failed to delete item", error=str(e)) logger.error("Failed to delete item: %s", e)
raise QueueRepositoryError( raise QueueRepositoryError(f"Failed to delete item: {e}") from e
f"Failed to delete item: {str(e)}"
) from e
finally: finally:
if manage_session: if manage_session:
await session.close() await session.close()
async def clear_completed( async def clear_all(
self, self,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> int: ) -> int:
"""Clear all completed download items. """Clear all download items from the queue.
Args: Args:
db: Optional existing database session db: Optional existing database session
Returns: Returns:
Number of items cleared Number of items cleared
Raises: Raises:
QueueRepositoryError: If operation fails QueueRepositoryError: If operation fails
""" """
session = db or self._db_session_factory() session = db or self._db_session_factory()
manage_session = db is None manage_session = db is None
try: try:
count = await DownloadQueueService.clear_completed(session) # Get all items first to count them
all_items = await DownloadQueueService.get_all(session)
count = len(all_items)
# Delete each item
for item in all_items:
await DownloadQueueService.delete(session, item.id)
if manage_session: if manage_session:
await session.commit() await session.commit()
logger.info("Cleared completed items from queue", count=count) logger.info("Cleared all items from queue: count=%d", count)
return count return count
except Exception as e: except Exception as e:
if manage_session: if manage_session:
await session.rollback() await session.rollback()
logger.error("Failed to clear completed items", error=str(e)) logger.error("Failed to clear queue: %s", e)
raise QueueRepositoryError( raise QueueRepositoryError(f"Failed to clear queue: {e}") from e
f"Failed to clear completed: {str(e)}"
) from e
finally:
if manage_session:
await session.close()
async def get_all_items(
self,
db: Optional[AsyncSession] = None,
) -> List[DownloadItem]:
"""Get all download items regardless of status.
Args:
db: Optional existing database session
Returns:
List of all download items
Raises:
QueueRepositoryError: If query fails
"""
session = db or self._db_session_factory()
manage_session = db is None
try:
db_items = await DownloadQueueService.get_all(
session, with_series=True
)
return [self._from_db_model(item) for item in db_items]
except Exception as e:
logger.error("Failed to get all items", error=str(e))
raise QueueRepositoryError(
f"Failed to get all items: {str(e)}"
) from e
finally:
if manage_session:
await session.close()
async def retry_failed_items(
self,
max_retries: int = 3,
db: Optional[AsyncSession] = None,
) -> List[DownloadItem]:
"""Retry failed downloads that haven't exceeded max retries.
Args:
max_retries: Maximum number of retry attempts
db: Optional existing database session
Returns:
List of items marked for retry
Raises:
QueueRepositoryError: If operation fails
"""
session = db or self._db_session_factory()
manage_session = db is None
try:
db_items = await DownloadQueueService.retry_failed(
session, max_retries
)
if manage_session:
await session.commit()
return [self._from_db_model(item) for item in db_items]
except Exception as e:
if manage_session:
await session.rollback()
logger.error("Failed to retry failed items", error=str(e))
raise QueueRepositoryError(
f"Failed to retry failed items: {str(e)}"
) from e
finally: finally:
if manage_session: if manage_session:
await session.close() await session.close()
@ -732,22 +430,31 @@ def get_queue_repository(
db_session_factory: Optional[Callable[[], AsyncSession]] = None, db_session_factory: Optional[Callable[[], AsyncSession]] = None,
) -> QueueRepository: ) -> QueueRepository:
"""Get or create the QueueRepository singleton. """Get or create the QueueRepository singleton.
Args: Args:
db_session_factory: Optional factory function for database sessions. db_session_factory: Optional factory function for database sessions.
If not provided, uses default from connection module. If not provided, uses default from connection module.
Returns: Returns:
QueueRepository singleton instance QueueRepository singleton instance
""" """
global _queue_repository_instance global _queue_repository_instance
if _queue_repository_instance is None: if _queue_repository_instance is None:
if db_session_factory is None: if db_session_factory is None:
# Use default session factory # Use default session factory
from src.server.database.connection import get_async_session_factory from src.server.database.connection import get_async_session_factory
db_session_factory = get_async_session_factory db_session_factory = get_async_session_factory
_queue_repository_instance = QueueRepository(db_session_factory) _queue_repository_instance = QueueRepository(db_session_factory)
return _queue_repository_instance return _queue_repository_instance
def reset_queue_repository() -> None:
"""Reset the QueueRepository singleton.
Used for testing to ensure fresh state between tests.
"""
global _queue_repository_instance
_queue_repository_instance = None

8
src/server/services/scan_service.py
View File

@ -415,7 +415,7 @@ class ScanService:
message="Initializing scan...", message="Initializing scan...",
) )
except Exception as e: except Exception as e:
logger.error("Failed to start progress tracking", error=str(e)) logger.error("Failed to start progress tracking: %s", e)
# Emit scan started event # Emit scan started event
await self._emit_scan_event({ await self._emit_scan_event({
@ -479,7 +479,7 @@ class ScanService:
folder=scan_progress.folder, folder=scan_progress.folder,
) )
except Exception as e: except Exception as e:
logger.debug("Progress update skipped", error=str(e)) logger.debug("Progress update skipped: %s", e)
# Emit progress event with key as primary identifier # Emit progress event with key as primary identifier
await self._emit_scan_event({ await self._emit_scan_event({
@ -541,7 +541,7 @@ class ScanService:
error_message=completion_context.message, error_message=completion_context.message,
) )
except Exception as e: except Exception as e:
logger.debug("Progress completion skipped", error=str(e)) logger.debug("Progress completion skipped: %s", e)
# Emit completion event # Emit completion event
await self._emit_scan_event({ await self._emit_scan_event({
@ -598,7 +598,7 @@ class ScanService:
error_message="Scan cancelled by user", error_message="Scan cancelled by user",
) )
except Exception as e: except Exception as e:
logger.debug("Progress cancellation skipped", error=str(e)) logger.debug("Progress cancellation skipped: %s", e)
logger.info("Scan cancelled") logger.info("Scan cancelled")
return True return True

103
tests/unit/test_download_service.py
View File

@ -25,8 +25,11 @@ from src.server.services.download_service import DownloadService, DownloadServic
class MockQueueRepository: class MockQueueRepository:
"""Mock implementation of QueueRepository for testing. """Mock implementation of QueueRepository for testing.
This provides an in-memory storage that mimics the database repository This provides an in-memory storage that mimics the simplified database
behavior without requiring actual database connections. repository behavior without requiring actual database connections.
Note: The repository is simplified - status, priority, progress are
now managed in-memory by DownloadService, not stored in database.
""" """
def __init__(self): def __init__(self):
@ -42,81 +45,19 @@ class MockQueueRepository:
"""Get item by ID from in-memory storage.""" """Get item by ID from in-memory storage."""
return self._items.get(item_id) return self._items.get(item_id)
async def get_pending_items(self) -> List[DownloadItem]: async def get_all_items(self) -> List[DownloadItem]:
"""Get all pending items.""" """Get all items in storage."""
return [ return list(self._items.values())
item for item in self._items.values()
if item.status == DownloadStatus.PENDING
]
async def get_active_item(self) -> Optional[DownloadItem]: async def set_error(
"""Get the currently active item."""
for item in self._items.values():
if item.status == DownloadStatus.DOWNLOADING:
return item
return None
async def get_completed_items(
self, limit: int = 100
) -> List[DownloadItem]:
"""Get completed items."""
completed = [
item for item in self._items.values()
if item.status == DownloadStatus.COMPLETED
]
return completed[:limit]
async def get_failed_items(self, limit: int = 50) -> List[DownloadItem]:
"""Get failed items."""
failed = [
item for item in self._items.values()
if item.status == DownloadStatus.FAILED
]
return failed[:limit]
async def update_status(
self, self,
item_id: str, item_id: str,
status: DownloadStatus, error: str,
error: Optional[str] = None
) -> bool: ) -> bool:
"""Update item status.""" """Set error message on an item."""
if item_id not in self._items: if item_id not in self._items:
return False return False
self._items[item_id].status = status self._items[item_id].error = error
if error:
self._items[item_id].error = error
if status == DownloadStatus.COMPLETED:
self._items[item_id].completed_at = datetime.now(timezone.utc)
elif status == DownloadStatus.DOWNLOADING:
self._items[item_id].started_at = datetime.now(timezone.utc)
return True
async def update_progress(
self,
item_id: str,
progress: float,
downloaded: int,
total: int,
speed: float
) -> bool:
"""Update download progress."""
if item_id not in self._items:
return False
item = self._items[item_id]
if item.progress is None:
from src.server.models.download import DownloadProgress
item.progress = DownloadProgress(
percent=progress,
downloaded_bytes=downloaded,
total_bytes=total,
speed_bps=speed
)
else:
item.progress.percent = progress
item.progress.downloaded_bytes = downloaded
item.progress.total_bytes = total
item.progress.speed_bps = speed
return True return True
async def delete_item(self, item_id: str) -> bool: async def delete_item(self, item_id: str) -> bool:
@ -126,15 +67,11 @@ class MockQueueRepository:
return True return True
return False return False
async def clear_completed(self) -> int: async def clear_all(self) -> int:
"""Clear all completed items.""" """Clear all items."""
completed_ids = [ count = len(self._items)
item_id for item_id, item in self._items.items() self._items.clear()
if item.status == DownloadStatus.COMPLETED return count
]
for item_id in completed_ids:
del self._items[item_id]
return len(completed_ids)
@pytest.fixture @pytest.fixture
@ -505,9 +442,9 @@ class TestPersistence:
) )
# Item should be saved in mock repository # Item should be saved in mock repository
pending_items = await mock_queue_repository.get_pending_items() all_items = await mock_queue_repository.get_all_items()
assert len(pending_items) == 1 assert len(all_items) == 1
assert pending_items[0].serie_id == "series-1" assert all_items[0].serie_id == "series-1"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_queue_recovery_after_restart( async def test_queue_recovery_after_restart(