feat: run background task to upload images of submission to s3

src/backend/app/db/db_models.py

@@ -551,3 +551,15 @@ class DbTilesPath(Base):
     tile_source = cast(str, Column(String))
     background_task_id = cast(str, Column(String))
     created_at = cast(datetime, Column(DateTime, default=timestamp))
+
+
+class DbSubmissionPhotos(Base):
+    """Keeping track of submission photos for a project."""
+
+    __tablename__ = "submission_photos"
+
+    id = cast(int, Column(Integer, primary_key=True))
+    project_id = cast(int, Column(Integer))
+    task_id = cast(int, Column(Integer))
+    submission_id = cast(str, Column(String))
+    s3_path = cast(str, Column(String))

src/backend/app/s3.py

@@ -1,5 +1,6 @@
 """Initialise the S3 buckets for FMTM to function."""
 
+from fastapi import HTTPException
 import json
 import sys
 from io import BytesIO
@@ -9,8 +10,10 @@
 from minio import Minio
 from minio.commonconfig import CopySource
 from minio.deleteobjects import DeleteObject
+from minio.error import S3Error
 
 from app.config import settings
+from app.models.enums import HTTPStatus
 
 
 def s3_client():
@@ -37,6 +40,32 @@ def s3_client():
     # print(creds.access_key)
     # print(creds.secret_key)
 
