feat: add s3 function to get path org_id/proj_id

src/backend/app/s3.py

@@ -8,6 +8,8 @@
 from loguru import logger as log
 from minio import Minio
 from minio.commonconfig import CopySource
+from sqlalchemy import text
+from sqlalchemy.orm import Session
 
 from app.config import settings
 
@@ -113,7 +115,7 @@ def get_obj_from_bucket(bucket_name: str, s3_path: str) -> BytesIO:
         response = client.get_object(bucket_name, s3_path)
         return BytesIO(response.read())
     except Exception as e:
-        raise ValueError(str(e))
+        raise ValueError(str(e)) from e
     finally:
         if response:
             response.close()
@@ -159,6 +161,16 @@ def copy_obj_bucket_to_bucket(
     return True
 
 
+async def get_bucket_path(db: Session, project_id: int) -> str:
+    """For a given project id, return the S3 bucket path."""
+    log.debug(f"Getting org id for project: {project_id}")
+    query = text(f"SELECT organisation_id FROM projects WHERE id = {project_id};")
+    result = db.execute(query)
+    org_id = result.fetchone()[0]
+    log.debug(f"Extracted org id: {org_id}")
+    return f"/{org_id}/{project_id}"
+
+
 def create_bucket_if_not_exists(client: Minio, bucket_name: str, is_public: bool):
     """Checks if a bucket exits, else creates it."""
     if not client.bucket_exists(bucket_name):