Coverage for src/usaspending/resources/recipients_resource.py: 33%

1"""Recipient resource implementation.""" 

2 

3from __future__ import annotations 

4from typing import Optional, TYPE_CHECKING 

5 

6from .base_resource import BaseResource 

7from ..logging_config import USASpendingLogger 

8 

9if TYPE_CHECKING: 

10 from ..queries.recipients_search import RecipientsSearch 

11 from ..models.recipient import Recipient 

12 

13logger = USASpendingLogger.get_logger(__name__) 

14 

15 

16class RecipientsResource(BaseResource): 

17 """Resource for recipient-related operations. 

18 

19 Provides access to recipient search and retrieval endpoints. 

20 """ 

21 

22 def find_by_recipient_id(self, recipient_id: str) -> Optional["Recipient"]: 

23 """Retrieve a single recipient by ID. 

24 

25 Args: 

26 recipient_id: Unique recipient identifier 

27 

28 Returns: 

29 Recipient model instance 

30 

31 Raises: 

32 ValidationError: If recipient_id is invalid 

33 APIError: If recipient not found 

34 """ 

35 logger.debug(f"Retrieving recipient by ID: {recipient_id}") 

36 from ..queries.recipient_query import RecipientQuery 

37 

38 return RecipientQuery(self._client).find_by_id(recipient_id) 

39 

40 def search(self) -> "RecipientsSearch": 

41 """Create a new recipient search query builder. 

42 

43 Returns: 

44 RecipientsSearch query builder for recipient searches 

45  

46 Example: 

47 >>> recipients = client.recipients.search() 

48 ... .with_keyword("california") 

49 ... .with_award_type("contracts") 

50 ... .order_by("amount", "desc") 

51 ... .limit(10) 

52 """ 

53 logger.debug("Creating new RecipientsSearch query builder for recipient searches") 

54 from ..queries.recipients_search import RecipientsSearch 

55 

56 return RecipientsSearch(self._client) 

57 

58 def find_by_duns(self, duns: str) -> Optional["Recipient"]: 

59 """Retrieve a single recipient by DUNS number. 

60 

61 Args: 

62 duns: Unique DUNS identifier 

63 

64 Returns: 

65 Recipient model instance or None if not found 

66 

67 Raises: 

68 ValidationError: If duns is invalid 

69 """ 

70 logger.debug(f"Searching recipient by DUNS: {duns}") 

71 from ..queries.recipients_search import RecipientsSearch 

72 recipients = RecipientsSearch(self._client).with_keyword(duns).limit(4) 

73 # Return the parent recipient if available, otherwise first result 

74 for r in recipients: 

75 if "-P" in r.recipient_id: 

76 return r 

77 # Return first result if no parent found (avoids hanging len() call) 

78 return recipients.first() 

79 

80 def find_by_uei(self, uei: str) -> Optional["Recipient"]: 

81 """Retrieve a single recipient by UEI number. 

82 

83 Args: 

84 uei: Unique Entity Identifier 

85 

86 Returns: 

87 Recipient model instance or None if not found 

88 

89 Raises: 

90 ValidationError: If uei is invalid 

91 """ 

92 logger.debug(f"Searching recipient by UEI: {uei}") 

93 from ..queries.recipients_search import RecipientsSearch 

94 recipients = RecipientsSearch(self._client).with_keyword(uei).limit(4) 

95 # Return the parent recipient if available, otherwise first result 

96 for r in recipients: 

97 if "-P" in r.recipient_id: 

98 return r 

99 # Return first result if no parent found (avoids hanging len() call) 

100 return recipients.first() 

101