Coverage for src/usaspending/queries/agency_award_summary.py: 100%

1"""Agency award summary query implementation for retrieving award aggregations.""" 

2 

3from typing import TYPE_CHECKING, Optional, Dict, Any, List 

4from .single_resource_base import SingleResourceBase 

5from ..exceptions import ValidationError 

6from ..client import USASpending 

7from ..logging_config import USASpendingLogger 

8 

9if TYPE_CHECKING: 

10 pass 

11 

12logger = USASpendingLogger.get_logger(__name__) 

13 

14 

15class AgencyAwardSummary(SingleResourceBase): 

16 """Retrieve agency award summary data from the USAspending API. 

17  

18 This query class handles fetching aggregated award information including 

19 transaction counts and obligations filtered by fiscal year, agency type, 

20 and award type codes. 

21 """ 

22 

23 def __init__(self, client: USASpending): 

24 """Initialize AgencyAwardSummary with client. 

25  

26 Args: 

27 client: USASpending client instance 

28 """ 

29 super().__init__(client) 

30 logger.debug("AgencyAwardSummary initialized with client: %s", client) 

31 

32 @property 

33 def _endpoint(self) -> str: 

34 """Base endpoint for agency award summary retrieval.""" 

35 return "/agency/" 

36 

37 def _construct_endpoint(self, resource_id: str) -> str: 

38 """Construct the full endpoint URL for agency award summary. 

39  

40 Args: 

41 resource_id: The toptier_code for the agency 

42  

43 Returns: 

44 Full endpoint path including /awards/ 

45 """ 

46 return f"{self._endpoint}{resource_id}/awards/" 

47 

48 def find_by_id(self, toptier_code: str) -> Dict[str, Any]: 

49 """Not used for award summary - use get_awards_summary instead. 

50  

51 Raises: 

52 NotImplementedError: This method should not be used directly 

53 """ 

54 raise NotImplementedError( 

55 "Use get_awards_summary() method instead for award summary data" 

56 ) 

57 

58 def get_awards_summary( 

59 self, 

60 toptier_code: str, 

61 fiscal_year: Optional[int] = None, 

62 agency_type: str = "awarding", 

63 award_type_codes: Optional[List[str]] = None 

64 ) -> Dict[str, Any]: 

65 """Retrieve agency award summary with optional filters. 

66  

67 Args: 

68 toptier_code: The toptier code of an agency (3-4 digit string) 

69 fiscal_year: Optional fiscal year for the data (defaults to current) 

70 agency_type: "awarding" or "funding" (defaults to "awarding") 

71 award_type_codes: Optional list of award type codes to filter by 

72  

73 Returns: 

74 Dictionary containing: 

75 - toptier_code: Agency toptier code 

76 - fiscal_year: Fiscal year of data 

77 - latest_action_date: Latest transaction date 

78 - transaction_count: Number of transactions 

79 - obligations: Total obligations amount 

80 - messages: Any API messages 

81  

82 Raises: 

83 ValidationError: If toptier_code is invalid or agency_type is invalid 

84 APIError: If API request fails 

85 """ 

86 # Validate toptier_code 

87 if not toptier_code: 

88 raise ValidationError("toptier_code is required") 

89 

90 toptier_code = str(toptier_code).strip() 

91 if not toptier_code.isdigit() or len(toptier_code) not in [3, 4]: 

92 raise ValidationError( 

93 f"Invalid toptier_code: {toptier_code}. " 

94 "Must be a 3-4 digit numeric string" 

95 ) 

96 

97 # Validate agency_type 

98 if agency_type not in ["awarding", "funding"]: 

99 raise ValidationError( 

100 f"Invalid agency_type: {agency_type}. " 

101 "Must be 'awarding' or 'funding'" 

102 ) 

103 

104 logger.debug( 

105 "Fetching award summary for toptier_code: %s, fiscal_year: %s, " 

106 "agency_type: %s, award_type_codes: %s", 

107 toptier_code, fiscal_year, agency_type, award_type_codes 

108 ) 

109 

110 # Build params 

111 params = {"agency_type": agency_type} 

112 

113 if fiscal_year is not None: 

114 params["fiscal_year"] = fiscal_year 

115 

116 if award_type_codes: 

117 # Convert to list if needed and filter out None/empty values 

118 if isinstance(award_type_codes, (set, frozenset)): 

119 award_type_codes = list(award_type_codes) 

120 award_type_codes = [code for code in award_type_codes if code] 

121 

122 if award_type_codes: 

123 # API expects award_type_codes as array parameter 

124 params["award_type_codes"] = award_type_codes 

125 

126 # Construct endpoint 

127 endpoint = self._construct_endpoint(toptier_code) 

128 

129 # Make API request with params 

130 response = self._client._make_request( 

131 "GET", 

132 endpoint, 

133 params=params 

134 ) 

135 

136 return response