Standardize upload task signatures to accept file_id parameter

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/tasks/upload_to_email.py

@@ -15,6 +15,7 @@ from jinja2 import Environment, FileSystemLoader, select_autoescape
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 
 logger = logging.getLogger(__name__)
 
@@ -159,14 +160,32 @@ def _send_email_with_smtp(msg, filename, recipients):
         logger.error(error_msg)
         return {"status": "Failed", "reason": error_msg, "error": str(e)}
 
-@celery.task(base=BaseTaskWithRetry)
+@celery.task(base=BaseTaskWithRetry, bind=True)
-def upload_to_email(file_path: str, recipients=None, subject=None, message=None, template_name="default.html", include_metadata=True):
+def upload_to_email(self, file_path: str, file_id: int = None, recipients=None, subject=None, message=None, template_name="default.html", include_metadata=True):
     """
     Sends a file via email to the specified recipients.
     If recipients is None, uses the configured default email recipient.
+    
+    Args:
+        file_path: Path to the file to send
+        file_id: Optional file ID to associate with logs
+        recipients: Optional list of recipient email addresses
+        subject: Optional email subject
+        message: Optional custom message
+        template_name: Email template to use
+        include_metadata: Whether to include metadata in the email
     """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting email send: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_email", "in_progress", f"Sending via email: {os.path.basename(file_path)}", file_id=file_id
+    )
+    
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_email", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename
     filename = os.path.basename(file_path)
@@ -174,16 +193,19 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None,
     # Check if email settings are configured
     if not settings.email_host:
         error_msg = "Email host is not configured"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_email", "skipped", error_msg, file_id=file_id)
         return {"status": "Skipped", "reason": error_msg}
         
     # Log email configuration for debugging
-    logger.debug(f"Email config - Host: {settings.email_host}, Port: {settings.email_port}, " 
+    logger.debug(f"[{task_id}] Email config - Host: {settings.email_host}, Port: {settings.email_port}, " 
                 f"Username: {settings.email_username}, TLS: {settings.email_use_tls}")
 
     # Process recipients
     recipients, error = _prepare_recipients(recipients)
     if error:
+        logger.error(f"[{task_id}] {error}")
+        log_task_progress(task_id, "upload_to_email", "skipped", error, file_id=file_id)
         return {"status": "Skipped", "reason": error}
 
     # Use provided subject or create default
@@ -237,8 +259,13 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None,
         # Send the email through SMTP
         error_result = _send_email_with_smtp(msg, filename, recipients)
         if error_result:
+            logger.error(f"[{task_id}] Failed to send email: {error_result.get('reason')}")
+            log_task_progress(task_id, "upload_to_email", "failure", error_result.get('reason'), file_id=file_id)
             return error_result
 
+        logger.info(f"[{task_id}] Successfully sent {filename} via email to {len(recipients)} recipients")
+        log_task_progress(task_id, "upload_to_email", "success", f"Sent via email: {filename}", file_id=file_id)
+        
         return {
             "status": "Completed",
             "file": file_path,
@@ -250,5 +277,6 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None,
         
     except Exception as e:
         error_msg = f"Failed to send {filename} via email: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_email", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_ftp.py

@@ -5,16 +5,31 @@ import ftplib
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 import logging
 
 logger = logging.getLogger(__name__)
 
-@celery.task(base=BaseTaskWithRetry)
+@celery.task(base=BaseTaskWithRetry, bind=True)
-def upload_to_ftp(file_path: str):
+def upload_to_ftp(self, file_path: str, file_id: int = None):
-    """Uploads a file to an FTP server in the configured folder."""
+    """
+    Uploads a file to an FTP server in the configured folder.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting FTP upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_ftp", "in_progress", f"Uploading to FTP: {os.path.basename(file_path)}", file_id=file_id
+    )
 
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename
     filename = os.path.basename(file_path)
@@ -22,7 +37,8 @@ def upload_to_ftp(file_path: str):
     # Check if FTP settings are configured
     if not settings.ftp_host:
         error_msg = "FTP host is not configured"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id)
         raise ValueError(error_msg)
 
     try:
@@ -123,7 +139,8 @@ def upload_to_ftp(file_path: str):
         # Close FTP connection
         ftp.quit()
         
-        logger.info(f"Successfully uploaded {filename} to FTP server at {settings.ftp_host}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to FTP server at {settings.ftp_host}")
+        log_task_progress(task_id, "upload_to_ftp", "success", f"Uploaded to FTP: {filename}", file_id=file_id)
         return {
             "status": "Completed", 
             "file": file_path, 
@@ -134,5 +151,6 @@ def upload_to_ftp(file_path: str):
     
     except Exception as e:
         error_msg = f"Failed to upload {filename} to FTP server: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_google_drive.py

@@ -16,6 +16,7 @@ from google.auth.exceptions import RefreshError
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 
 logger = logging.getLogger(__name__)
 
@@ -142,12 +143,27 @@ def truncate_property_value(key, value, max_bytes=100):
         
     return str_value
 
-@celery.task(base=BaseTaskWithRetry)
+@celery.task(base=BaseTaskWithRetry, bind=True)
-def upload_to_google_drive(file_path: str, include_metadata=True):
+def upload_to_google_drive(self, file_path: str, file_id: int = None, include_metadata=True):
-    """Uploads a file to Google Drive in the configured folder with optional metadata."""
+    """
+    Uploads a file to Google Drive in the configured folder with optional metadata.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+        include_metadata: Whether to include metadata in the upload
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting Google Drive upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_google_drive", "in_progress", f"Uploading to Google Drive: {os.path.basename(file_path)}", file_id=file_id
+    )
 
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename from path
     filename = os.path.basename(file_path)
@@ -161,7 +177,10 @@ def upload_to_google_drive(file_path: str, include_metadata=True):
         # Get Google Drive service
         service = get_google_drive_service()
         if not service:
-            raise Exception("Failed to initialize Google Drive service")
+            error_msg = "Failed to initialize Google Drive service"
+            logger.error(f"[{task_id}] {error_msg}")
+            log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id)
+            raise Exception(error_msg)
 
         # Prepare the file metadata
         file_metadata = {
@@ -221,16 +240,17 @@ def upload_to_google_drive(file_path: str, include_metadata=True):
         ).execute()
         
         # Log success details
