bvandeusen/GallerySubscriber
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

correcting worker db connection issue

Browse Source
This commit is contained in:
Bryan Van Deusen
2026-01-27 18:05:54 -05:00
parent 13b4da5ca0
commit c7a259a63b
1 changed files with 123 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/tasks/downloads.py
+91 -66
View File
@@ -44,6 +44,9 @@ async def _cleanup_engine(engine):
async def _download_source_async(source_id: int, download_id: int = None) -> dict: async def _download_source_async(source_id: int, download_id: int = None) -> dict:
"""Async implementation of source download. """Async implementation of source download.
Uses separate DB sessions for setup and result writing to avoid
connection timeouts during long-running gallery-dl downloads.
Args: Args:
source_id: ID of the source to download source_id: ID of the source to download
download_id: Optional ID of a pre-created Download record (QUEUED state) download_id: Optional ID of a pre-created Download record (QUEUED state)
@@ -51,6 +54,7 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
Returns: Returns:
Dict with download results Dict with download results
""" """
# --- Phase 1: Setup (read source info, mark as RUNNING, get credentials) ---
session_factory, engine = get_async_session() session_factory, engine = get_async_session()
try: try:
@@ -69,7 +73,6 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
if not source.enabled: if not source.enabled:
logger.info(f"Source {source.subscription.name}/{source.platform} is disabled, skipping") logger.info(f"Source {source.subscription.name}/{source.platform} is disabled, skipping")
# If there's a queued download record, mark it as skipped
if download_id: if download_id:
dl_result = await session.execute( dl_result = await session.execute(
select(Download).where(Download.id == download_id) select(Download).where(Download.id == download_id)
@@ -80,8 +83,12 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
await session.commit() await session.commit()
return {"source_id": source_id, "status": "skipped", "reason": "disabled"} return {"source_id": source_id, "status": "skipped", "reason": "disabled"}
# Capture values we need after the session closes
subscription_name = source.subscription.name subscription_name = source.subscription.name
logger.info(f"Starting download for {subscription_name}/{source.platform}") source_url = source.url
source_platform = source.platform
source_metadata = dict(source.metadata_ or {})
logger.info(f"Starting download for {subscription_name}/{source_platform}")
# Use existing download record or create a new one # Use existing download record or create a new one
if download_id: if download_id:
@@ -98,8 +105,8 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
if not download_id or download is None: if not download_id or download is None:
download = Download( download = Download(
source_id=source.id, source_id=source_id,
url=source.url, url=source_url,
status=DownloadStatus.RUNNING, status=DownloadStatus.RUNNING,
started_at=utcnow(), started_at=utcnow(),
) )
@@ -107,51 +114,101 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
await session.commit() await session.commit()
await session.refresh(download) await session.refresh(download)
actual_download_id = download.id
# Broadcast download started event # Broadcast download started event
publish_event("download.started", { publish_event("download.started", {
"download_id": download.id, "download_id": actual_download_id,
"source_id": source.id, "source_id": source_id,
"subscription_name": subscription_name, "subscription_name": subscription_name,
"platform": source.platform, "platform": source_platform,
"url": source.url, "url": source_url,
}) })
try: # Get credentials before closing session
# Get credentials
cred_manager = CredentialManager(session) cred_manager = CredentialManager(session)
cookies_path = None cookies_path = None
auth_token = None auth_token = None
# Discord uses token auth, others use cookies if source_platform == "discord":
if source.platform == "discord": auth_token = await cred_manager.get_token(source_platform)
auth_token = await cred_manager.get_token(source.platform)
if not auth_token: if not auth_token:
logger.warning(f"No token for {source.platform}, download may fail") logger.warning(f"No token for {source_platform}, download may fail")
else: else:
cookies_path = await cred_manager.get_cookies_path(source.platform) cookies_path = await cred_manager.get_cookies_path(source_platform)
if not cookies_path and source.platform in ["patreon", "subscribestar"]: if not cookies_path and source_platform in ["patreon", "subscribestar"]:
logger.warning(f"No credentials for {source.platform}, download may fail") logger.warning(f"No credentials for {source_platform}, download may fail")
# Build source config from metadata # Session is now closed - DB connection released
source_config = SourceConfig.from_dict(source.metadata_ or {}) finally:
await _cleanup_engine(engine)
# Execute download # --- Phase 2: Execute download (no DB connection held) ---
source_config = SourceConfig.from_dict(source_metadata)
dl_result = None
dl_error = None
try:
gdl_service = GalleryDLService() gdl_service = GalleryDLService()
dl_result = await gdl_service.download( dl_result = await gdl_service.download(
url=source.url, url=source_url,
subscription_name=subscription_name, subscription_name=subscription_name,
platform=source.platform, platform=source_platform,
source_config=source_config, source_config=source_config,
cookies_path=cookies_path, cookies_path=cookies_path,
auth_token=auth_token, auth_token=auth_token,
) )
except Exception as e:
logger.exception(f"Unexpected error downloading {subscription_name}/{source_platform}: {e}")
dl_error = e
# Update download record # --- Phase 3: Write results (fresh DB session) ---
session_factory2, engine2 = get_async_session()
try:
async with session_factory2() as session:
# Re-fetch download and source with fresh session
download = (await session.execute(
select(Download).where(Download.id == actual_download_id)
)).scalar()
source = (await session.execute(
select(Source).where(Source.id == source_id)
)).scalar()
if not download or not source:
logger.error(f"Could not re-fetch download {actual_download_id} or source {source_id} for result update")
return {"source_id": source_id, "status": "error", "error": "Lost DB records"}
if dl_error:
download.status = DownloadStatus.FAILED
download.error_type = DBErrorType.UNKNOWN_ERROR
download.error_message = str(dl_error)
download.completed_at = utcnow()
source.error_count = (source.error_count or 0) + 1
source.last_check = utcnow()
await session.commit()
publish_event("download.failed", {
"download_id": actual_download_id,
"source_id": source_id,
"subscription_name": subscription_name,
"platform": source_platform,
"error_type": "unknown_error",
"error_message": str(dl_error),
})
return {
"source_id": source_id,
"download_id": actual_download_id,
"status": "error",
"error": str(dl_error),
}
# Update download record with results
download.completed_at = utcnow() download.completed_at = utcnow()
download.file_count = dl_result.files_downloaded download.file_count = dl_result.files_downloaded
download.metadata_ = { download.metadata_ = {
"stdout": dl_result.stdout[:10000] if dl_result.stdout else None, # Truncate "stdout": dl_result.stdout[:10000] if dl_result.stdout else None,
"stderr": dl_result.stderr[:5000] if dl_result.stderr else None, "stderr": dl_result.stderr[:5000] if dl_result.stderr else None,
"duration_seconds": dl_result.duration_seconds, "duration_seconds": dl_result.duration_seconds,
} }
@@ -160,14 +217,13 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
download.status = DownloadStatus.COMPLETED download.status = DownloadStatus.COMPLETED
source.last_success = utcnow() source.last_success = utcnow()
source.error_count = 0 source.error_count = 0
logger.info(f"Download completed for {subscription_name}/{source.platform}: {dl_result.files_downloaded} files") logger.info(f"Download completed for {subscription_name}/{source_platform}: {dl_result.files_downloaded} files")
# Broadcast download completed event
publish_event("download.completed", { publish_event("download.completed", {
"download_id": download.id, "download_id": actual_download_id,
"source_id": source.id, "source_id": source_id,
"subscription_name": subscription_name, "subscription_name": subscription_name,
"platform": source.platform, "platform": source_platform,
"file_count": dl_result.files_downloaded, "file_count": dl_result.files_downloaded,
"duration_seconds": dl_result.duration_seconds, "duration_seconds": dl_result.duration_seconds,
}) })
@@ -176,62 +232,31 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
download.error_type = dl_result.error_type.value if dl_result.error_type else None download.error_type = dl_result.error_type.value if dl_result.error_type else None
download.error_message = dl_result.error_message download.error_message = dl_result.error_message
source.error_count = (source.error_count or 0) + 1 source.error_count = (source.error_count or 0) + 1
logger.warning(f"Download failed for {subscription_name}/{source.platform}: {dl_result.error_message}") logger.warning(f"Download failed for {subscription_name}/{source_platform}: {dl_result.error_message}")
# Broadcast download failed event
publish_event("download.failed", { publish_event("download.failed", {
"download_id": download.id, "download_id": actual_download_id,
"source_id": source.id, "source_id": source_id,
"subscription_name": subscription_name, "subscription_name": subscription_name,
"platform": source.platform, "platform": source_platform,
"error_type": dl_result.error_type.value if dl_result.error_type else None, "error_type": dl_result.error_type.value if dl_result.error_type else None,
"error_message": dl_result.error_message, "error_message": dl_result.error_message,
}) })
# Update source last_check
source.last_check = utcnow() source.last_check = utcnow()
await session.commit() await session.commit()
return { return {
"source_id": source_id, "source_id": source_id,
"download_id": download.id, "download_id": actual_download_id,
"status": "completed" if dl_result.success else "failed", "status": "completed" if dl_result.success else "failed",
"files_downloaded": dl_result.files_downloaded, "files_downloaded": dl_result.files_downloaded,
"error_type": dl_result.error_type.value if dl_result.error_type else None, "error_type": dl_result.error_type.value if dl_result.error_type else None,
"error_message": dl_result.error_message, "error_message": dl_result.error_message,
"duration_seconds": dl_result.duration_seconds, "duration_seconds": dl_result.duration_seconds,
} }
except Exception as e:
logger.exception(f"Unexpected error downloading {subscription_name}/{source.platform}: {e}")
download.status = DownloadStatus.FAILED
download.error_type = DBErrorType.UNKNOWN_ERROR
download.error_message = str(e)
download.completed_at = utcnow()
source.error_count = (source.error_count or 0) + 1
source.last_check = utcnow()
await session.commit()
# Broadcast download failed event
publish_event("download.failed", {
"download_id": download.id,
"source_id": source.id,
"subscription_name": subscription_name,
"platform": source.platform,
"error_type": "unknown_error",
"error_message": str(e),
})
return {
"source_id": source_id,
"download_id": download.id,
"status": "error",
"error": str(e),
}
finally: finally:
await _cleanup_engine(engine) await _cleanup_engine(engine2)
async def _scheduled_check_async() -> dict: async def _scheduled_check_async() -> dict: