refactor(api): replace direct celery tasks with queue manager in bulk add
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import re
|
||||
from typing import List, Dict, Any
|
||||
from fastapi import APIRouter, Request, Depends
|
||||
from typing import List, Dict, Any, Dict, Any
|
||||
from fastapi import APIRouter, Request, Depends, Request, Depends
|
||||
from pydantic import BaseModel
|
||||
import logging
|
||||
|
||||
@@ -11,7 +11,10 @@ from routes.auth.middleware import require_auth_from_state, User
|
||||
from routes.utils.get_info import get_spotify_info
|
||||
from routes.utils.celery_queue_manager import download_queue_manager
|
||||
|
||||
# Assuming these imports are available for queue management and Spotify info
|
||||
# Import authentication dependencies
|
||||
from routes.auth.middleware import require_auth_from_state, User
|
||||
|
||||
# Import queue management and Spotify info
|
||||
from routes.utils.get_info import (
|
||||
get_client,
|
||||
get_track,
|
||||
@@ -19,6 +22,7 @@ from routes.utils.get_info import (
|
||||
get_playlist,
|
||||
get_artist,
|
||||
)
|
||||
from routes.utils.celery_queue_manager import download_queue_manager
|
||||
|
||||
router = APIRouter()
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -29,6 +33,7 @@ class BulkAddLinksRequest(BaseModel):
|
||||
|
||||
|
||||
@router.post("/bulk-add-spotify-links")
|
||||
async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, current_user: User = Depends(require_auth_from_state)):
|
||||
async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, current_user: User = Depends(require_auth_from_state)):
|
||||
added_count = 0
|
||||
failed_links = []
|
||||
@@ -53,6 +58,7 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, cur
|
||||
spotify_type = match.group(1)
|
||||
spotify_id = match.group(2)
|
||||
logger.debug(f"Extracted from link: spotify_type={spotify_type}, spotify_id={spotify_id}")
|
||||
logger.debug(f"Extracted from link: spotify_type={spotify_type}, spotify_id={spotify_id}")
|
||||
|
||||
try:
|
||||
# Get basic info to confirm existence and get name/artist
|
||||
@@ -102,13 +108,29 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, cur
|
||||
# Add to download queue using the queue manager
|
||||
task_id = download_queue_manager.add_task(task_data)
|
||||
|
||||
if task_id:
|
||||
added_count += 1
|
||||
logger.debug(f"Added {added_count}/{total_links} {spotify_type} '{item_name}' ({spotify_id}) to queue with task_id: {task_id}.")
|
||||
# Prepare task data for the queue manager
|
||||
task_data = {
|
||||
"download_type": spotify_type,
|
||||
"url": spotify_url,
|
||||
"name": item_name,
|
||||
"artist": artist_name,
|
||||
"spotify_id": spotify_id,
|
||||
"type": spotify_type,
|
||||
"username": current_user.username,
|
||||
"orig_request": dict(req.query_params),
|
||||
}
|
||||
|
||||
# Add to download queue using the queue manager
|
||||
task_id = download_queue_manager.add_task(task_data)
|
||||
|
||||
if task_id:
|
||||
added_count += 1
|
||||
logger.debug(f"Added {added_count}/{total_links} {spotify_type} '{item_name}' ({spotify_id}) to queue with task_id: {task_id}.")
|
||||
else:
|
||||
logger.warning(
|
||||
f"Unsupported Spotify type for download: {spotify_type} for link: {link}"
|
||||
)
|
||||
logger.warning(f"Failed to add {spotify_type} '{item_name}' ({spotify_id}) to queue.")
|
||||
failed_links.append(link)
|
||||
continue
|
||||
|
||||
|
||||
Reference in New Issue
Block a user