-        file_id = file.get('id')
+        file_id_gd = file.get('id')
         web_view_link = file.get('webViewLink')
         
-        logger.info(f"Successfully uploaded {filename} to Google Drive with ID: {file_id}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to Google Drive with ID: {file_id_gd}")
-        logger.info(f"File accessible at: {web_view_link}")
+        logger.info(f"[{task_id}] File accessible at: {web_view_link}")
+        log_task_progress(task_id, "upload_to_google_drive", "success", f"Uploaded to Google Drive: {filename}", file_id=file_id)
         
         result = {
             "status": "Completed", 
             "file_path": file_path,
-            "google_drive_file_id": file_id,
+            "google_drive_file_id": file_id_gd,
             "google_drive_web_link": web_view_link
         }
         
@@ -242,5 +262,6 @@ def upload_to_google_drive(file_path: str, include_metadata=True):
 
     except Exception as e:
         error_msg = f"Failed to upload {filename} to Google Drive: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_sftp.py

@@ -8,21 +8,35 @@ from app.config import settings
 from app.celery_app import celery
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path
+from app.utils import log_task_progress
 
 logger = logging.getLogger(__name__)
 
-@celery.task(base=BaseTaskWithRetry)
+@celery.task(base=BaseTaskWithRetry, bind=True)
-def upload_to_sftp(file_path: str):
+def upload_to_sftp(self, file_path: str, file_id: int = None):
     """
     Upload a file to an SFTP server.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
     """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting SFTP upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_sftp", "in_progress", f"Uploading to SFTP: {os.path.basename(file_path)}", file_id=file_id
+    )
+    
     if not os.path.exists(file_path):
         error_msg = f"File not found: {file_path}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_sftp", "failure", error_msg, file_id=file_id)
         raise FileNotFoundError(error_msg)
     
     if not (settings.sftp_host and settings.sftp_port and settings.sftp_username):
-        logger.info("SFTP upload skipped: Missing configuration")
+        error_msg = "SFTP upload skipped: Missing configuration"
+        logger.info(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_sftp", "skipped", error_msg, file_id=file_id)
         return {"status": "Skipped", "reason": "SFTP settings not configured"}
     
     filename = os.path.basename(file_path)
@@ -102,9 +116,10 @@ def upload_to_sftp(file_path: str):
                 logger.warning(f"Failed to create directory structure {remote_dir}: {str(e)}")
         
         # Upload the file
-        logger.info(f"Uploading {filename} to SFTP at {remote_path}")
+        logger.info(f"[{task_id}] Uploading {filename} to SFTP at {remote_path}")
         sftp.put(file_path, remote_path)
-        logger.info(f"Successfully uploaded {filename} to SFTP at {remote_path}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to SFTP at {remote_path}")
+        log_task_progress(task_id, "upload_to_sftp", "success", f"Uploaded to SFTP: {filename}", file_id=file_id)
         
         # Close connections
         sftp.close()
@@ -126,5 +141,6 @@ def upload_to_sftp(file_path: str):
             pass
             
         error_msg = f"Failed to upload {filename} to SFTP server: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_sftp", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_webdav.py

@@ -6,16 +6,31 @@ from urllib.parse import urljoin
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 import logging
 
 logger = logging.getLogger(__name__)
 
-@celery.task(base=BaseTaskWithRetry)
+@celery.task(base=BaseTaskWithRetry, bind=True)
-def upload_to_webdav(file_path: str):
+def upload_to_webdav(self, file_path: str, file_id: int = None):
-    """Uploads a file to a WebDAV server in the configured folder."""
+    """
+    Uploads a file to a WebDAV server in the configured folder.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting WebDAV upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_webdav", "in_progress", f"Uploading to WebDAV: {os.path.basename(file_path)}", file_id=file_id
+    )
 
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename
     filename = os.path.basename(file_path)
@@ -23,7 +38,8 @@ def upload_to_webdav(file_path: str):
     # Check if WebDAV settings are configured
     if not settings.webdav_url:
         error_msg = "WebDAV URL is not configured"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
         raise ValueError(error_msg)
 
     # Construct the full upload URL
@@ -53,14 +69,17 @@ def upload_to_webdav(file_path: str):
 
         # Check if upload was successful
         if response.status_code in (200, 201, 204):
-            logger.info(f"Successfully uploaded {filename} to WebDAV at {webdav_url}.")
+            logger.info(f"[{task_id}] Successfully uploaded {filename} to WebDAV at {webdav_url}.")
+            log_task_progress(task_id, "upload_to_webdav", "success", f"Uploaded to WebDAV: {filename}", file_id=file_id)
             return {"status": "Completed", "file": file_path, "url": webdav_url}
         else:
             error_msg = f"Failed to upload {filename} to WebDAV: {response.status_code} - {response.text}"
-            logger.error(error_msg)
+            logger.error(f"[{task_id}] {error_msg}")
+            log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
             raise Exception(error_msg)
     
     except Exception as e:
         error_msg = f"Error uploading {filename} to WebDAV: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)