first draft of attachment presigned url generation, delete routes for updating and deleting attachments

Author: rbuels

backend/app/api/routes/attachments.py

@@ -1,13 +1,17 @@
 import uuid
 from typing import Any
 
+import boto3
 from fastapi import APIRouter, HTTPException
+from fastapi.responses import RedirectResponse
 from sqlmodel import func, select
 
-from app.api.deps import CurrentUser, SessionDep
+from app.api.deps import AwsDep, CurrentUser, SessionDep
+from app.core.config import Settings
 from app.models.attachments import (
     Attachment,
     AttachmentCreate,
+    AttachmentCreatePublic,
     AttachmentPublic,
     AttachmentsPublic,
     AttachmentUpdate,
@@ -22,7 +26,7 @@ def read_attachments(
     session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
 ) -> Any:
     """
-    Retrieve attachments.
+    Retrieve list of all attachments.
     """
     count_statement = select(func.count()).select_from(Attachment)
     count = session.exec(count_statement).one()
@@ -31,64 +35,68 @@ def read_attachments(
 
     return AttachmentsPublic(data=attachments, count=count)
 
+@router.get("/{id}/content")
+def read_attachment_content(session: SessionDep, aws_client: AwsDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
+    """
+    Get attachment content by ID.
+    """
+    attachment = session.get(Attachment, id)
+    if not attachment:
+        raise HTTPException(status_code=404, detail="Attachment not found")
+
+    try:
+        presigned_url = aws_client.generate_presigned_url(
+            "get_object",
+            Params={
+                "Bucket": Settings.AWS_S3_ATTACHMENTS_BUCKET,
+                "Key": attachment.storage_path,
+                "ContentDisposition": f"attachment; filename={attachment.file_name}",
+            },
+            ExpiresIn=3600,
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail="Could not generate presigned URL")
+
+    return RedirectResponse(status_code=302, url=presigned_url)
+
 
 @router.get("/{id}", response_model=AttachmentPublic)
 def read_attachment(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
     """
-    Get attachment by ID.
+    Get attachment details by ID.
     """
     attachment = session.get(Attachment, id)
     if not attachment:
         raise HTTPException(status_code=404, detail="Attachment not found")
     return attachment
 
 
-@router.post("/", response_model=AttachmentPublic)
+@router.post("/", response_model=AttachmentCreatePublic)
 def create_attachment(
-    *, session: SessionDep, current_user: CurrentUser, attachment_in: AttachmentCreate
+    *, session: SessionDep, aws_client: AwsDep, current_user: CurrentUser, attachment_in: AttachmentCreate
 ) -> Any:
     """
-    Create new attachment.
+    Create a new attachment.
     """
     attachment = Attachment.model_validate(attachment_in)
     session.add(attachment)
     session.commit()
     session.refresh(attachment)
-    return attachment
-
 
-@router.put("/{id}", response_model=AttachmentPublic)
-def update_attachment(
-    *,
-    session: SessionDep,
-    current_user: CurrentUser,
-    id: uuid.UUID,
-    attachment_in: AttachmentUpdate,
-) -> Any:
-    """
-    Update an attachment.
-    """
-    attachment = session.get(Attachment, id)
-    if not attachment:
-        raise HTTPException(status_code=404, detail="Attachment not found")
-    update_dict = attachment_in.model_dump(exclude_unset=True)
-    attachment.sqlmodel_update(update_dict)
-    session.add(attachment)
-    session.commit()
-    session.refresh(attachment)
-    return attachment
+    try:
+        presigned_upload_url = aws_client.generate_presigned_url(
+            "put_object",
+            Params={
+                "Bucket": Settings.AWS_S3_ATTACHMENTS_BUCKET,
+                "Key": attachment.storage_path,
+                "ContentDisposition": f"attachment; filename={attachment.file_name}",
+            },
+            ExpiresIn=3600,
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail="Could not generate presigned URL")
 
+    resmodel = AttachmentCreatePublic.model_validate(attachment)
+    resmodel.upload_url = presigned_upload_url
 
-@router.delete("/{id}")
-def delete_attachment(
-    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
-) -> Message:
-    """
-    Delete an attachment.
-    """
-    attachment = session.get(Attachment, id)
-    if not attachment:
-        raise HTTPException(status_code=404, detail="Attachment not found")
-    session.delete(attachment)
-    session.commit()
-    return Message(message="Attachment deleted successfully")
+    return resmodel

backend/app/models/attachments.py

@@ -38,9 +38,19 @@ class Attachment(AttachmentBase, table=True):
     )
     patient: Patient | None = Relationship(back_populates="attachments")
 
+    # get the blob storage path, which is assembled from the patient
+    # id and the attachment ID
+    @property
+    def storage_path(self):
+        return f"patients/{self.patient_id}/attachments/{self.id}"
+
 # Properties to return via API, id is always required
 class AttachmentPublic(AttachmentBase):
     id: uuid.UUID
+    patient_id: uuid.UUID = Field(nullable=False)
+
+class AttachmentCreatePublic(AttachmentPublic):
+    upload_url: str = Field(nullable=False) # presigned URL for uploading to blob storage
 
 class AttachmentsPublic(SQLModel):
     data: list[AttachmentPublic]

backend/app/tests/api/routes/test_attachments.py

@@ -38,9 +38,10 @@ def test_create_attachment(
     content = response.json()
     assert content["file_name"] == data["file_name"]
     assert content["mime_type"] == data["mime_type"]
+    assert content["upload_url"] is not None
     assert "id" in content
 
-def test_read_attachment(
+def test_read_attachment_details(
     client: TestClient, superuser_token_headers: Dict[str, str], db: Session
 ) -> None:
     # Create test patient
@@ -73,8 +74,10 @@ def test_read_attachment(
     content = response.json()
     assert content["file_name"] == attachment.file_name
     assert content["id"] == str(attachment.id)
+    assert content["patient_id"] == str(attachment.patient_id)
 
-def test_read_attachments(
+
+def test_list_attachments(
     client: TestClient, superuser_token_headers: Dict[str, str], db: Session
 ) -> None:
     # Create test patient
@@ -113,77 +116,6 @@ def test_read_attachments(
     assert content["count"] >= 2
     assert len(content["data"]) >= 2
 
-def test_update_attachment(
-    client: TestClient, superuser_token_headers: Dict[str, str], db: Session
-) -> None:
-    # Create test patient
-    patient = Patient(
-        name="Test Patient",
-        dob=datetime(2000, 1, 1),
-        contact_info="[email protected]"
-    )
-    db.add(patient)
-    db.commit()
-    db.refresh(patient)
-
-    # Create test attachment
-    attachment = Attachment(
-        file_name="test.pdf",
-        mime_type="application/pdf",
-        storage_path="patients/attachments/test.pdf",
-        patient_id=patient.id
-    )
-    db.add(attachment)
-    db.commit()
-    db.refresh(attachment)
-
-    data = {"file_name": "updated.pdf"}
-    response = client.put(
-        f"/api/v1/attachments/{attachment.id}",
-        headers=superuser_token_headers,
-        json=data,
-    )
-    assert response.status_code == 200
-    content = response.json()
-    assert content["file_name"] == data["file_name"]
-    assert content["id"] == str(attachment.id)
-
-def test_delete_attachment(
-    client: TestClient, superuser_token_headers: Dict[str, str], db: Session
-) -> None:
-    # Create test patient
-    patient = Patient(
-        name="Test Patient",
-        dob=datetime(2000, 1, 1),
-        contact_info="[email protected]"
-    )
-    db.add(patient)
-    db.commit()
-    db.refresh(patient)
-
-    # Create test attachment
-    attachment = Attachment(
-        file_name="test.pdf",
-        mime_type="application/pdf",
-        storage_path="patients/attachments/test.pdf",
-        patient_id=patient.id
-    )
-    db.add(attachment)
-    db.commit()
-    db.refresh(attachment)
-
-    response = client.delete(
-        f"/api/v1/attachments/{attachment.id}",
-        headers=superuser_token_headers,
-    )
-    assert response.status_code == 200
-    
-    # Verify attachment was deleted
-    response = client.get(
-        f"/api/v1/attachments/{attachment.id}",
-        headers=superuser_token_headers,
-    )
-    assert response.status_code == 404
 
 def test_attachment_not_found(
     client: TestClient, superuser_token_headers: Dict[str, str]