Added Party CSV Export Functionality

backend/src/modules/location/location_router.py

@@ -8,7 +8,7 @@
 )
 from src.modules.location.location_service import LocationService
 
-location_router = APIRouter(prefix="/locations", tags=["locations"])
+location_router = APIRouter(prefix="/api/locations", tags=["locations"])
 
 
 @location_router.get("/", response_model=PaginatedLocationResponse)

backend/src/modules/party/party_router.py

@@ -1,5 +1,12 @@
-from fastapi import APIRouter, Depends, Query
-from src.core.authentication import authenticate_admin, authenticate_by_role, authenticate_staff_or_admin
+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,
+    authenticate_staff_or_admin,
+)
 
 from .party_model import PaginatedPartiesResponse, Party
 from .party_service import PartyService
@@ -64,6 +71,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.get("/{party_id}")
 async def get_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
@@ -230,3 +232,141 @@ 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
+                ),
+                selectinload(PartyEntity.contact_two).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 = ""
+            if party_entity.contact_two:
+                contact_two_full_name = f"{party_entity.contact_two.account.first_name} {party_entity.contact_two.account.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 ""
+                )
+                if party_entity.contact_two.account:
+                    contact_two_email = party_entity.contact_two.account.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()

backend/test/modules/location/location_router_crud_test.py

@@ -40,7 +40,7 @@ def get_mock_location_service():
     app.dependency_overrides[get_session] = override_get_session
 
     async with AsyncClient(
-        transport=ASGITransport(app=app), base_url="http://test"
+        transport=ASGITransport(app=app), base_url="http://test/api"
     ) as ac:
         yield ac
 
@@ -64,7 +64,7 @@ def get_mock_location_service():
 
     async with AsyncClient(
         transport=ASGITransport(app=app),
-        base_url="http://test",
+        base_url="http://test/api",
         headers={"Authorization": "Bearer admin"},
     ) as ac:
         yield ac