bvandeusen
080a1254c7
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 <noreply@anthropic.com>
342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""Subscription management API endpoints."""
|
|
|
|
from quart import Blueprint, request, jsonify, current_app
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.subscription import Subscription
|
|
from app.models.source import Source
|
|
from app.models.download import Download, DownloadStatus
|
|
from app.tasks.downloads import download_source
|
|
from app.api.settings import get_setting_value
|
|
|
|
bp = Blueprint("subscriptions", __name__)
|
|
|
|
|
|
@bp.route("", methods=["GET"])
|
|
async def list_subscriptions():
|
|
"""List all subscriptions with optional filtering and pagination."""
|
|
# Query parameters
|
|
enabled = request.args.get("enabled")
|
|
search = request.args.get("search")
|
|
page = int(request.args.get("page", 1))
|
|
per_page = min(int(request.args.get("per_page", 50)), 100)
|
|
|
|
async with current_app.db_engine.connect() as conn:
|
|
async with AsyncSession(bind=conn) as session:
|
|
# 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}%"))
|
|
|
|
# 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()
|
|
|
|
# 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()
|
|
|
|
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"])
|
|
async def create_subscription():
|
|
"""Create a new subscription with optional sources."""
|
|
data = await request.get_json()
|
|
|
|
# Validate required fields
|
|
if not data.get("name"):
|
|
return jsonify({"error": "Name is required"}), 400
|
|
|
|
async with current_app.db_engine.connect() as conn:
|
|
async with AsyncSession(bind=conn) as session:
|
|
# 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", {}),
|
|
)
|
|
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("/<int:subscription_id>", 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:
|
|
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()
|
|
|
|
if not subscription:
|
|
return jsonify({"error": "Subscription not found"}), 404
|
|
|
|
return jsonify(subscription.to_dict())
|
|
|
|
|
|
@bp.route("/<int:subscription_id>", methods=["PATCH"])
|
|
async def update_subscription(subscription_id: int):
|
|
"""Update a subscription."""
|
|
data = await request.get_json()
|
|
|
|
async with current_app.db_engine.connect() as 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()
|
|
|
|
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"])
|
|
)
|
|
if existing.scalar():
|
|
return jsonify({"error": "Subscription with this name already exists"}), 409
|
|
|
|
# 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("/<int:subscription_id>", 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:
|
|
async with AsyncSession(bind=conn) as session:
|
|
result = await session.execute(
|
|
select(Subscription).where(Subscription.id == subscription_id)
|
|
)
|
|
subscription = result.scalar()
|
|
|
|
if not subscription:
|
|
return jsonify({"error": "Subscription not found"}), 404
|
|
|
|
name = subscription.name
|
|
await session.delete(subscription)
|
|
await session.commit()
|
|
|
|
current_app.logger.info(f"Deleted subscription: {name}")
|
|
return "", 204
|
|
|
|
|
|
@bp.route("/<int:subscription_id>/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:
|
|
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()
|
|
|
|
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]
|
|
})
|
|
|
|
|
|
@bp.route("/<int:subscription_id>/sources", methods=["POST"])
|
|
async def add_source_to_subscription(subscription_id: int):
|
|
"""Add a new source to a subscription."""
|
|
data = await request.get_json()
|
|
|
|
# Validate required fields
|
|
required = ["platform", "url"]
|
|
missing = [f for f in required if not data.get(f)]
|
|
if missing:
|
|
return jsonify({"error": f"Missing required fields: {', '.join(missing)}"}), 400
|
|
|
|
# Validate platform
|
|
valid_platforms = ["patreon", "subscribestar", "hentaifoundry", "discord", "pixiv", "deviantart"]
|
|
if data["platform"] not in valid_platforms:
|
|
return jsonify({"error": f"Invalid platform. Must be one of: {', '.join(valid_platforms)}"}), 400
|
|
|
|
async with current_app.db_engine.connect() as conn:
|
|
async with AsyncSession(bind=conn) as session:
|
|
# 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"]
|
|
)
|
|
)
|
|
if existing.scalar():
|
|
return jsonify({"error": "Source with this platform and URL already exists"}), 409
|
|
|
|
# Store name before commit (attributes expire after commit)
|
|
subscription_name = subscription.name
|
|
|
|
# Get default check interval from database settings
|
|
default_interval = await get_setting_value(
|
|
session, "download.schedule_interval", 28800
|
|
)
|
|
|
|
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("/<int:subscription_id>/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:
|
|
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()
|
|
|
|
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])
|
|
)
|
|
)
|
|
active_source_ids = set(active_result.scalars().all())
|
|
|
|
# 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
|
|
|
|
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
|