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 <noreply@anthropic.com>
This commit is contained in:
+220
-228
@@ -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("/<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:
|
||||
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("/<int:subscription_id>", 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("/<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:
|
||||
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("/<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:
|
||||
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("/<int:subscription_id>/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("/<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:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user