Coverage for src/usaspending/resources/award_resource.py: 95%

1"""Award resource implementation.""" 

2 

3from __future__ import annotations 

4from typing import TYPE_CHECKING, Optional 

5 

6from .base_resource import BaseResource 

7from ..logging_config import USASpendingLogger 

8 

9if TYPE_CHECKING: 

10 from ..queries.awards_search import AwardsSearch 

11 from ..models.award import Award 

12 

13logger = USASpendingLogger.get_logger(__name__) 

14 

15 

16class AwardResource(BaseResource): 

17 """Resource for award-related operations. 

18 

19 Provides access to award search and retrieval endpoints. 

20 """ 

21 

22 def find_by_generated_id(self, generated_award_id: str) -> "Award": 

23 """Retrieve a single award by the system's internally generated 

24 award entry ID (e.g. "CONT_AWD_80GSFC18C0008_8000_-NONE-_-NONE-") 

25 

26 Args: 

27 generated_award_id: Unique award identifier 

28 

29 Returns: 

30 Award model instance 

31 

32 Raises: 

33 : If generated_award_id is invalid 

34 APIError: If award not found 

35 """ 

36 logger.debug(f"Retrieving award by ID: {generated_award_id}") 

37 from ..queries.award_query import AwardQuery 

38 

39 return AwardQuery(self._client).find_by_generated_id(generated_award_id) 

40 

41 def find_by_award_id(self, award_id: str) -> Optional["Award"]: 

42 """Find an award by its PIID or FAIN unique identifier. 

43 Args: 

44 award_id: Unique identifier for the award (PIID or FAIN) 

45 

46 Returns: 

47 Award model instance if found, otherwise None 

48 """ 

49 logger.debug(f"Finding award by ID: {award_id}") 

50 from ..queries.awards_search import AwardsSearch 

51 

52 # Get counts by award type 

53 search_result = AwardsSearch(self._client).with_award_ids(award_id).count_awards_by_type() 

54 

55 # Find which type has exactly one result 

56 matching_types = [(award_type, count) for award_type, count in search_result.items() if count == 1] 

57 

58 if len(matching_types) != 1: 

59 total_awards = sum(count for count in search_result.values() if count > 0) 

60 if total_awards == 0: 

61 logger.info(f"No awards found for ID {award_id}") 

62 else: 

63 logger.warning(f"Expected exactly one award for ID {award_id}, found {total_awards} awards across {len(matching_types)} types") 

64 logger.debug(f"Search result: {search_result}") 

65 return None 

66 

67 award_type, _ = matching_types[0] 

68 logger.info(f"Found 1 award of type {award_type} for ID {award_id}") 

69 

70 # Map API response keys to method names 

71 method_mapping = { 

72 "contracts": "contracts", 

73 "grants": "grants", 

74 "idvs": "idvs", 

75 "loans": "loans", 

76 "direct_payments": "direct_payments", 

77 "other": "other" 

78 } 

79 

80 method_name = method_mapping.get(award_type) 

81 if not method_name: 

82 logger.error(f"Unknown award type from API: {award_type}") 

83 return None 

84 

85 # Create search and apply the appropriate filter 

86 awards_search = AwardsSearch(self._client) 

87 if hasattr(awards_search, method_name): 

88 logger.debug(f"Calling .{method_name}() method on AwardsSearch") 

89 awards_search = getattr(awards_search, method_name)() 

90 return awards_search.with_award_ids(award_id).first() 

91 else: 

92 logger.error(f"Method {method_name} not found on AwardsSearch") 

93 return None 

94 

95 

96 

97 

98 def search(self) -> AwardsSearch: 

99 """Create a new award search query builder. 

100 

101 Returns: 

102 AwardSearch query builder for chaining filters 

103 

104 Example: 

105 >>> awards = client.awards.search() 

106 ... .for_agency("NASA") 

107 ... .in_state("TX") 

108 ... .fiscal_years(2023, 2024) 

109 ... .limit(10) 

110 """ 

111 logger.debug("Creating new AwardsSearch query builder") 

112 from ..queries.awards_search import AwardsSearch 

113 

114 return AwardsSearch(self._client)