Coverage for src/usaspending/queries/recipient_query.py: 0%

1from typing import TYPE_CHECKING 

2from .single_resource_base import SingleResourceBase 

3from ..exceptions import ValidationError 

4from ..client import USASpending 

5from ..logging_config import USASpendingLogger 

6 

7if TYPE_CHECKING: 

8 from ..models.recipient import Recipient 

9 

10logger = USASpendingLogger.get_logger(__name__) 

11 

12 

13class RecipientQuery(SingleResourceBase): 

14 """Retrieve a single-recipient""" 

15 

16 def __init__(self, client: USASpending): 

17 super().__init__(client) 

18 

19 @property 

20 def _endpoint(self) -> str: 

21 """Base endpoint for single recipient retrieval.""" 

22 return "/recipient/" 

23 

24 def find_by_id(self, recipient_id: str) -> "Recipient": 

25 """Filter by unique recipient identifier.""" 

26 if not recipient_id: 

27 raise ValidationError("recipient_id is required") 

28 

29 # Make API request 

30 response = self._get_resource(recipient_id) 

31 

32 # Create model instance 

33 from ..models.recipient import Recipient 

34 

35 return Recipient(response, client=self._client) 

36 

37 def _clean_recipient_id(self, recipient_id: str) -> str: 

38 """Clean recipient ID format. 

39 

40 Handles cases like "abc123-['C','R']" -> "abc123-R" 

41 """ 

42 import re 

43 

44 # Pattern for list suffix: -[...] 

45 pattern = r"^(.+?)-\[\s*([^\]]+)\]$" 

46 match = re.match(pattern, recipient_id) 

47 

48 if not match: 

49 return recipient_id 

50 

51 base = match.group(1) 

52 tokens = match.group(2) 

53 

54 # Extract and clean tokens 

55 clean_tokens = [] 

56 for token in tokens.split(","): 

57 clean_token = token.strip().strip("'\"").upper() 

58 if clean_token: 

59 clean_tokens.append(clean_token) 

60 

61 # Prefer 'R' if present, otherwise first token 

62 suffix = ( 

63 "R" if "R" in clean_tokens else (clean_tokens[0] if clean_tokens else "") 

64 ) 

65 

66 return f"{base}-{suffix}" if suffix else base