From d83e320a82d3ad98626f074a5bc0abf661f652dd Mon Sep 17 00:00:00 2001 From: Phlogi Date: Wed, 27 Aug 2025 09:43:01 +0200 Subject: [PATCH 1/3] refactor(api): replace direct celery tasks with queue manager in bulk add --- routes/content/bulk_add.py | 68 ++++++++++++++++---------------------- 1 file changed, 29 insertions(+), 39 deletions(-) diff --git a/routes/content/bulk_add.py b/routes/content/bulk_add.py index b5471ea..38a2824 100644 --- a/routes/content/bulk_add.py +++ b/routes/content/bulk_add.py @@ -1,12 +1,15 @@ import re from typing import List, Dict, Any -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Request, Depends from pydantic import BaseModel import logging -# 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_spotify_info -from routes.utils.celery_tasks import download_track, download_album, download_playlist +from routes.utils.celery_queue_manager import download_queue_manager router = APIRouter() logger = logging.getLogger(__name__) @@ -15,7 +18,7 @@ class BulkAddLinksRequest(BaseModel): links: List[str] @router.post("/bulk-add-spotify-links") -async def bulk_add_spotify_links(request: BulkAddLinksRequest): +async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, current_user: User = Depends(require_auth_from_state)): added_count = 0 failed_links = [] total_links = len(request.links) @@ -32,6 +35,7 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest): spotify_type = match.group(1) spotify_id = match.group(2) + 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 @@ -54,41 +58,27 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest): # Construct URL for the download task spotify_url = f"https://open.spotify.com/{spotify_type}/{spotify_id}" - # Add to Celery queue based on type - if spotify_type == "track": - download_track.delay( - url=spotify_url, - spotify_id=spotify_id, - type=spotify_type, - name=item_name, - artist=artist_name, - download_type="track", - ) - elif spotify_type == "album": - download_album.delay( - url=spotify_url, - spotify_id=spotify_id, - type=spotify_type, - name=item_name, - artist=artist_name, - download_type="album", - ) - elif spotify_type == "playlist": - download_playlist.delay( - url=spotify_url, - spotify_id=spotify_id, - type=spotify_type, - name=item_name, - artist=artist_name, - download_type="playlist", - ) - else: - logger.warning(f"Unsupported Spotify type for download: {spotify_type} for link: {link}") - failed_links.append(link) - continue + # 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), + } - added_count += 1 - logger.debug(f"Added {added_count+1}/{total_links} {spotify_type} '{item_name}' ({spotify_id}) to queue.") + # 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"Failed to add {spotify_type} '{item_name}' ({spotify_id}) to queue.") + failed_links.append(link) except Exception as e: logger.error(f"Error processing Spotify link {link}: {e}", exc_info=True) @@ -105,4 +95,4 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest): "message": message, "count": added_count, "failed_links": failed_links, - } \ No newline at end of file + } From 957928bfa050b870027697e21dc60b3dd1827034 Mon Sep 17 00:00:00 2001 From: Phlogi Date: Wed, 27 Aug 2025 09:43:01 +0200 Subject: [PATCH 2/3] refactor(api): replace direct celery tasks with queue manager in bulk add --- routes/content/bulk_add.py | 34 ++++++++++++++++++++++++++++------ 1 file changed, 28 insertions(+), 6 deletions(-) diff --git a/routes/content/bulk_add.py b/routes/content/bulk_add.py index 52ecd12..daa0cd0 100644 --- a/routes/content/bulk_add.py +++ b/routes/content/bulk_add.py @@ -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 From 7b7e32c92378eb404dbc77c376c66f4dc15d4fb6 Mon Sep 17 00:00:00 2001 From: che-pj Date: Wed, 27 Aug 2025 21:39:08 +0200 Subject: [PATCH 3/3] fixup after merge/rebase to dev --- routes/content/bulk_add.py | 28 ++-------------------------- 1 file changed, 2 insertions(+), 26 deletions(-) diff --git a/routes/content/bulk_add.py b/routes/content/bulk_add.py index daa0cd0..61ea837 100644 --- a/routes/content/bulk_add.py +++ b/routes/content/bulk_add.py @@ -1,5 +1,5 @@ import re -from typing import List, Dict, Any, Dict, Any +from typing import List from fastapi import APIRouter, Request, Depends, Request, Depends from pydantic import BaseModel import logging @@ -33,7 +33,6 @@ 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 = [] @@ -45,7 +44,7 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, cur # but still handle potential errors during info retrieval or unsupported types # Extract type and ID from the link directly using regex match = re.match( - r"https://open\.spotify\.com(?:/intl-[a-z]{2})?/(track|album|playlist|artist)/([a-zA-Z0-9]+)(?:\?.*)?", + r"https://open\.spotify\.com(?:/[a-z]{2})?/(track|album|playlist|artist)/([a-zA-Z0-9]+)(?:\?.*)?", link, ) if not match: @@ -108,24 +107,6 @@ 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}.") @@ -134,11 +115,6 @@ async def bulk_add_spotify_links(request: BulkAddLinksRequest, req: Request, cur failed_links.append(link) continue - added_count += 1 - logger.debug( - f"Added {added_count + 1}/{total_links} {spotify_type} '{item_name}' ({spotify_id}) to queue." - ) - except Exception as e: logger.error(f"Error processing Spotify link {link}: {e}", exc_info=True) failed_links.append(link)