diff --git a/TODO.md b/TODO.md index d9b25b2..92164bd 100644 --- a/TODO.md +++ b/TODO.md @@ -94,10 +94,10 @@ ### 🎯 Milestone 10: **"Import Infrastructure" Release** – Queue & Worker Setup -* [ ] Implement import queue system -* [ ] Add background worker process for playlist imports -* [ ] Support multiple concurrent import jobs -* [ ] Implement priority handling for import jobs +* [x] Implement import queue system +* [x] Add background worker process for playlist imports +* [x] Support multiple concurrent import jobs +* [x] Implement priority handling for import jobs ### 🎯 Milestone 11: **"Progress Pulse" Release** – Import Status Tracking diff --git a/musicround/__init__.py b/musicround/__init__.py index 0fae7cc..b2504db 100644 --- a/musicround/__init__.py +++ b/musicround/__init__.py @@ -335,6 +335,16 @@ def create_app(config=None): # Register error handlers from musicround.errors import register_error_handlers register_error_handlers(app) + + # Initialize import queue and background workers + from musicround.helpers.import_queue import ImportQueue, ImportWorker + worker_count = int(os.environ.get('IMPORT_WORKER_COUNT', '2')) + import_queue = ImportQueue() + workers = [ImportWorker(app, import_queue) for _ in range(worker_count)] + for w in workers: + w.start() + app.config['import_queue'] = import_queue + app.config['import_workers'] = workers # Try to create database tables if they don't exist with app.app_context(): diff --git a/musicround/helpers/import_queue.py b/musicround/helpers/import_queue.py new file mode 100644 index 0000000..daa9807 --- /dev/null +++ b/musicround/helpers/import_queue.py @@ -0,0 +1,98 @@ +"""Import queue and worker implementation for asynchronous playlist imports.""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass, field +from queue import PriorityQueue, Empty +from typing import Optional +from flask import current_app +from flask_login import login_user, logout_user + +from musicround.models import User, db +from musicround.helpers.import_helper import ImportHelper + + +@dataclass(order=True) +class ImportJob: + """Represents a single import job.""" + + priority: int + service_name: str = field(compare=False) + item_type: str = field(compare=False) + item_id: str = field(compare=False) + user_id: int = field(compare=False) + + +class ImportQueue: + """Priority queue for import jobs.""" + + def __init__(self) -> None: + self._queue: PriorityQueue[tuple[int, int, ImportJob]] = PriorityQueue() + self._counter = 0 + self._lock = threading.Lock() + + def add_job(self, job: ImportJob) -> None: + """Add a job to the queue.""" + with self._lock: + self._counter += 1 + self._queue.put((job.priority, self._counter, job)) + + def get_job(self, timeout: Optional[float] = None) -> Optional[ImportJob]: + """Retrieve the next job from the queue.""" + try: + _, _, job = self._queue.get(timeout=timeout) + return job + except Empty: + return None + + def task_done(self) -> None: + """Signal that a previously fetched job is complete.""" + self._queue.task_done() + + +class ImportWorker(threading.Thread): + """Background worker thread for processing import jobs.""" + + def __init__(self, app, queue: ImportQueue) -> None: + super().__init__(daemon=True) + self.app = app + self.queue = queue + self._stop_event = threading.Event() + + def stop(self) -> None: + """Stop the worker loop.""" + self._stop_event.set() + + def run(self) -> None: + with self.app.app_context(): + while not self._stop_event.is_set(): + job = self.queue.get_job(timeout=1.0) + if job is None: + continue + self._process_job(job) + self.queue.task_done() + + def _process_job(self, job: ImportJob) -> None: + user = User.query.get(job.user_id) + if not user: + current_app.logger.error("Import job for unknown user %s", job.user_id) + return + with self.app.test_request_context(): + login_user(user) + try: + current_app.logger.info( + "Processing import job: service=%s type=%s id=%s user=%s priority=%s", + job.service_name, + job.item_type, + job.item_id, + job.user_id, + job.priority, + ) + ImportHelper.import_item(job.service_name, job.item_type, job.item_id) + db.session.commit() + except Exception as exc: # pylint: disable=broad-except + current_app.logger.error("Import job failed: %s", exc, exc_info=True) + db.session.rollback() + finally: + logout_user() diff --git a/musicround/routes/import_songs.py b/musicround/routes/import_songs.py index 40e72ba..ffbb213 100644 --- a/musicround/routes/import_songs.py +++ b/musicround/routes/import_songs.py @@ -85,22 +85,23 @@ def import_playlist(): flash("No playlist ID provided for import.", "danger") return redirect(request.referrer or url_for('core.search')) - result = ImportHelper.import_item(service_name='spotify', item_type='playlist', item_id=playlist_id, oauth_spotify=oauth.spotify) - - imported_count = result.get('imported_count', 0) - skipped_count = result.get('skipped_count', 0) - error_count = result.get('error_count', 0) - errors = result.get("errors", []) + priority = int(request.form.get('priority', 10)) + queue = current_app.config.get('import_queue') + if not queue: + flash("Import queue not initialized.", "danger") + return redirect(url_for('core.view_songs')) - if imported_count > 0: - flash(f'Successfully imported {imported_count} songs from playlist! ({skipped_count} skipped, {error_count} errors).', 'success') - elif skipped_count > 0 and error_count == 0: - flash(f'All {skipped_count} songs were already in the database.', 'info') - elif error_count > 0: - flash(f'Playlist import: {imported_count} new, {skipped_count} skipped, {error_count} errors. Errors: {", ".join(errors)}', 'warning') - else: - flash(f'Error importing playlist: {", ".join(errors) if errors else "No songs imported, playlist might be empty or an unknown issue occurred."}', 'danger') - + from musicround.helpers.import_queue import ImportJob + + job = ImportJob( + priority=priority, + service_name='spotify', + item_type='playlist', + item_id=playlist_id, + user_id=current_user.id, + ) + queue.add_job(job) + flash('Playlist import queued.', 'info') return redirect(url_for('core.view_songs')) return render_template('service_import.html', diff --git a/musicround/version.py b/musicround/version.py index 5592e3b..80207ae 100644 --- a/musicround/version.py +++ b/musicround/version.py @@ -5,10 +5,10 @@ This module provides version details that can be displayed in both CLI and UI. # Version information VERSION_INFO = { - "version": "1.9.0", - "release_name": "Documentation Dynamo", - "release_date": "2025-05-12", - "build_number": "20250512001" + "version": "1.10.0", + "release_name": "Import Infrastructure", + "release_date": "2025-05-27", + "build_number": "20250527001" } def get_version_str(include_build=False):