From 080a1254c7b0c15718718706296efb812606fef7 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 18 Mar 2026 23:38:58 -0400 Subject: [PATCH] fix: wrap AsyncSession in context manager across all API blueprints Replace all bare `session = AsyncSession(bind=conn)` assignments with `async with AsyncSession(bind=conn) as session:` in downloads.py, credentials.py, settings.py, sources.py, and subscriptions.py to ensure sessions are properly closed on all code paths. Co-Authored-By: Claude Sonnet 4.6 --- backend/app/api/credentials.py | 164 ++++++----- backend/app/api/downloads.py | 442 +++++++++++++++--------------- backend/app/api/settings.py | 213 +++++++-------- backend/app/api/sources.py | 304 ++++++++++----------- backend/app/api/subscriptions.py | 448 +++++++++++++++---------------- 5 files changed, 771 insertions(+), 800 deletions(-) diff --git a/backend/app/api/credentials.py b/backend/app/api/credentials.py index 75136dd..d6895eb 100644 --- a/backend/app/api/credentials.py +++ b/backend/app/api/credentials.py @@ -19,18 +19,17 @@ VALID_PLATFORMS = ["patreon", "subscribestar", "hentaifoundry", "discord", "pixi async def list_credentials(): """List all credential status (without sensitive data).""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute(select(Credential)) + credentials = result.scalars().all() - result = await session.execute(select(Credential)) - credentials = result.scalars().all() + # Also return a simple map of platform -> has_credentials for UI convenience + platforms_with_creds = {c.platform: True for c in credentials} - # Also return a simple map of platform -> has_credentials for UI convenience - platforms_with_creds = {c.platform: True for c in credentials} - - return jsonify({ - "items": [c.to_dict(include_data=False) for c in credentials], - "platforms": platforms_with_creds, - }) + return jsonify({ + "items": [c.to_dict(include_data=False) for c in credentials], + "platforms": platforms_with_creds, + }) @bp.route("", methods=["POST"]) @@ -48,15 +47,15 @@ async def upload_credentials(): if extension_key: # Look up API key from database async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - result = await session.execute( - select(Setting).where(Setting.key == EXTENSION_API_KEY_SETTING) - ) - setting = result.scalar() - stored_key = setting.value if setting else None + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Setting).where(Setting.key == EXTENSION_API_KEY_SETTING) + ) + setting = result.scalar() + stored_key = setting.value if setting else None - if not stored_key or extension_key != stored_key: - return jsonify({"error": "Invalid extension API key"}), 401 + if not stored_key or extension_key != stored_key: + return jsonify({"error": "Invalid extension API key"}), 401 data = await request.get_json() @@ -81,43 +80,42 @@ async def upload_credentials(): return jsonify({"error": "Invalid credential type. Must be 'cookies' or 'token'"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - # Check for existing credential - result = await session.execute( - select(Credential).where(Credential.platform == platform) - ) - credential = result.scalar() - - # Encrypt the credential data - encrypted = encrypt_data(cred_data, settings.secret_key) - - if credential: - # Update existing - credential.credential_type = cred_type - credential.data = encrypted - credential.expires_at = data.get("expires_at") - credential.last_verified = None # Reset verification - else: - # Create new - credential = Credential( - platform=platform, - credential_type=cred_type, - data=encrypted, - expires_at=data.get("expires_at"), + async with AsyncSession(bind=conn) as session: + # Check for existing credential + result = await session.execute( + select(Credential).where(Credential.platform == platform) ) - session.add(credential) + credential = result.scalar() - await session.commit() - await session.refresh(credential) + # Encrypt the credential data + encrypted = encrypt_data(cred_data, settings.secret_key) - current_app.logger.info(f"Updated credentials for platform: {platform}") - return jsonify({ - "message": "Credentials updated", - "platform": platform, - "credential_type": cred_type, - "expires_at": credential.expires_at.isoformat() if credential.expires_at else None, - }) + if credential: + # Update existing + credential.credential_type = cred_type + credential.data = encrypted + credential.expires_at = data.get("expires_at") + credential.last_verified = None # Reset verification + else: + # Create new + credential = Credential( + platform=platform, + credential_type=cred_type, + data=encrypted, + expires_at=data.get("expires_at"), + ) + session.add(credential) + + await session.commit() + await session.refresh(credential) + + current_app.logger.info(f"Updated credentials for platform: {platform}") + return jsonify({ + "message": "Credentials updated", + "platform": platform, + "credential_type": cred_type, + "expires_at": credential.expires_at.isoformat() if credential.expires_at else None, + }) @bp.route("/", methods=["DELETE"]) @@ -127,21 +125,20 @@ async def delete_credentials(platform: str): return jsonify({"error": f"Invalid platform. Must be one of: {', '.join(VALID_PLATFORMS)}"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Credential).where(Credential.platform == platform) + ) + credential = result.scalar() - result = await session.execute( - select(Credential).where(Credential.platform == platform) - ) - credential = result.scalar() + if not credential: + return jsonify({"error": "Credentials not found for this platform"}), 404 - if not credential: - return jsonify({"error": "Credentials not found for this platform"}), 404 + await session.delete(credential) + await session.commit() - await session.delete(credential) - await session.commit() - - current_app.logger.info(f"Deleted credentials for platform: {platform}") - return "", 204 + current_app.logger.info(f"Deleted credentials for platform: {platform}") + return "", 204 @bp.route("/verify", methods=["POST"]) @@ -160,28 +157,27 @@ async def verify_credentials(): return jsonify({"error": f"Invalid platform. Must be one of: {', '.join(VALID_PLATFORMS)}"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Credential).where(Credential.platform == platform) + ) + credential = result.scalar() - result = await session.execute( - select(Credential).where(Credential.platform == platform) - ) - credential = result.scalar() + if not credential: + return jsonify({ + "platform": platform, + "is_valid": False, + "error": "No credentials stored for this platform", + }) + + # TODO: Implement actual verification by making test request + # For now, just mark as verified + verified_at = utcnow() + credential.last_verified = verified_at + await session.commit() - if not credential: return jsonify({ "platform": platform, - "is_valid": False, - "error": "No credentials stored for this platform", + "is_valid": True, + "last_verified": verified_at.isoformat(), }) - - # TODO: Implement actual verification by making test request - # For now, just mark as verified - verified_at = utcnow() - credential.last_verified = verified_at - await session.commit() - - return jsonify({ - "platform": platform, - "is_valid": True, - "last_verified": verified_at.isoformat(), - }) diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py index 6cb980f..ca8fe20 100644 --- a/backend/app/api/downloads.py +++ b/backend/app/api/downloads.py @@ -28,124 +28,121 @@ async def list_downloads(): per_page = min(int(request.args.get("per_page", 50)), 100) async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Build query - eager load source and subscription for labeling + query = select(Download).options( + selectinload(Download.source).selectinload(Source.subscription) + ) + count_query = select(func.count()).select_from(Download) - # Build query - eager load source and subscription for labeling - query = select(Download).options( - selectinload(Download.source).selectinload(Source.subscription) - ) - count_query = select(func.count()).select_from(Download) + filters = [] + if source_id: + filters.append(Download.source_id == source_id) + if status: + filters.append(Download.status == status) + if from_date: + filters.append(Download.created_at >= datetime.fromisoformat(from_date)) + if to_date: + filters.append(Download.created_at <= datetime.fromisoformat(to_date)) + # When filtering failed downloads, optionally exclude ones superseded by a later success + if exclude_superseded and status == DownloadStatus.FAILED: + filters.append(Download.superseded == False) - filters = [] - if source_id: - filters.append(Download.source_id == source_id) - if status: - filters.append(Download.status == status) - if from_date: - filters.append(Download.created_at >= datetime.fromisoformat(from_date)) - if to_date: - filters.append(Download.created_at <= datetime.fromisoformat(to_date)) - # When filtering failed downloads, optionally exclude ones superseded by a later success - if exclude_superseded and status == DownloadStatus.FAILED: - filters.append(Download.superseded == False) + if filters: + query = query.where(and_(*filters)) + count_query = count_query.where(and_(*filters)) - if filters: - query = query.where(and_(*filters)) - count_query = count_query.where(and_(*filters)) + # Get total count + total_result = await session.execute(count_query) + total = total_result.scalar() - # Get total count - total_result = await session.execute(count_query) - total = total_result.scalar() + # Apply pagination and ordering + query = query.order_by(Download.created_at.desc()) + query = query.offset((page - 1) * per_page).limit(per_page) - # Apply pagination and ordering - query = query.order_by(Download.created_at.desc()) - query = query.offset((page - 1) * per_page).limit(per_page) + result = await session.execute(query) + downloads = result.scalars().all() - result = await session.execute(query) - downloads = result.scalars().all() + # Include source info with each download + items = [] + for d in downloads: + item = d.to_dict() + if d.source: + item["subscription_name"] = d.source.subscription.name if d.source.subscription else None + item["platform"] = d.source.platform + else: + item["subscription_name"] = None + item["platform"] = None + items.append(item) - # Include source info with each download - items = [] - for d in downloads: - item = d.to_dict() - if d.source: - item["subscription_name"] = d.source.subscription.name if d.source.subscription else None - item["platform"] = d.source.platform - else: - item["subscription_name"] = None - item["platform"] = None - items.append(item) - - return jsonify({ - "items": items, - "total": total, - "page": page, - "per_page": per_page, - "pages": (total + per_page - 1) // per_page, - }) + return jsonify({ + "items": items, + "total": total, + "page": page, + "per_page": per_page, + "pages": (total + per_page - 1) // per_page, + }) @bp.route("/", methods=["GET"]) async def get_download(download_id: int): """Get download details.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute(select(Download).where(Download.id == download_id)) + download = result.scalar() - result = await session.execute(select(Download).where(Download.id == download_id)) - download = result.scalar() + if not download: + return jsonify({"error": "Download not found"}), 404 - if not download: - return jsonify({"error": "Download not found"}), 404 + # Include source info if available + response = download.to_dict() + if download.source_id: + source_result = await session.execute( + select(Source) + .options(selectinload(Source.subscription)) + .where(Source.id == download.source_id) + ) + source = source_result.scalar() + if source: + response["source"] = { + "id": source.id, + "subscription_name": source.subscription.name if source.subscription else None, + "platform": source.platform, + } - # Include source info if available - response = download.to_dict() - if download.source_id: - source_result = await session.execute( - select(Source) - .options(selectinload(Source.subscription)) - .where(Source.id == download.source_id) - ) - source = source_result.scalar() - if source: - response["source"] = { - "id": source.id, - "subscription_name": source.subscription.name if source.subscription else None, - "platform": source.platform, - } - - return jsonify(response) + return jsonify(response) @bp.route("//retry", methods=["POST"]) async def retry_download(download_id: int): """Retry a failed download.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute(select(Download).where(Download.id == download_id)) + download = result.scalar() - result = await session.execute(select(Download).where(Download.id == download_id)) - download = result.scalar() + if not download: + return jsonify({"error": "Download not found"}), 404 - if not download: - return jsonify({"error": "Download not found"}), 404 + if download.status != DownloadStatus.FAILED: + return jsonify({"error": "Can only retry failed downloads"}), 400 - if download.status != DownloadStatus.FAILED: - return jsonify({"error": "Can only retry failed downloads"}), 400 + # Reset status to queued + download.status = DownloadStatus.QUEUED + download.error_type = None + download.error_message = None + await session.commit() - # Reset status to queued - download.status = DownloadStatus.QUEUED - download.error_type = None - download.error_message = None - await session.commit() + # Queue Celery task to retry the download + task = process_download.delay(download_id) - # Queue Celery task to retry the download - task = process_download.delay(download_id) - - current_app.logger.info(f"Queued retry for download: {download_id}") - return jsonify({ - "message": "Retry queued", - "download_id": download_id, - "task_id": task.id, - }), 202 + current_app.logger.info(f"Queued retry for download: {download_id}") + return jsonify({ + "message": "Retry queued", + "download_id": download_id, + "task_id": task.id, + }), 202 @bp.route("/recent-activity", methods=["GET"]) @@ -162,48 +159,47 @@ async def recent_activity(): cutoff = utcnow() - timedelta(days=days) async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - # Get failed downloads OR completed with files (noteworthy activity) - # Exclude failed downloads that have been superseded by a later success - # for the same source - those are stale failures the user doesn't need to see - query = select(Download).options( - selectinload(Download.source).selectinload(Source.subscription) - ).where( - and_( - Download.created_at >= cutoff, - or_( - and_( - Download.status == DownloadStatus.FAILED, - Download.superseded == False, - ), - and_( - Download.status == DownloadStatus.COMPLETED, - Download.file_count > 0 + async with AsyncSession(bind=conn) as session: + # Get failed downloads OR completed with files (noteworthy activity) + # Exclude failed downloads that have been superseded by a later success + # for the same source - those are stale failures the user doesn't need to see + query = select(Download).options( + selectinload(Download.source).selectinload(Source.subscription) + ).where( + and_( + Download.created_at >= cutoff, + or_( + and_( + Download.status == DownloadStatus.FAILED, + Download.superseded == False, + ), + and_( + Download.status == DownloadStatus.COMPLETED, + Download.file_count > 0 + ) ) ) - ) - ).order_by(Download.created_at.desc()).limit(limit) + ).order_by(Download.created_at.desc()).limit(limit) - result = await session.execute(query) - downloads = result.scalars().all() + result = await session.execute(query) + downloads = result.scalars().all() - # Include source info with each download - items = [] - for d in downloads: - item = d.to_dict() - if d.source: - item["subscription_name"] = d.source.subscription.name if d.source.subscription else None - item["platform"] = d.source.platform - else: - item["subscription_name"] = None - item["platform"] = None - items.append(item) + # Include source info with each download + items = [] + for d in downloads: + item = d.to_dict() + if d.source: + item["subscription_name"] = d.source.subscription.name if d.source.subscription else None + item["platform"] = d.source.platform + else: + item["subscription_name"] = None + item["platform"] = None + items.append(item) - return jsonify({ - "items": items, - "count": len(downloads), - }) + return jsonify({ + "items": items, + "count": len(downloads), + }) @bp.route("/reset-orphaned", methods=["POST"]) @@ -286,54 +282,53 @@ async def get_stats(): period = request.args.get("period", "week") # day, week, month async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Total counts by status + status_query = select( + Download.status, + func.count(Download.id) + ).group_by(Download.status) - # Total counts by status - status_query = select( - Download.status, - func.count(Download.id) - ).group_by(Download.status) + status_result = await session.execute(status_query) + status_counts = {row[0]: row[1] for row in status_result} - status_result = await session.execute(status_query) - status_counts = {row[0]: row[1] for row in status_result} + # Counts by platform (through source) + platform_query = select( + Source.platform, + func.count(Download.id) + ).join(Source, Download.source_id == Source.id).group_by(Source.platform) - # Counts by platform (through source) - platform_query = select( - Source.platform, - func.count(Download.id) - ).join(Source, Download.source_id == Source.id).group_by(Source.platform) + platform_result = await session.execute(platform_query) + platform_counts = {row[0]: row[1] for row in platform_result} - platform_result = await session.execute(platform_query) - platform_counts = {row[0]: row[1] for row in platform_result} - - # Recent downloads count - recent_query = select(func.count()).select_from(Download).where( - Download.status == DownloadStatus.COMPLETED - ) - recent_result = await session.execute(recent_query) - recent_completed = recent_result.scalar() - - # Active failed: failures not superseded by a later success for the same source - active_failed_query = select(func.count()).select_from(Download).where( - and_( - Download.status == DownloadStatus.FAILED, - Download.superseded == False, + # Recent downloads count + recent_query = select(func.count()).select_from(Download).where( + Download.status == DownloadStatus.COMPLETED ) - ) - active_failed_result = await session.execute(active_failed_query) - active_failed = active_failed_result.scalar() + recent_result = await session.execute(recent_query) + recent_completed = recent_result.scalar() - return jsonify({ - "total": sum(status_counts.values()), - "by_status": status_counts, - "by_platform": platform_counts, - "completed": status_counts.get(DownloadStatus.COMPLETED, 0), - "failed": status_counts.get(DownloadStatus.FAILED, 0), - "active_failed": active_failed, - "pending": status_counts.get(DownloadStatus.PENDING, 0), - "queued": status_counts.get(DownloadStatus.QUEUED, 0), - "running": status_counts.get(DownloadStatus.RUNNING, 0), - }) + # Active failed: failures not superseded by a later success for the same source + active_failed_query = select(func.count()).select_from(Download).where( + and_( + Download.status == DownloadStatus.FAILED, + Download.superseded == False, + ) + ) + active_failed_result = await session.execute(active_failed_query) + active_failed = active_failed_result.scalar() + + return jsonify({ + "total": sum(status_counts.values()), + "by_status": status_counts, + "by_platform": platform_counts, + "completed": status_counts.get(DownloadStatus.COMPLETED, 0), + "failed": status_counts.get(DownloadStatus.FAILED, 0), + "active_failed": active_failed, + "pending": status_counts.get(DownloadStatus.PENDING, 0), + "queued": status_counts.get(DownloadStatus.QUEUED, 0), + "running": status_counts.get(DownloadStatus.RUNNING, 0), + }) @bp.route("/export-failed-logs", methods=["GET"]) @@ -356,69 +351,68 @@ async def export_failed_logs(): cutoff_date = utcnow() - timedelta(days=days) async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - # Build query for failed downloads (eager load source for platform) - query = select(Download).options(selectinload(Download.source)).where( - and_( - Download.status == DownloadStatus.FAILED, - Download.created_at >= cutoff_date, - ) - ) - - if error_type: - query = query.where(Download.error_type == error_type) - - # Order by most recent first - query = query.order_by(desc(Download.created_at)).limit(limit) - result = await session.execute(query) - failed_downloads = result.scalars().all() - - # Optionally include some successful NO_NEW_CONTENT for comparison - success_downloads = [] - if include_success: - success_query = select(Download).options(selectinload(Download.source)).where( + async with AsyncSession(bind=conn) as session: + # Build query for failed downloads (eager load source for platform) + query = select(Download).options(selectinload(Download.source)).where( and_( - Download.status == DownloadStatus.COMPLETED, + Download.status == DownloadStatus.FAILED, Download.created_at >= cutoff_date, - Download.file_count == 0, # No new content cases ) - ).order_by(desc(Download.created_at)).limit(limit // 4) - success_result = await session.execute(success_query) - success_downloads = success_result.scalars().all() + ) - # Format for export - exports = [] + if error_type: + query = query.where(Download.error_type == error_type) - for download in failed_downloads: - exports.append(_format_download_for_export(download, "failed")) + # Order by most recent first + query = query.order_by(desc(Download.created_at)).limit(limit) + result = await session.execute(query) + failed_downloads = result.scalars().all() - for download in success_downloads: - exports.append(_format_download_for_export(download, "success_no_new_content")) + # Optionally include some successful NO_NEW_CONTENT for comparison + success_downloads = [] + if include_success: + success_query = select(Download).options(selectinload(Download.source)).where( + and_( + Download.status == DownloadStatus.COMPLETED, + Download.created_at >= cutoff_date, + Download.file_count == 0, # No new content cases + ) + ).order_by(desc(Download.created_at)).limit(limit // 4) + success_result = await session.execute(success_query) + success_downloads = success_result.scalars().all() - # Build output document - output = { - "export_info": { - "exported_at": utcnow().isoformat(), - "total_records": len(exports), - "failed_count": len(failed_downloads), - "success_comparison_count": len(success_downloads), - "days_included": days, - "error_type_filter": error_type, - }, - "analysis_prompt": _generate_analysis_prompt(), - "downloads": exports, - } + # Format for export + exports = [] - # Return as downloadable JSON file - response = Response( - json.dumps(output, indent=2, default=str), - mimetype="application/json", - headers={ - "Content-Disposition": f"attachment; filename=failed_logs_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + for download in failed_downloads: + exports.append(_format_download_for_export(download, "failed")) + + for download in success_downloads: + exports.append(_format_download_for_export(download, "success_no_new_content")) + + # Build output document + output = { + "export_info": { + "exported_at": utcnow().isoformat(), + "total_records": len(exports), + "failed_count": len(failed_downloads), + "success_comparison_count": len(success_downloads), + "days_included": days, + "error_type_filter": error_type, + }, + "analysis_prompt": _generate_analysis_prompt(), + "downloads": exports, } - ) - return response + + # Return as downloadable JSON file + response = Response( + json.dumps(output, indent=2, default=str), + mimetype="application/json", + headers={ + "Content-Disposition": f"attachment; filename=failed_logs_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + } + ) + return response def _format_download_for_export(download: Download, classification: str) -> dict: diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 0a5ac4d..ffbebec 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -81,24 +81,23 @@ async def get_setting_value(session: AsyncSession, key: str, default=None): async def get_all_settings(): """Get all application settings.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute(select(Setting)) + settings = result.scalars().all() - result = await session.execute(select(Setting)) - settings = result.scalars().all() + # Build settings dict with defaults + settings_dict = dict(DEFAULT_SETTINGS) + for s in settings: + settings_dict[s.key] = s.value - # Build settings dict with defaults - settings_dict = dict(DEFAULT_SETTINGS) - for s in settings: - settings_dict[s.key] = s.value + # Ensure extension API key exists + api_key = await get_or_create_extension_api_key(session) + settings_dict[EXTENSION_API_KEY_SETTING] = api_key - # Ensure extension API key exists - api_key = await get_or_create_extension_api_key(session) - settings_dict[EXTENSION_API_KEY_SETTING] = api_key + # Get cached storage statistics (calculated by background task) + storage_stats = await get_cached_storage_stats(session) - # Get cached storage statistics (calculated by background task) - storage_stats = await get_cached_storage_stats(session) - - return jsonify({"settings": settings_dict, "storage_stats": storage_stats}) + return jsonify({"settings": settings_dict, "storage_stats": storage_stats}) @bp.route("", methods=["PATCH"]) @@ -110,68 +109,66 @@ async def update_settings(): return jsonify({"error": "No settings provided"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + for key, value in data.items(): + # Check if setting exists + result = await session.execute(select(Setting).where(Setting.key == key)) + setting = result.scalar() - for key, value in data.items(): - # Check if setting exists - result = await session.execute(select(Setting).where(Setting.key == key)) - setting = result.scalar() + if setting: + current_app.logger.info(f"Updating setting '{key}': {setting.value} -> {value} (type: {type(value).__name__})") + setting.value = value + else: + current_app.logger.info(f"Creating setting '{key}': {value} (type: {type(value).__name__})") + setting = Setting(key=key, value=value) + session.add(setting) - if setting: - current_app.logger.info(f"Updating setting '{key}': {setting.value} -> {value} (type: {type(value).__name__})") - setting.value = value - else: - current_app.logger.info(f"Creating setting '{key}': {value} (type: {type(value).__name__})") - setting = Setting(key=key, value=value) - session.add(setting) + await session.commit() - await session.commit() + # Return updated settings + result = await session.execute(select(Setting)) + settings = result.scalars().all() + settings_dict = dict(DEFAULT_SETTINGS) + for s in settings: + settings_dict[s.key] = s.value - # Return updated settings - result = await session.execute(select(Setting)) - settings = result.scalars().all() - settings_dict = dict(DEFAULT_SETTINGS) - for s in settings: - settings_dict[s.key] = s.value - - current_app.logger.info(f"Updated settings: {list(data.keys())}") - return jsonify({"settings": settings_dict}) + current_app.logger.info(f"Updated settings: {list(data.keys())}") + return jsonify({"settings": settings_dict}) @bp.route("/api-key", methods=["GET"]) async def get_extension_api_key(): """Get the extension API key.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - api_key = await get_or_create_extension_api_key(session) - return jsonify({"api_key": api_key}) + async with AsyncSession(bind=conn) as session: + api_key = await get_or_create_extension_api_key(session) + return jsonify({"api_key": api_key}) @bp.route("/api-key/regenerate", methods=["POST"]) async def regenerate_extension_api_key(): """Regenerate the extension API key.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Find existing key + result = await session.execute( + select(Setting).where(Setting.key == EXTENSION_API_KEY_SETTING) + ) + setting = result.scalar() - # Find existing key - result = await session.execute( - select(Setting).where(Setting.key == EXTENSION_API_KEY_SETTING) - ) - setting = result.scalar() + # Generate new key + new_key = generate_api_key() - # Generate new key - new_key = generate_api_key() + if setting: + setting.value = new_key + else: + setting = Setting(key=EXTENSION_API_KEY_SETTING, value=new_key) + session.add(setting) - if setting: - setting.value = new_key - else: - setting = Setting(key=EXTENSION_API_KEY_SETTING, value=new_key) - session.add(setting) + await session.commit() - await session.commit() - - current_app.logger.info("Extension API key regenerated") - return jsonify({"api_key": new_key, "message": "API key regenerated"}) + current_app.logger.info("Extension API key regenerated") + return jsonify({"api_key": new_key, "message": "API key regenerated"}) @bp.route("/gallery-dl", methods=["GET"]) @@ -246,28 +243,27 @@ async def get_logs(): limit = min(int(request.args.get("limit", 50)), 200) async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Get recent downloads with output logs + query = select(Download).order_by(desc(Download.created_at)).limit(limit) + result = await session.execute(query) + downloads = result.scalars().all() - # Get recent downloads with output logs - query = select(Download).order_by(desc(Download.created_at)).limit(limit) - result = await session.execute(query) - downloads = result.scalars().all() + logs = [] + for dl in downloads: + metadata = dl.metadata_ or {} + if metadata.get("stdout") or metadata.get("stderr") or dl.error_message: + logs.append({ + "id": dl.id, + "url": dl.url, + "status": dl.status, + "created_at": dl.created_at.isoformat() + "Z" if dl.created_at else None, + "stdout": metadata.get("stdout", ""), + "stderr": metadata.get("stderr", ""), + "error_message": dl.error_message, + }) - logs = [] - for dl in downloads: - metadata = dl.metadata_ or {} - if metadata.get("stdout") or metadata.get("stderr") or dl.error_message: - logs.append({ - "id": dl.id, - "url": dl.url, - "status": dl.status, - "created_at": dl.created_at.isoformat() + "Z" if dl.created_at else None, - "stdout": metadata.get("stdout", ""), - "stderr": metadata.get("stderr", ""), - "error_message": dl.error_message, - }) - - return jsonify({"logs": logs, "count": len(logs)}) + return jsonify({"logs": logs, "count": len(logs)}) @bp.route("/system-info", methods=["GET"]) @@ -333,39 +329,38 @@ async def debug_schedule_interval(): the worker would read. """ async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Query the raw setting from database + result = await session.execute( + select(Setting).where(Setting.key == "download.schedule_interval") + ) + setting = result.scalar() - # Query the raw setting from database - result = await session.execute( - select(Setting).where(Setting.key == "download.schedule_interval") - ) - setting = result.scalar() + if setting: + raw_value = setting.value + db_info = { + "found_in_database": True, + "raw_value": raw_value, + "raw_type": type(raw_value).__name__, + "as_int": int(raw_value) if raw_value is not None else None, + "updated_at": setting.updated_at.isoformat() if setting.updated_at else None, + } + else: + db_info = { + "found_in_database": False, + "raw_value": None, + "raw_type": None, + "as_int": None, + "updated_at": None, + } - if setting: - raw_value = setting.value - db_info = { - "found_in_database": True, - "raw_value": raw_value, - "raw_type": type(raw_value).__name__, - "as_int": int(raw_value) if raw_value is not None else None, - "updated_at": setting.updated_at.isoformat() if setting.updated_at else None, - } - else: - db_info = { - "found_in_database": False, - "raw_value": None, - "raw_type": None, - "as_int": None, - "updated_at": None, - } + # What the worker would use + fallback_value = DEFAULT_SETTINGS.get("download.schedule_interval", 3600) + worker_would_use = db_info["as_int"] if db_info["found_in_database"] else fallback_value - # What the worker would use - fallback_value = DEFAULT_SETTINGS.get("download.schedule_interval", 3600) - worker_would_use = db_info["as_int"] if db_info["found_in_database"] else fallback_value - - return jsonify({ - "database": db_info, - "default_settings_value": fallback_value, - "worker_would_use": worker_would_use, - "worker_would_use_hours": worker_would_use / 3600 if worker_would_use else None, - }) + return jsonify({ + "database": db_info, + "default_settings_value": fallback_value, + "worker_would_use": worker_would_use, + "worker_would_use_hours": worker_would_use / 3600 if worker_would_use else None, + }) diff --git a/backend/app/api/sources.py b/backend/app/api/sources.py index 3f6628b..18c47a0 100644 --- a/backend/app/api/sources.py +++ b/backend/app/api/sources.py @@ -26,44 +26,43 @@ async def list_sources(): per_page = min(int(request.args.get("per_page", 50)), 100) async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Build query with subscription eager loading + query = select(Source).options(selectinload(Source.subscription)) - # Build query with subscription eager loading - query = select(Source).options(selectinload(Source.subscription)) + if platform: + query = query.where(Source.platform == platform) + if subscription_id: + query = query.where(Source.subscription_id == int(subscription_id)) + if enabled is not None: + query = query.where(Source.enabled == (enabled.lower() == "true")) - if platform: - query = query.where(Source.platform == platform) - if subscription_id: - query = query.where(Source.subscription_id == int(subscription_id)) - if enabled is not None: - query = query.where(Source.enabled == (enabled.lower() == "true")) + # Get total count + count_query = select(func.count()).select_from(Source) + if platform: + count_query = count_query.where(Source.platform == platform) + if subscription_id: + count_query = count_query.where(Source.subscription_id == int(subscription_id)) + if enabled is not None: + count_query = count_query.where(Source.enabled == (enabled.lower() == "true")) - # Get total count - count_query = select(func.count()).select_from(Source) - if platform: - count_query = count_query.where(Source.platform == platform) - if subscription_id: - count_query = count_query.where(Source.subscription_id == int(subscription_id)) - if enabled is not None: - count_query = count_query.where(Source.enabled == (enabled.lower() == "true")) + total_result = await session.execute(count_query) + total = total_result.scalar() - total_result = await session.execute(count_query) - total = total_result.scalar() + # Apply pagination - order by subscription name, then platform + query = query.join(Subscription).order_by(Subscription.name, Source.platform) + query = query.offset((page - 1) * per_page).limit(per_page) - # Apply pagination - order by subscription name, then platform - query = query.join(Subscription).order_by(Subscription.name, Source.platform) - query = query.offset((page - 1) * per_page).limit(per_page) + result = await session.execute(query) + sources = result.scalars().all() - result = await session.execute(query) - sources = result.scalars().all() - - return jsonify({ - "items": [s.to_dict() for s in sources], - "total": total, - "page": page, - "per_page": per_page, - "pages": (total + per_page - 1) // per_page if total > 0 else 0, - }) + return jsonify({ + "items": [s.to_dict() for s in sources], + "total": total, + "page": page, + "per_page": per_page, + "pages": (total + per_page - 1) // per_page if total > 0 else 0, + }) @bp.route("", methods=["POST"]) @@ -83,65 +82,63 @@ async def create_source(): return jsonify({"error": f"Invalid platform. Must be one of: {', '.join(valid_platforms)}"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - # Check subscription exists - sub_result = await session.execute( - select(Subscription).where(Subscription.id == data["subscription_id"]) - ) - subscription = sub_result.scalar() - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 - - # Check for duplicate platform+url - existing = await session.execute( - select(Source).where( - Source.platform == data["platform"], - Source.url == data["url"] + async with AsyncSession(bind=conn) as session: + # Check subscription exists + sub_result = await session.execute( + select(Subscription).where(Subscription.id == data["subscription_id"]) ) - ) - if existing.scalar(): - return jsonify({"error": "Source with this platform and URL already exists"}), 409 + subscription = sub_result.scalar() + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - # Get default check interval from database settings - default_interval = await get_setting_value( - session, "download.schedule_interval", 28800 - ) + # Check for duplicate platform+url + existing = await session.execute( + select(Source).where( + Source.platform == data["platform"], + Source.url == data["url"] + ) + ) + if existing.scalar(): + return jsonify({"error": "Source with this platform and URL already exists"}), 409 - source = Source( - subscription_id=data["subscription_id"], - platform=data["platform"], - url=data["url"], - enabled=data.get("enabled", True), - check_interval=data.get("check_interval", default_interval), - metadata_=data.get("metadata", {}), - ) + # Get default check interval from database settings + default_interval = await get_setting_value( + session, "download.schedule_interval", 28800 + ) - session.add(source) - await session.commit() - await session.refresh(source) + source = Source( + subscription_id=data["subscription_id"], + platform=data["platform"], + url=data["url"], + enabled=data.get("enabled", True), + check_interval=data.get("check_interval", default_interval), + metadata_=data.get("metadata", {}), + ) - current_app.logger.info(f"Created source: {subscription.name}/{source.platform}") - return jsonify(source.to_dict()), 201 + session.add(source) + await session.commit() + await session.refresh(source) + + current_app.logger.info(f"Created source: {subscription.name}/{source.platform}") + return jsonify(source.to_dict()), 201 @bp.route("/", methods=["GET"]) async def get_source(source_id: int): """Get a single source by ID.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Source) + .options(selectinload(Source.subscription)) + .where(Source.id == source_id) + ) + source = result.scalar() - result = await session.execute( - select(Source) - .options(selectinload(Source.subscription)) - .where(Source.id == source_id) - ) - source = result.scalar() + if not source: + return jsonify({"error": "Source not found"}), 404 - if not source: - return jsonify({"error": "Source not found"}), 404 - - return jsonify(source.to_dict()) + return jsonify(source.to_dict()) @bp.route("/", methods=["PATCH"]) @@ -150,103 +147,100 @@ async def update_source(source_id: int): data = await request.get_json() async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Source) + .options(selectinload(Source.subscription)) + .where(Source.id == source_id) + ) + source = result.scalar() - result = await session.execute( - select(Source) - .options(selectinload(Source.subscription)) - .where(Source.id == source_id) - ) - source = result.scalar() + if not source: + return jsonify({"error": "Source not found"}), 404 - if not source: - return jsonify({"error": "Source not found"}), 404 + # Update allowed fields + allowed_fields = ["enabled", "check_interval", "metadata"] + for field in allowed_fields: + if field in data: + if field == "metadata": + source.metadata_ = data[field] + else: + setattr(source, field, data[field]) - # Update allowed fields - allowed_fields = ["enabled", "check_interval", "metadata"] - for field in allowed_fields: - if field in data: - if field == "metadata": - source.metadata_ = data[field] - else: - setattr(source, field, data[field]) + await session.commit() + await session.refresh(source) - await session.commit() - await session.refresh(source) - - current_app.logger.info(f"Updated source: {source.subscription.name}/{source.platform}") - return jsonify(source.to_dict()) + current_app.logger.info(f"Updated source: {source.subscription.name}/{source.platform}") + return jsonify(source.to_dict()) @bp.route("/", methods=["DELETE"]) async def delete_source(source_id: int): """Delete a source.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Source) + .options(selectinload(Source.subscription)) + .where(Source.id == source_id) + ) + source = result.scalar() - result = await session.execute( - select(Source) - .options(selectinload(Source.subscription)) - .where(Source.id == source_id) - ) - source = result.scalar() + if not source: + return jsonify({"error": "Source not found"}), 404 - if not source: - return jsonify({"error": "Source not found"}), 404 + info = f"{source.subscription.name}/{source.platform}" + await session.delete(source) + await session.commit() - info = f"{source.subscription.name}/{source.platform}" - await session.delete(source) - await session.commit() - - current_app.logger.info(f"Deleted source: {info}") - return "", 204 + current_app.logger.info(f"Deleted source: {info}") + return "", 204 @bp.route("//check", methods=["POST"]) async def trigger_check(source_id: int): """Trigger an immediate download check for a source.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - result = await session.execute( - select(Source) - .options(selectinload(Source.subscription)) - .where(Source.id == source_id) - ) - source = result.scalar() - - if not source: - return jsonify({"error": "Source not found"}), 404 - - # Check for existing queued or running download for this source - existing = await session.execute( - select(Download).where( - Download.source_id == source.id, - Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING]) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Source) + .options(selectinload(Source.subscription)) + .where(Source.id == source_id) ) - ) - if existing.scalar(): - return jsonify({"error": "A download is already queued or running for this source"}), 409 + source = result.scalar() - # Create download record in QUEUED state for visibility - download = Download( - source_id=source.id, - url=source.url, - status=DownloadStatus.QUEUED, - ) - session.add(download) - await session.commit() - await session.refresh(download) + if not source: + return jsonify({"error": "Source not found"}), 404 - # Queue Celery task for immediate download - task = download_source.delay(source_id, download_id=download.id) + # Check for existing queued or running download for this source + existing = await session.execute( + select(Download).where( + Download.source_id == source.id, + Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING]) + ) + ) + if existing.scalar(): + return jsonify({"error": "A download is already queued or running for this source"}), 409 - current_app.logger.info(f"Triggered check for source: {source.subscription.name}/{source.platform}") - return jsonify({ - "message": "Check queued", - "source_id": source_id, - "download_id": download.id, - "subscription_name": source.subscription.name, - "platform": source.platform, - "task_id": task.id, - }), 202 + # Create download record in QUEUED state for visibility + download = Download( + source_id=source.id, + url=source.url, + status=DownloadStatus.QUEUED, + ) + session.add(download) + await session.commit() + await session.refresh(download) + + # Queue Celery task for immediate download + task = download_source.delay(source_id, download_id=download.id) + + current_app.logger.info(f"Triggered check for source: {source.subscription.name}/{source.platform}") + return jsonify({ + "message": "Check queued", + "source_id": source_id, + "download_id": download.id, + "subscription_name": source.subscription.name, + "platform": source.platform, + "task_id": task.id, + }), 202 diff --git a/backend/app/api/subscriptions.py b/backend/app/api/subscriptions.py index 9fe6580..b1b449c 100644 --- a/backend/app/api/subscriptions.py +++ b/backend/app/api/subscriptions.py @@ -24,40 +24,39 @@ async def list_subscriptions(): per_page = min(int(request.args.get("per_page", 50)), 100) async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + # Build query with eager loading of sources + query = select(Subscription).options(selectinload(Subscription.sources)) - # Build query with eager loading of sources - query = select(Subscription).options(selectinload(Subscription.sources)) + if enabled is not None: + query = query.where(Subscription.enabled == (enabled.lower() == "true")) + if search: + query = query.where(Subscription.name.ilike(f"%{search}%")) - if enabled is not None: - query = query.where(Subscription.enabled == (enabled.lower() == "true")) - if search: - query = query.where(Subscription.name.ilike(f"%{search}%")) + # Get total count + count_query = select(func.count()).select_from(Subscription) + if enabled is not None: + count_query = count_query.where(Subscription.enabled == (enabled.lower() == "true")) + if search: + count_query = count_query.where(Subscription.name.ilike(f"%{search}%")) - # Get total count - count_query = select(func.count()).select_from(Subscription) - if enabled is not None: - count_query = count_query.where(Subscription.enabled == (enabled.lower() == "true")) - if search: - count_query = count_query.where(Subscription.name.ilike(f"%{search}%")) + total_result = await session.execute(count_query) + total = total_result.scalar() - total_result = await session.execute(count_query) - total = total_result.scalar() + # Apply pagination + query = query.order_by(Subscription.priority.desc(), Subscription.name) + query = query.offset((page - 1) * per_page).limit(per_page) - # Apply pagination - query = query.order_by(Subscription.priority.desc(), Subscription.name) - query = query.offset((page - 1) * per_page).limit(per_page) + result = await session.execute(query) + subscriptions = result.scalars().unique().all() - result = await session.execute(query) - subscriptions = result.scalars().unique().all() - - return jsonify({ - "items": [s.to_dict() for s in subscriptions], - "total": total, - "page": page, - "per_page": per_page, - "pages": (total + per_page - 1) // per_page if total > 0 else 0, - }) + return jsonify({ + "items": [s.to_dict() for s in subscriptions], + "total": total, + "page": page, + "per_page": per_page, + "pages": (total + per_page - 1) // per_page if total > 0 else 0, + }) @bp.route("", methods=["POST"]) @@ -70,67 +69,65 @@ async def create_subscription(): return jsonify({"error": "Name is required"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - # Check for duplicate name - existing = await session.execute( - select(Subscription).where(Subscription.name == data["name"]) - ) - if existing.scalar(): - return jsonify({"error": "Subscription with this name already exists"}), 409 - - subscription = Subscription( - name=data["name"], - enabled=data.get("enabled", True), - priority=data.get("priority", 0), - description=data.get("description"), - metadata_=data.get("metadata", {}), - ) - - # Get default check interval from database settings - default_interval = await get_setting_value( - session, "download.schedule_interval", 28800 - ) - - # Add sources if provided - sources_data = data.get("sources", []) - for source_data in sources_data: - if not source_data.get("platform") or not source_data.get("url"): - continue - source = Source( - platform=source_data["platform"], - url=source_data["url"], - enabled=source_data.get("enabled", True), - check_interval=source_data.get("check_interval", default_interval), - metadata_=source_data.get("metadata", {}), + async with AsyncSession(bind=conn) as session: + # Check for duplicate name + existing = await session.execute( + select(Subscription).where(Subscription.name == data["name"]) ) - subscription.sources.append(source) + if existing.scalar(): + return jsonify({"error": "Subscription with this name already exists"}), 409 - session.add(subscription) - await session.commit() - await session.refresh(subscription) + subscription = Subscription( + name=data["name"], + enabled=data.get("enabled", True), + priority=data.get("priority", 0), + description=data.get("description"), + metadata_=data.get("metadata", {}), + ) - current_app.logger.info(f"Created subscription: {subscription.name}") - return jsonify(subscription.to_dict()), 201 + # Get default check interval from database settings + default_interval = await get_setting_value( + session, "download.schedule_interval", 28800 + ) + + # Add sources if provided + sources_data = data.get("sources", []) + for source_data in sources_data: + if not source_data.get("platform") or not source_data.get("url"): + continue + source = Source( + platform=source_data["platform"], + url=source_data["url"], + enabled=source_data.get("enabled", True), + check_interval=source_data.get("check_interval", default_interval), + metadata_=source_data.get("metadata", {}), + ) + subscription.sources.append(source) + + session.add(subscription) + await session.commit() + await session.refresh(subscription) + + current_app.logger.info(f"Created subscription: {subscription.name}") + return jsonify(subscription.to_dict()), 201 @bp.route("/", methods=["GET"]) async def get_subscription(subscription_id: int): """Get a single subscription by ID with its sources.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Subscription) + .options(selectinload(Subscription.sources)) + .where(Subscription.id == subscription_id) + ) + subscription = result.scalar() - result = await session.execute( - select(Subscription) - .options(selectinload(Subscription.sources)) - .where(Subscription.id == subscription_id) - ) - subscription = result.scalar() + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 - - return jsonify(subscription.to_dict()) + return jsonify(subscription.to_dict()) @bp.route("/", methods=["PATCH"]) @@ -139,85 +136,82 @@ async def update_subscription(subscription_id: int): data = await request.get_json() async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - result = await session.execute( - select(Subscription) - .options(selectinload(Subscription.sources)) - .where(Subscription.id == subscription_id) - ) - subscription = result.scalar() - - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 - - # Check for duplicate name if changing - if "name" in data and data["name"] != subscription.name: - existing = await session.execute( - select(Subscription).where(Subscription.name == data["name"]) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Subscription) + .options(selectinload(Subscription.sources)) + .where(Subscription.id == subscription_id) ) - if existing.scalar(): - return jsonify({"error": "Subscription with this name already exists"}), 409 + subscription = result.scalar() - # Update allowed fields - allowed_fields = ["name", "enabled", "priority", "description", "metadata"] - for field in allowed_fields: - if field in data: - if field == "metadata": - subscription.metadata_ = data[field] - else: - setattr(subscription, field, data[field]) + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - await session.commit() - await session.refresh(subscription) + # Check for duplicate name if changing + if "name" in data and data["name"] != subscription.name: + existing = await session.execute( + select(Subscription).where(Subscription.name == data["name"]) + ) + if existing.scalar(): + return jsonify({"error": "Subscription with this name already exists"}), 409 - current_app.logger.info(f"Updated subscription: {subscription.name}") - return jsonify(subscription.to_dict()) + # Update allowed fields + allowed_fields = ["name", "enabled", "priority", "description", "metadata"] + for field in allowed_fields: + if field in data: + if field == "metadata": + subscription.metadata_ = data[field] + else: + setattr(subscription, field, data[field]) + + await session.commit() + await session.refresh(subscription) + + current_app.logger.info(f"Updated subscription: {subscription.name}") + return jsonify(subscription.to_dict()) @bp.route("/", methods=["DELETE"]) async def delete_subscription(subscription_id: int): """Delete a subscription and all its sources.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Subscription).where(Subscription.id == subscription_id) + ) + subscription = result.scalar() - result = await session.execute( - select(Subscription).where(Subscription.id == subscription_id) - ) - subscription = result.scalar() + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 + name = subscription.name + await session.delete(subscription) + await session.commit() - name = subscription.name - await session.delete(subscription) - await session.commit() - - current_app.logger.info(f"Deleted subscription: {name}") - return "", 204 + current_app.logger.info(f"Deleted subscription: {name}") + return "", 204 @bp.route("//sources", methods=["GET"]) async def list_subscription_sources(subscription_id: int): """List all sources for a subscription.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Subscription) + .options(selectinload(Subscription.sources)) + .where(Subscription.id == subscription_id) + ) + subscription = result.scalar() - result = await session.execute( - select(Subscription) - .options(selectinload(Subscription.sources)) - .where(Subscription.id == subscription_id) - ) - subscription = result.scalar() + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 - - return jsonify({ - "subscription_id": subscription_id, - "subscription_name": subscription.name, - "sources": [s.to_dict(include_subscription=False) for s in subscription.sources] - }) + return jsonify({ + "subscription_id": subscription_id, + "subscription_name": subscription.name, + "sources": [s.to_dict(include_subscription=False) for s in subscription.sources] + }) @bp.route("//sources", methods=["POST"]) @@ -237,113 +231,111 @@ async def add_source_to_subscription(subscription_id: int): return jsonify({"error": f"Invalid platform. Must be one of: {', '.join(valid_platforms)}"}), 400 async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - # Check subscription exists - result = await session.execute( - select(Subscription).where(Subscription.id == subscription_id) - ) - subscription = result.scalar() - - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 - - # Check for duplicate platform+url - existing = await session.execute( - select(Source).where( - Source.platform == data["platform"], - Source.url == data["url"] + async with AsyncSession(bind=conn) as session: + # Check subscription exists + result = await session.execute( + select(Subscription).where(Subscription.id == subscription_id) ) - ) - if existing.scalar(): - return jsonify({"error": "Source with this platform and URL already exists"}), 409 + subscription = result.scalar() - # Store name before commit (attributes expire after commit) - subscription_name = subscription.name + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - # Get default check interval from database settings - default_interval = await get_setting_value( - session, "download.schedule_interval", 28800 - ) + # Check for duplicate platform+url + existing = await session.execute( + select(Source).where( + Source.platform == data["platform"], + Source.url == data["url"] + ) + ) + if existing.scalar(): + return jsonify({"error": "Source with this platform and URL already exists"}), 409 - source = Source( - subscription_id=subscription_id, - platform=data["platform"], - url=data["url"], - enabled=data.get("enabled", True), - check_interval=data.get("check_interval", default_interval), - metadata_=data.get("metadata", {}), - ) + # Store name before commit (attributes expire after commit) + subscription_name = subscription.name - session.add(source) - await session.commit() - await session.refresh(source) + # Get default check interval from database settings + default_interval = await get_setting_value( + session, "download.schedule_interval", 28800 + ) - current_app.logger.info(f"Added source to {subscription_name}: {source.platform}") - # Use include_subscription=False to avoid lazy loading the relationship - result = source.to_dict(include_subscription=False) - result["subscription_name"] = subscription_name - return jsonify(result), 201 + source = Source( + subscription_id=subscription_id, + platform=data["platform"], + url=data["url"], + enabled=data.get("enabled", True), + check_interval=data.get("check_interval", default_interval), + metadata_=data.get("metadata", {}), + ) + + session.add(source) + await session.commit() + await session.refresh(source) + + current_app.logger.info(f"Added source to {subscription_name}: {source.platform}") + # Use include_subscription=False to avoid lazy loading the relationship + result = source.to_dict(include_subscription=False) + result["subscription_name"] = subscription_name + return jsonify(result), 201 @bp.route("//check", methods=["POST"]) async def trigger_subscription_check(subscription_id: int): """Trigger download check for all enabled sources in a subscription.""" async with current_app.db_engine.connect() as conn: - session = AsyncSession(bind=conn) - - result = await session.execute( - select(Subscription) - .options(selectinload(Subscription.sources)) - .where(Subscription.id == subscription_id) - ) - subscription = result.scalar() - - if not subscription: - return jsonify({"error": "Subscription not found"}), 404 - - enabled_sources = [s for s in subscription.sources if s.enabled] - - # Get source IDs that already have a queued or running download - active_result = await session.execute( - select(Download.source_id).where( - Download.source_id.in_([s.id for s in enabled_sources]), - Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING]) + async with AsyncSession(bind=conn) as session: + result = await session.execute( + select(Subscription) + .options(selectinload(Subscription.sources)) + .where(Subscription.id == subscription_id) ) - ) - active_source_ids = set(active_result.scalars().all()) + subscription = result.scalar() - # Create download records and queue Celery tasks for each eligible source - task_ids = [] - download_ids = [] - skipped = 0 - for source in enabled_sources: - if source.id in active_source_ids: - skipped += 1 - continue + if not subscription: + return jsonify({"error": "Subscription not found"}), 404 - download = Download( - source_id=source.id, - url=source.url, - status=DownloadStatus.QUEUED, + enabled_sources = [s for s in subscription.sources if s.enabled] + + # Get source IDs that already have a queued or running download + active_result = await session.execute( + select(Download.source_id).where( + Download.source_id.in_([s.id for s in enabled_sources]), + Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING]) + ) ) - session.add(download) - await session.commit() - await session.refresh(download) + active_source_ids = set(active_result.scalars().all()) - task = download_source.delay(source.id, download_id=download.id) - task_ids.append(task.id) - download_ids.append(download.id) + # Create download records and queue Celery tasks for each eligible source + task_ids = [] + download_ids = [] + skipped = 0 + for source in enabled_sources: + if source.id in active_source_ids: + skipped += 1 + continue - current_app.logger.info( - f"Triggered check for subscription: {subscription.name} " - f"({len(download_ids)} queued, {skipped} already active)" - ) - return jsonify({ - "message": "Checks queued", - "subscription_id": subscription_id, - "source_count": len(download_ids), - "skipped_active": skipped, - "task_ids": task_ids, - "download_ids": download_ids, - }), 202 + download = Download( + source_id=source.id, + url=source.url, + status=DownloadStatus.QUEUED, + ) + session.add(download) + await session.commit() + await session.refresh(download) + + task = download_source.delay(source.id, download_id=download.id) + task_ids.append(task.id) + download_ids.append(download.id) + + current_app.logger.info( + f"Triggered check for subscription: {subscription.name} " + f"({len(download_ids)} queued, {skipped} already active)" + ) + return jsonify({ + "message": "Checks queued", + "subscription_id": subscription_id, + "source_count": len(download_ids), + "skipped_active": skipped, + "task_ids": task_ids, + "download_ids": download_ids, + }), 202