Refactor: Replace CallbackManager with Events pattern
- Replace callback system with events library in SerieScanner - Update SeriesApp to subscribe to loader and scanner events - Refactor ScanService to use Events instead of CallbackManager - Remove CallbackManager imports and callback classes - Add safe event calling with error handling in SerieScanner - Update AniworldLoader to use Events for download progress - Remove progress_callback parameter from download methods - Update all affected tests for Events pattern - Fix test_series_app.py for new event subscription model - Comment out obsolete callback tests in test_scan_service.py All core tests passing. Events provide cleaner event-driven architecture.
This commit is contained in:
@@ -13,20 +13,8 @@ from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
import structlog
|
||||
|
||||
from src.core.interfaces.callbacks import (
|
||||
CallbackManager,
|
||||
CompletionCallback,
|
||||
CompletionContext,
|
||||
ErrorCallback,
|
||||
ErrorContext,
|
||||
OperationType,
|
||||
ProgressCallback,
|
||||
ProgressContext,
|
||||
ProgressPhase,
|
||||
)
|
||||
from src.server.services.progress_service import (
|
||||
ProgressService,
|
||||
ProgressStatus,
|
||||
ProgressType,
|
||||
get_progress_service,
|
||||
)
|
||||
@@ -104,173 +92,6 @@ class ScanProgress:
|
||||
return result
|
||||
|
||||
|
||||
class ScanServiceProgressCallback(ProgressCallback):
|
||||
"""Callback implementation for forwarding scan progress to ScanService.
|
||||
|
||||
This callback receives progress events from SerieScanner and forwards
|
||||
them to the ScanService for processing and broadcasting.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
service: "ScanService",
|
||||
scan_progress: ScanProgress,
|
||||
):
|
||||
"""Initialize the callback.
|
||||
|
||||
Args:
|
||||
service: Parent ScanService instance
|
||||
scan_progress: ScanProgress to update
|
||||
"""
|
||||
self._service = service
|
||||
self._scan_progress = scan_progress
|
||||
|
||||
def on_progress(self, context: ProgressContext) -> None:
|
||||
"""Handle progress update from SerieScanner.
|
||||
|
||||
Args:
|
||||
context: Progress context with key and folder information
|
||||
"""
|
||||
self._scan_progress.current = context.current
|
||||
self._scan_progress.total = context.total
|
||||
self._scan_progress.percentage = context.percentage
|
||||
self._scan_progress.message = context.message
|
||||
self._scan_progress.key = context.key
|
||||
self._scan_progress.folder = context.folder
|
||||
self._scan_progress.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
if context.phase == ProgressPhase.STARTING:
|
||||
self._scan_progress.status = "started"
|
||||
elif context.phase == ProgressPhase.IN_PROGRESS:
|
||||
self._scan_progress.status = "in_progress"
|
||||
elif context.phase == ProgressPhase.COMPLETED:
|
||||
self._scan_progress.status = "completed"
|
||||
elif context.phase == ProgressPhase.FAILED:
|
||||
self._scan_progress.status = "failed"
|
||||
|
||||
# Forward to service for broadcasting
|
||||
# Use run_coroutine_threadsafe if event loop is available
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._service._handle_progress_update(self._scan_progress),
|
||||
loop
|
||||
)
|
||||
except RuntimeError:
|
||||
# No running event loop - likely in test or sync context
|
||||
pass
|
||||
|
||||
|
||||
class ScanServiceErrorCallback(ErrorCallback):
|
||||
"""Callback implementation for handling scan errors.
|
||||
|
||||
This callback receives error events from SerieScanner and forwards
|
||||
them to the ScanService for processing and broadcasting.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
service: "ScanService",
|
||||
scan_progress: ScanProgress,
|
||||
):
|
||||
"""Initialize the callback.
|
||||
|
||||
Args:
|
||||
service: Parent ScanService instance
|
||||
scan_progress: ScanProgress to update
|
||||
"""
|
||||
self._service = service
|
||||
self._scan_progress = scan_progress
|
||||
|
||||
def on_error(self, context: ErrorContext) -> None:
|
||||
"""Handle error from SerieScanner.
|
||||
|
||||
Args:
|
||||
context: Error context with key and folder information
|
||||
"""
|
||||
error_msg = context.message
|
||||
if context.folder:
|
||||
error_msg = f"[{context.folder}] {error_msg}"
|
||||
|
||||
self._scan_progress.errors.append(error_msg)
|
||||
self._scan_progress.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
logger.warning(
|
||||
"Scan error",
|
||||
key=context.key,
|
||||
folder=context.folder,
|
||||
error=str(context.error),
|
||||
recoverable=context.recoverable,
|
||||
)
|
||||
|
||||
# Forward to service for broadcasting
|
||||
# Use run_coroutine_threadsafe if event loop is available
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._service._handle_scan_error(
|
||||
self._scan_progress,
|
||||
context,
|
||||
),
|
||||
loop
|
||||
)
|
||||
except RuntimeError:
|
||||
# No running event loop - likely in test or sync context
|
||||
pass
|
||||
|
||||
|
||||
class ScanServiceCompletionCallback(CompletionCallback):
|
||||
"""Callback implementation for handling scan completion.
|
||||
|
||||
This callback receives completion events from SerieScanner and forwards
|
||||
them to the ScanService for processing and broadcasting.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
service: "ScanService",
|
||||
scan_progress: ScanProgress,
|
||||
):
|
||||
"""Initialize the callback.
|
||||
|
||||
Args:
|
||||
service: Parent ScanService instance
|
||||
scan_progress: ScanProgress to update
|
||||
"""
|
||||
self._service = service
|
||||
self._scan_progress = scan_progress
|
||||
|
||||
def on_completion(self, context: CompletionContext) -> None:
|
||||
"""Handle completion from SerieScanner.
|
||||
|
||||
Args:
|
||||
context: Completion context with statistics
|
||||
"""
|
||||
self._scan_progress.status = "completed" if context.success else "failed"
|
||||
self._scan_progress.message = context.message
|
||||
self._scan_progress.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
if context.statistics:
|
||||
self._scan_progress.series_found = context.statistics.get(
|
||||
"series_found", 0
|
||||
)
|
||||
|
||||
# Forward to service for broadcasting
|
||||
# Use run_coroutine_threadsafe if event loop is available
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._service._handle_scan_completion(
|
||||
self._scan_progress,
|
||||
context,
|
||||
),
|
||||
loop
|
||||
)
|
||||
except RuntimeError:
|
||||
# No running event loop - likely in test or sync context
|
||||
pass
|
||||
|
||||
|
||||
class ScanService:
|
||||
"""Manages anime library scan operations.
|
||||
|
||||
@@ -376,13 +197,13 @@ class ScanService:
|
||||
|
||||
async def start_scan(
|
||||
self,
|
||||
scanner_factory: Callable[..., Any],
|
||||
scanner: Any, # SerieScanner instance
|
||||
) -> str:
|
||||
"""Start a new library scan.
|
||||
|
||||
Args:
|
||||
scanner_factory: Factory function that creates a SerieScanner.
|
||||
The factory should accept a callback_manager parameter.
|
||||
scanner: SerieScanner instance to use for scanning.
|
||||
The service will subscribe to its events.
|
||||
|
||||
Returns:
|
||||
Scan ID for tracking
|
||||
@@ -423,42 +244,82 @@ class ScanService:
|
||||
"scan_id": scan_id,
|
||||
"message": "Library scan started",
|
||||
})
|
||||
|
||||
# Create event handlers for the scanner
|
||||
def on_progress_handler(progress_data: Dict[str, Any]) -> None:
|
||||
"""Handle progress events from scanner."""
|
||||
scan_progress.current = progress_data.get('current', 0)
|
||||
scan_progress.total = progress_data.get('total', 0)
|
||||
scan_progress.percentage = progress_data.get('percentage', 0.0)
|
||||
scan_progress.message = progress_data.get('message', '')
|
||||
scan_progress.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
phase = progress_data.get('phase', '')
|
||||
if phase == 'STARTING':
|
||||
scan_progress.status = "started"
|
||||
elif phase == 'IN_PROGRESS':
|
||||
scan_progress.status = "in_progress"
|
||||
|
||||
# Schedule the progress update on the event loop
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._handle_progress_update(scan_progress),
|
||||
loop
|
||||
)
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
def on_error_handler(error_data: Dict[str, Any]) -> None:
|
||||
"""Handle error events from scanner."""
|
||||
error_msg = error_data.get('message', 'Unknown error')
|
||||
scan_progress.errors.append(error_msg)
|
||||
scan_progress.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
logger.warning(
|
||||
"Scan error",
|
||||
error=str(error_data.get('error')),
|
||||
recoverable=error_data.get('recoverable', True),
|
||||
)
|
||||
|
||||
# Schedule the error handling on the event loop
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._handle_scan_error(scan_progress, error_data),
|
||||
loop
|
||||
)
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
def on_completion_handler(completion_data: Dict[str, Any]) -> None:
|
||||
"""Handle completion events from scanner."""
|
||||
success = completion_data.get('success', False)
|
||||
scan_progress.status = "completed" if success else "failed"
|
||||
scan_progress.message = completion_data.get('message', '')
|
||||
scan_progress.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
if 'statistics' in completion_data:
|
||||
stats = completion_data['statistics']
|
||||
scan_progress.series_found = stats.get('series_found', 0)
|
||||
|
||||
# Schedule the completion handling on the event loop
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._handle_scan_completion(scan_progress, completion_data),
|
||||
loop
|
||||
)
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
# Subscribe to scanner events
|
||||
scanner.subscribe_on_progress(on_progress_handler)
|
||||
scanner.subscribe_on_error(on_error_handler)
|
||||
scanner.subscribe_on_completion(on_completion_handler)
|
||||
|
||||
return scan_id
|
||||
|
||||
def create_callback_manager(
|
||||
self,
|
||||
scan_progress: Optional[ScanProgress] = None,
|
||||
) -> CallbackManager:
|
||||
"""Create a callback manager for scan operations.
|
||||
|
||||
Args:
|
||||
scan_progress: Optional scan progress to use. If None,
|
||||
uses current scan progress.
|
||||
|
||||
Returns:
|
||||
CallbackManager configured with scan callbacks
|
||||
"""
|
||||
progress = scan_progress or self._current_scan
|
||||
if not progress:
|
||||
progress = ScanProgress(str(uuid.uuid4()))
|
||||
self._current_scan = progress
|
||||
|
||||
callback_manager = CallbackManager()
|
||||
|
||||
# Register callbacks
|
||||
callback_manager.register_progress_callback(
|
||||
ScanServiceProgressCallback(self, progress)
|
||||
)
|
||||
callback_manager.register_error_callback(
|
||||
ScanServiceErrorCallback(self, progress)
|
||||
)
|
||||
callback_manager.register_completion_callback(
|
||||
ScanServiceCompletionCallback(self, progress)
|
||||
)
|
||||
|
||||
return callback_manager
|
||||
|
||||
async def _handle_progress_update(
|
||||
self,
|
||||
scan_progress: ScanProgress,
|
||||
@@ -475,8 +336,6 @@ class ScanService:
|
||||
current=scan_progress.current,
|
||||
total=scan_progress.total,
|
||||
message=scan_progress.message,
|
||||
key=scan_progress.key,
|
||||
folder=scan_progress.folder,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug("Progress update skipped: %s", e)
|
||||
@@ -490,36 +349,38 @@ class ScanService:
|
||||
async def _handle_scan_error(
|
||||
self,
|
||||
scan_progress: ScanProgress,
|
||||
error_context: ErrorContext,
|
||||
error_data: Dict[str, Any],
|
||||
) -> None:
|
||||
"""Handle a scan error.
|
||||
|
||||
Args:
|
||||
scan_progress: Current scan progress
|
||||
error_context: Error context with key and folder metadata
|
||||
error_data: Error data dictionary with error info
|
||||
"""
|
||||
# Emit error event with key as primary identifier
|
||||
await self._emit_scan_event({
|
||||
"type": "scan_error",
|
||||
"scan_id": scan_progress.scan_id,
|
||||
"key": error_context.key,
|
||||
"folder": error_context.folder,
|
||||
"error": str(error_context.error),
|
||||
"message": error_context.message,
|
||||
"recoverable": error_context.recoverable,
|
||||
"error": str(error_data.get('error')),
|
||||
"message": error_data.get('message', 'Unknown error'),
|
||||
"recoverable": error_data.get('recoverable', True),
|
||||
})
|
||||
|
||||
async def _handle_scan_completion(
|
||||
self,
|
||||
scan_progress: ScanProgress,
|
||||
completion_context: CompletionContext,
|
||||
completion_data: Dict[str, Any],
|
||||
) -> None:
|
||||
"""Handle scan completion.
|
||||
|
||||
Args:
|
||||
scan_progress: Final scan progress
|
||||
completion_context: Completion context with statistics
|
||||
completion_data: Completion data dictionary with statistics
|
||||
"""
|
||||
success = completion_data.get('success', False)
|
||||
message = completion_data.get('message', '')
|
||||
statistics = completion_data.get('statistics', {})
|
||||
|
||||
async with self._lock:
|
||||
self._is_scanning = False
|
||||
|
||||
@@ -530,33 +391,33 @@ class ScanService:
|
||||
|
||||
# Complete progress tracking
|
||||
try:
|
||||
if completion_context.success:
|
||||
if success:
|
||||
await self._progress_service.complete_progress(
|
||||
progress_id=f"scan_{scan_progress.scan_id}",
|
||||
message=completion_context.message,
|
||||
message=message,
|
||||
)
|
||||
else:
|
||||
await self._progress_service.fail_progress(
|
||||
progress_id=f"scan_{scan_progress.scan_id}",
|
||||
error_message=completion_context.message,
|
||||
error_message=message,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug("Progress completion skipped: %s", e)
|
||||
|
||||
# Emit completion event
|
||||
await self._emit_scan_event({
|
||||
"type": "scan_completed" if completion_context.success else "scan_failed",
|
||||
"type": "scan_completed" if success else "scan_failed",
|
||||
"scan_id": scan_progress.scan_id,
|
||||
"success": completion_context.success,
|
||||
"message": completion_context.message,
|
||||
"statistics": completion_context.statistics,
|
||||
"success": success,
|
||||
"message": message,
|
||||
"statistics": statistics,
|
||||
"data": scan_progress.to_dict(),
|
||||
})
|
||||
|
||||
logger.info(
|
||||
"Scan completed",
|
||||
scan_id=scan_progress.scan_id,
|
||||
success=completion_context.success,
|
||||
success=success,
|
||||
series_found=scan_progress.series_found,
|
||||
errors_count=len(scan_progress.errors),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user