+def object_exists(bucket_name: str, s3_path: str) -> bool:
+    """Check if an object exists in an S3 bucket using stat_object.
+
+    Args:
+        bucket_name (str): The name of the S3 bucket.
+        s3_path (str): The path of the object in the S3 bucket.
+
+    Returns:
+        bool: True if the object exists, False otherwise.
+    """
+    client = s3_client()
+
+    try:
+        # stat_object will return metadata if the object exists
+        client.stat_object(bucket_name, s3_path)
+        return True
+    except S3Error as e:
+        if e.code == 'NoSuchKey':
+            return False
+        else:
+            # Handle other exceptions
+            raise HTTPException(
+                status_code = HTTPStatus.BAD_REQUEST, 
+                detail = str(e)
+            ) from e
+
 
 def add_file_to_bucket(bucket_name: str, file_path: str, s3_path: str):
     """Upload a file from the filesystem to an S3 bucket.

src/backend/app/submissions/submission_crud.py

@@ -33,6 +33,7 @@
 from loguru import logger as log
 
 # from osm_fieldwork.json2osm import json2osm
+from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.central.central_crud import (
@@ -44,7 +45,7 @@
 from app.db import db_models
 from app.models.enums import HTTPStatus
 from app.projects import project_crud, project_deps
-from app.s3 import add_obj_to_bucket, get_obj_from_bucket
+from app.s3 import add_obj_to_bucket, get_obj_from_bucket, object_exists
 from app.tasks import tasks_crud
 
 # async def convert_json_to_osm(file_path):
@@ -565,3 +566,89 @@ async def get_submission_detail(
 #         all_features.append(feature)
 
 #     return geojson.FeatureCollection(features=all_features)
+
+
+async def upload_attachment_to_s3(
+        project_id: int, 
+        instance_ids: list,
+        background_task_id: uuid.UUID,
+        db: Session
+    ):
+    """
+    Uploads attachments to S3 for a given project and instance IDs.
+
+    Args:
+        project_id (int): The ID of the project.
+        instance_ids (list): List of instance IDs.
+        background_task_id (uuid.UUID): The ID of the background task.
+        db (Session): The database session.
+
+    Returns:
+        bool: True if the upload is successful.
+
+    Raises:
+        Exception: If an error occurs during the upload process.
+    """
+    try:
+        project = await project_deps.get_project_by_id(db,project_id)
+        db_xform = await project_deps.get_project_xform(db, project_id)
+        odk_central = await project_deps.get_odk_credentials(db, project_id)
+        xform = get_odk_form(odk_central)
+        s3_bucket = settings.S3_BUCKET_NAME
+
+        for instance_id in instance_ids:
+            submission_detail = await get_submission_detail(instance_id, project, db)
+            attachments = submission_detail["verification"]["image"]
+
+            if not isinstance(attachments, list):
+                attachments = [attachments]
+
+            for idx, filename in enumerate(attachments):
+                s3_key = f"fmtm-data/{project.organisation_id}/{project_id}/{instance_id}/{idx+1}.jpeg"
+
+                if object_exists(s3_bucket, s3_key):
+                    log.warning(f"Object {s3_key} already exists in S3. Skipping upload.")
+                    continue
+
+                try:
+                    if attachment:= xform.getMedia(
+                        project.odkid, 
+                        str(instance_id), 
+                        db_xform.odk_form_id, 
+                        str(filename)
+                    ):
+                        # Convert the attachment to a BytesIO stream
+                        image_stream = io.BytesIO(attachment)
+
+                        # Upload the attachment to S3
+                        add_obj_to_bucket(s3_bucket, image_stream, s3_key, content_type="image/jpeg")
+
+                        # Generate the image URL
+                        img_url = f"{settings.S3_DOWNLOAD_ROOT}/{settings.S3_BUCKET_NAME}/{s3_key}"
+
+                        # Insert the record into submission_photos table
+                        sql = text("""
+                        INSERT INTO submission_photos (project_id, task_id, submission_id, s3_path)
+                        VALUES (:project_id, :task_id, :submission_id, :s3_path)
+                        """)
+                        db.execute(sql, {
+                            "project_id": project_id,
+                            "task_id": submission_detail["task_id"],
+                            "submission_id": instance_id,
+                            "s3_path": img_url
+                        })
+
+                except Exception as e:
+                    log.warning(f"Failed to process {filename} for instance {instance_id}: {e}")
+                    continue
+
+        db.commit()
+        return True
+
+    except Exception as e:
+        log.warning(str(e))
+        # Update background task status to FAILED
+        update_bg_task_sync = async_to_sync(
+            project_crud.update_background_task_status_in_database
+        )
+        update_bg_task_sync(db, background_task_id, 2, str(e)) 

src/backend/app/submissions/submission_routes.py

@@ -18,11 +18,13 @@
 """Routes associated with data submission to and from ODK Central."""
 
 import json
+import uuid
 from io import BytesIO
+from loguru import logger as log
 from typing import Annotated, Optional
 
 import geojson
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException, Query, BackgroundTasks
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import FileResponse, JSONResponse, Response
 from sqlalchemy.orm import Session
@@ -333,8 +335,10 @@ async def get_submission_form_fields(
 
 @router.get("/submission_table")
 async def submission_table(
+    background_tasks: BackgroundTasks,
     page: int = Query(1, ge=1),
     results_per_page: int = Query(13, le=100),
+    background_task_id: Optional[uuid.UUID] = None,
     task_id: Optional[int] = None,
     submitted_by: Optional[str] = None,
     review_state: Optional[str] = None,
@@ -379,6 +383,19 @@ async def submission_table(
     data = await submission_crud.get_submission_by_project(project, filters, db)
     count = data.get("@odata.count", 0)
     submissions = data.get("value", [])
+    instance_ids = []
+    for submission in submissions:
+        if submission["__system"]["attachmentsPresent"] != 0:
+            instance_ids.append(submission["__id"])
+
+    if instance_ids:
+        background_task_id = await project_crud.insert_background_task_into_database(
+            db, "upload_submission_photos", project.id
+        )
+        log.info("uploading submission photos to s3")
+        background_tasks.add_task(
+            submission_crud.upload_attachment_to_s3, project.id, instance_ids, background_task_id, db
+        )
 
     if task_id:
         submissions = [sub for sub in submissions if sub.get("task_id") == str(task_id)]

src/backend/migrations/006-create-submission-photos-table.sql

@@ -0,0 +1,17 @@
+-- ## Migration to:
+-- * Create submission_photos table
+
+-- Start a transaction
+BEGIN;
+
+CREATE TABLE submission_photos (
+    id SERIAL PRIMARY KEY,
+    project_id INTEGER NOT NULL,
+    task_id INTEGER NOT NULL,
+    submission_id VARCHAR NOT NULL,
+    s3_path VARCHAR NOT NULL
+);
+ALTER TABLE public.submission_photos OWNER TO fmtm;
+
+-- Commit the transaction
+COMMIT;

src/backend/migrations/revert/006-create-submission-photos-table.sql

@@ -0,0 +1,10 @@
+-- ## Revert Migration to:
+-- * Drop submission_photos table
+
+-- Start a transaction
+BEGIN;
+
+DROP TABLE IF EXISTS public.submission_photos;
+
+-- Commit the transaction
+COMMIT;