Enhance proposals endpoint with optional username, cycle, and SAF status filters for MX Webapp

src/nsls2api/api/v1/proposal_api.py

@@ -104,19 +104,26 @@ async def get_proposals(
     beamline: Annotated[list[str], Query()] = [],
     cycle: Annotated[list[str], Query()] = [],
     facility: Annotated[list[FacilityName], Query()] = [FacilityName.nsls2],
+    username: str | None = Query(None, description="Filter proposals by username"),
+    saf_status: list[str] | None = Query(default=None, description="Filter proposals and SAF's by SAF status"),
     page_size: int = Query(10, ge=1, le=200),
     page: int = Query(1, ge=1),
     include_directories: bool = False,
 ):
+
+
     proposal_list = await proposal_service.fetch_proposals(
         proposal_id=proposal_id,
         beamline=beamline,
         cycle=cycle,
         facility=facility,
+        username=username,
+        saf_status=saf_status,
         page_size=page_size,
         page=page,
         include_directories=include_directories,
     )
+
 
     response_model = {
         "proposals": proposal_list,

src/nsls2api/services/proposal_service.py

@@ -345,12 +345,16 @@ async def fetch_proposals(
     beamline: list[str] | None = None,
     cycle: list[str] | None = None,
     facility: list[str] | None = None,
+    username: str | None = None,
+    saf_status: list[str] | None = None,
     page_size: int = 10,
     page: int = 1,
     include_directories: bool = False,
+
 ) -> Optional[list[ProposalFullDetails]]:
     query = []
 
+
     if beamline:
         beamline_upper = [beamline_name.upper() for beamline_name in beamline]
         query.append(In(Proposal.instruments, beamline_upper))
@@ -360,6 +364,11 @@ async def fetch_proposals(
 
     if proposal_id:
         query.append(In(Proposal.proposal_id, proposal_id))
+
+    if username is not None:
+        username = username.strip()
+    if username:
+        query.append(ElemMatch(Proposal.users, {"username": username}))
 
     if len(query) == 0:
         proposals = (
@@ -378,6 +387,25 @@ async def fetch_proposals(
             .to_list()
         )
 
+    # SAF filtering: query narrows matching proposals, this block trims nested SAFs.
+    if saf_status:
+        filtered_proposals = []
+        for proposal in proposals:
+            proposal.safs = [
+                saf
+                for saf in (proposal.safs or [])
+                if saf.saf_id
+                and (not saf_status or (saf.status or "").strip().upper() in {s.strip().upper() for s in saf_status})
+            ]
+
+            if not proposal.safs:
+                continue
+
+            filtered_proposals.append(proposal)
+
+        proposals = filtered_proposals
+
+
     # Add directories field to each proposal
     if include_directories:
         detailed_proposals = []

src/nsls2api/tests/services/test_proposal_service.py

@@ -190,3 +190,56 @@ async def test_data_sessions_invalid_beamline(admin_api_key):
     body = resp.json()
     assert body["count"] == 0
     assert body["proposals"] == []
+
+@pytest.mark.anyio
+async def test_fetch_proposals_by_username_exists():
+    """Test filtering proposals by username that exists in test data"""
+    proposal = await proposal_service.proposal_by_id(test_proposal_id)
+    if proposal.users and len(proposal.users) > 0:
+        test_username = proposal.users[0].username
+
+        results = await proposal_service.fetch_proposals(username=test_username)
+
+        assert len(results) > 0
+
+        for p in results:
+            usernames = [u.username for u in p.users]
+            assert test_username in usernames
+
+
+@pytest.mark.anyio
+async def test_fetch_proposals_by_username_not_exists():
+    """Test filtering by username that doesn't exist"""
+    results = await proposal_service.fetch_proposals(username="nonexistent_user_xyz")
+
+    assert len(results) == 0
+
+
+@pytest.mark.anyio
+async def test_fetch_proposals_username_whitespace():
+    """Test username filtering with whitespace edge cases"""
+    results_whitespace = await proposal_service.fetch_proposals(username="   ")
+
+    results_no_filter = await proposal_service.fetch_proposals()
+
+    assert len(results_whitespace) == len(results_no_filter)
+
+
+@pytest.mark.anyio
+async def test_fetch_proposals_username_with_pagination():
+    """Test username filter works correctly with pagination"""
+    proposal = await proposal_service.proposal_by_id(test_proposal_id)
+    if proposal.users and len(proposal.users) > 0:
+        test_username = proposal.users[0].username
+
+        page_1 = await proposal_service.fetch_proposals(
+            username=test_username,
+            page_size=5,
+            page=1
+        )
+
+        assert len(page_1) > 0
+
+        for p in page_1:
+            usernames = [u.username for u in p.users]
+            assert test_username in usernames