Added Party CSV Export Functionality

backend/src/modules/party/party_router.py

@@ -1,4 +1,7 @@
-from fastapi import APIRouter, Depends, Query
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import Response
 from src.core.authentication import (
     authenticate_admin,
     authenticate_by_role,
@@ -112,6 +115,51 @@ async def list_parties(
     )
 
 
+@party_router.get("/csv")
+async def get_parties_csv(
+    start_date: str = Query(..., description="Start date in YYYY-MM-DD format"),
+    end_date: str = Query(..., description="End date in YYYY-MM-DD format"),
+    party_service: PartyService = Depends(),
+    _=Depends(authenticate_admin),
+) -> Response:
+    """
+    Returns parties within the specified date range as a CSV file.
+
+    Query Parameters:
+    - start_date: Start date in YYYY-MM-DD format (required)
+    - end_date: End date in YYYY-MM-DD format (required)
+
+    Returns:
+    - CSV file stream with party data
+
+    Raises:
+    - 400: If date format is invalid
+    """
+    try:
+        start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
+        end_datetime = datetime.strptime(end_date, "%Y-%m-%d")
+
+        end_datetime = end_datetime.replace(
+            hour=23, minute=59, second=59, microsecond=999999
+        )
+    except ValueError:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid date format. Use YYYY-MM-DD format for dates.",
+        )
+
+    parties = await party_service.get_parties_by_date_range(
+        start_datetime, end_datetime
+    )
+    csv_content = await party_service.export_parties_to_csv(parties)
+
+    return Response(
+        content=csv_content,
+        media_type="text/csv",
+        headers={"Content-Disposition": "attachment; filename=parties.csv"},
+    )
+
+
 @party_router.put("/{party_id}")
 async def update_party(
     party_id: int,

backend/src/modules/party/party_service.py

@@ -1,3 +1,5 @@
+import csv
+import io
 import math
 from datetime import datetime, timedelta
 from typing import List
@@ -290,9 +292,7 @@ async def create_party_from_admin_dto(self, dto: AdminCreatePartyDTO) -> Party:
         location = await self._validate_and_get_location(dto.place_id)
 
         # Get contact_one by email
-        contact_one = await self._get_student_by_email(
-            dto.contact_one_email
-        )
+        contact_one = await self._get_student_by_email(dto.contact_one_email)
 
         # Create party data with contact_two information directly
         party_data = PartyData(
@@ -465,3 +465,136 @@ def _calculate_haversine_distance(
 
         r = 3959
         return c * r
+
+    async def export_parties_to_csv(self, parties: List[Party]) -> str:
+        """
+        Export a list of parties to CSV format.
+
+        Args:
+            parties: List of Party models to export
+
+        Returns:
+            CSV content as a string
+        """
+        if not parties:
+            output = io.StringIO()
+            writer = csv.writer(output)
+            writer.writerow(
+                [
+                    "Fully formatted address",
+                    "Date of Party",
+                    "Time of Party",
+                    "Contact One Full Name",
+                    "Contact One Email",
+                    "Contact One Phone Number",
+                    "Contact One Contact Preference",
+                    "Contact Two Full Name",
+                    "Contact Two Email",
+                    "Contact Two Phone Number",
+                    "Contact Two Contact Preference",
+                ]
+            )
+            return output.getvalue()
+
+        party_ids = [party.id for party in parties]
+
+        result = await self.session.execute(
+            select(PartyEntity)
+            .options(
+                selectinload(PartyEntity.location),
+                selectinload(PartyEntity.contact_one).selectinload(
+                    StudentEntity.account
+                ),
+            )
+            .where(PartyEntity.id.in_(party_ids))
+        )
+        party_entities = result.scalars().all()
+
+        party_entity_map = {party.id: party for party in party_entities}
+
+        output = io.StringIO()
+        writer = csv.writer(output)
+
+        writer.writerow(
+            [
+                "Fully formatted address",
+                "Date of Party",
+                "Time of Party",
+                "Contact One Full Name",
+                "Contact One Email",
+                "Contact One Phone Number",
+                "Contact One Contact Preference",
+                "Contact Two Full Name",
+                "Contact Two Email",
+                "Contact Two Phone Number",
+                "Contact Two Contact Preference",
+            ]
+        )
+
+        for party in parties:
+            party_entity = party_entity_map.get(party.id)
+            if party_entity is None:
+                continue
+
+            # Format address
+            formatted_address = ""
+            if party_entity.location:
+                formatted_address = party_entity.location.formatted_address or ""
+
+            # Format date and time
+            party_date = (
+                party.party_datetime.strftime("%Y-%m-%d")
+                if party.party_datetime
+                else ""
+            )
+            party_time = (
+                party.party_datetime.strftime("%H:%M:%S")
+                if party.party_datetime
+                else ""
+            )
+
+            contact_one_full_name = ""
+            contact_one_email = ""
+            contact_one_phone = ""
+            contact_one_preference = ""
+            if party_entity.contact_one:
+                contact_one_full_name = f"{party_entity.contact_one.account.first_name} {party_entity.contact_one.account.last_name}"
+                contact_one_phone = party_entity.contact_one.phone_number or ""
+                contact_one_preference = (
+                    party_entity.contact_one.contact_preference.value
+                    if party_entity.contact_one.contact_preference
+                    else ""
+                )
+                if party_entity.contact_one.account:
+                    contact_one_email = party_entity.contact_one.account.email or ""
+
+            contact_two_full_name = ""
+            contact_two_email = ""
+            contact_two_phone = ""
+            contact_two_preference = ""
+            contact_two_full_name = f"{party_entity.contact_two_first_name} {party_entity.contact_two_last_name}"
+            contact_two_phone = party_entity.contact_two_phone_number or ""
+            contact_two_preference = (
+                party_entity.contact_two_contact_preference.value
+                if party_entity.contact_two_contact_preference
+                else ""
+            )
+            contact_two_email = party_entity.contact_two_email or ""
+
+            writer.writerow(
+                [
+                    formatted_address,
+                    party_date,
+                    party_time,
+                    contact_one_full_name,
+                    contact_one_email,
+                    contact_one_phone,
+                    contact_one_preference,
+                    contact_two_full_name,
+                    contact_two_email,
+                    contact_two_phone,
+                    contact_two_preference,
+                ]
+            )
+
+        return output.getvalue()