Coverage for src/usaspending/models/agency.py: 92%

1"""Agency model for USASpending data.""" 

2 

3from __future__ import annotations 

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

5from dataclasses import dataclass 

6from ..utils.formatter import to_float, to_int, to_date 

7from datetime import date 

8from functools import cached_property 

9from .lazy_record import LazyRecord 

10from ..logging_config import USASpendingLogger 

11from ..config import ( 

12 CONTRACT_CODES, 

13 GRANT_CODES, 

14 IDV_CODES, 

15 LOAN_CODES, 

16 DIRECT_PAYMENT_CODES, 

17 OTHER_CODES 

18) 

19 

20if TYPE_CHECKING: 

21 from ..client import USASpending 

22 from ..queries.awards_search import AwardsSearch 

23 from .subtier_agency import SubTierAgency 

24 

25logger = USASpendingLogger.get_logger(__name__) 

26 

27# Create data class for def_codes 

28@dataclass 

29class DefCode: 

30 code: str 

31 public_law: str 

32 title: Optional[str] = None 

33 urls: Optional[List[str]] = None 

34 disaster: Optional[str] = None 

35 

36 

37class Agency(LazyRecord): 

38 """Rich wrapper around a USAspending toptier agency record. 

39  

40 This model represents a toptier agency with its essential properties. 

41 For subtier agency information, use the SubTierAgency model separately. 

42 """ 

43 

44 def __init__(self, data: Dict[str, Any], client: USASpending, subtier_data: Optional[Dict[str, Any]] = None): 

45 """Initialize Agency instance. 

46 

47 Args: 

48 data: Toptier agency data merged with top-level agency fields 

49 client: USASpending client instance 

50 subtier_data: Optional subtier agency data for subtier_agency property 

51 """ 

52 super().__init__(data, client) 

53 

54 # Store subtier data separately 

55 self._subtier_data = subtier_data 

56 

57 

58 def _fetch_details(self) -> Optional[Dict[str, Any]]: 

59 """Fetch full agency details if we have a toptier_code and client. 

60  

61 Returns: 

62 Full agency data from the API, or None if unable to fetch 

63 """ 

64 # Try to get toptier_code from existing data 

65 toptier_code = None 

66 if "toptier_code" in self._data: 

67 toptier_code = self._data["toptier_code"] 

68 elif "code" in self._data: 

69 toptier_code = self._data["code"] 

70 

71 try: 

72 # Fetch full agency details using the toptier_code 

73 from ..queries.agency_query import AgencyQuery 

74 query = AgencyQuery(self._client) 

75 

76 # Get fiscal_year if available in current data 

77 fiscal_year = self._data.get("fiscal_year") 

78 full_agency = query._get_resource_with_params(toptier_code, fiscal_year) 

79 

80 return full_agency 

81 except Exception as e: 

82 # Log but don't raise - lazy loading should fail gracefully 

83 logger.debug(f"Could not fetch agency details for {toptier_code}: {e}") 

84 return None 

85 

86 def _get_award_summary( 

87 self, 

88 award_type_codes: Optional[List[str]] = None, 

89 fiscal_year: Optional[int] = None, 

90 agency_type: str = "awarding" 

91 ) -> Optional[Dict[str, Any]]: 

92 """Fetch award summary data for a given agency code 

93  

94 Args: 

95 award_type_codes: Optional list of award type codes to filter 

96 fiscal_year: If none, defaults to the current fiscal year 

97 agency_type: "awarding" or "funding" 

98  

99 Returns: 

100 Award summary data dict or None if unable to fetch 

101 """ 

102 # Get toptier code 

103 toptier_code = self.code 

104 if not toptier_code: 

105 logger.error("Cannot fetch agency award summaries without agency code.") 

106 return None 

107 

108 try: 

109 from ..queries.agency_award_summary import AgencyAwardSummary 

110 query = AgencyAwardSummary(self._client) 

111 

112 return query.get_awards_summary( 

113 toptier_code=toptier_code, 

114 fiscal_year=fiscal_year, 

115 agency_type=agency_type, 

116 award_type_codes=award_type_codes 

117 ) 

118 except Exception as e: 

119 logger.error( 

120 f"Could not fetch award summary for {toptier_code}: {e}" 

121 ) 

122 return {} 

123 

124 # Properties from full agency API endpoint 

125 

126 @property 

127 def fiscal_year(self) -> Optional[int]: 

128 """Fiscal year for the agency data.""" 

129 fiscal_year = self._lazy_get("fiscal_year") 

130 return to_int(fiscal_year) 

131 

132 @property 

133 def toptier_code(self) -> Optional[str]: 

134 """ 

135 Agency toptier code (3-4 digit string). 

136 This is the Treasury Account Fund Symbol (TAFS). 

137 """ 

138 return self._lazy_get("toptier_code","code") 

139 

140 @property 

141 def code(self) -> Optional[str]: 

142 """Alias for toptier_code.""" 

143 return self.toptier_code 

144 

145 @property 

146 def name(self) -> Optional[str]: 

147 """Primary agency name.""" 

148 # Agency now contains toptier data directly 

149 return self._lazy_get("name") 

150 

151 @property 

152 def abbreviation(self) -> Optional[str]: 

153 """Primary agency abbreviation.""" 

154 # Agency now contains toptier data directly 

155 return self._lazy_get("abbreviation") 

156 

157 @property 

158 def id(self): 

159 """Internal identifier from USASpending.gov """ 

160 return self.agency_id 

161 

162 @property 

163 def agency_id(self) -> Optional[int]: 

164 """Internal identifier from USASpending.gov""" 

165 agency_id = self._lazy_get("agency_id","id") 

166 return to_int(agency_id) 

167 

168 @property 

169 def icon_filename(self) -> Optional[str]: 

170 """Filename of the agency's icon/logo.""" 

171 return self._lazy_get("icon_filename") 

172 

173 @property 

174 def mission(self) -> Optional[str]: 

175 """Agency mission statement.""" 

176 return self._lazy_get("mission") 

177 

178 @property 

179 def website(self) -> Optional[str]: 

180 """Agency website URL.""" 

181 return self._lazy_get("website") 

182 

183 @property 

184 def congressional_justification_url(self) -> Optional[str]: 

185 """URL to the agency's congressional justification.""" 

186 return self._lazy_get("congressional_justification_url") 

187 

188 @property 

189 def about_agency_data(self) -> Optional[str]: 

190 """Additional information about the agency's data.""" 

191 return self._lazy_get("about_agency_data") 

192 

193 @property 

194 def subtier_agency_count(self) -> Optional[int]: 

195 """Number of subtier agencies under this agency.""" 

196 count = self._lazy_get("subtier_agency_count") 

197 return to_int(count) 

198 

199 @property 

200 def messages(self) -> List[str]: 

201 """API messages related to this agency data.""" 

202 messages = self._lazy_get("messages", default=[]) 

203 if not isinstance(messages, list): 

204 return [] 

205 return messages 

206 

207 @property 

208 def def_codes(self) -> List[DefCode]: 

209 """List of Disaster Emergency Fund Codes (DEFC) for this agency. 

210  

211 Returns: 

212 List of DefCode dataclass instances 

213 """ 

214 def_codes_data = self._lazy_get("def_codes", default=[]) 

215 if not isinstance(def_codes_data, list): 

216 return [] 

217 

218 result = [] 

219 for code_data in def_codes_data: 

220 if isinstance(code_data, dict): 

221 # Handle the case where urls might be a string or list 

222 urls = code_data.get("urls") 

223 if isinstance(urls, str): 

224 urls = [urls] if urls else None 

225 elif urls and not isinstance(urls, list): 

226 urls = None 

227 

228 def_code = DefCode( 

229 code=code_data.get("code", ""), 

230 public_law=code_data.get("public_law", ""), 

231 title=code_data.get("title"), 

232 urls=urls, 

233 disaster=code_data.get("disaster") 

234 ) 

235 result.append(def_code) 

236 

237 return result 

238 

239 # Properties derived or related to the agency record 

240 # These properties are not included in the agency detail API endpoint 

241 # (generally, they come from a related agency properties in an award) 

242 # so they cannot be lazy-loaded. 

243 

244 @property 

245 def has_agency_page(self) -> bool: 

246 """Whether this agency has a dedicated page on USASpending.gov.""" 

247 return bool(self.get_value(["has_agency_page"], default=False)) 

248 

249 @property 

250 def office_agency_name(self) -> Optional[str]: 

251 """Name of the specific office within the agency.""" 

252 return self.get_value("office_agency_name") 

253 

254 @property 

255 def slug(self) -> Optional[str]: 

256 """URL slug for this agency.""" 

257 return self.get_value("slug") 

258 

259 @property 

260 def obligations(self) -> Optional[float]: 

261 """ Return current fiscal year's total obligations """ 

262 return self.total_obligations 

263 

264 # Related and derived resources. 

265 # Some of these properties are provided by search query 

266 # results, others are helper methods that provide quick access 

267 # to related award and transaction data. 

268 

269 @cached_property 

270 def total_obligations(self) -> Optional[float]: 

271 """ Return current fiscal year's total obligations """ 

272 obligations = self.get_value(["total_obligations","obligations"]) 

273 if not obligations: 

274 # If not present, fetch from award summary 

275 obligations = self.get_obligations() 

276 return obligations 

277 

278 @cached_property 

279 def latest_action_date(self) -> Optional[date]: 

280 """Date of the most recent action for this agency's awards.""" 

281 

282 # Check if value is present already (often provided in search results) 

283 latest_action_date_string = self.get_value("latest_action_date") 

284 

285 # If not, fetch from agency award summary endpoint 

286 if not latest_action_date_string: 

287 summary = self._get_award_summary() 

288 latest_action_date_string = summary.get("latest_action_date") 

289 

290 return to_date(latest_action_date_string) 

291 

292 @cached_property 

293 def transaction_count(self) -> Optional[int]: 

294 """Total transaction count for this agency across all awards.""" 

295 

296 # Check if value is present already (often provided in search results) 

297 transaction_count = self.get_value("transaction_count") 

298 

299 # If not, fetch from agency award summary endpoint 

300 if not transaction_count: 

301 transaction_count = self.get_transaction_count() 

302 

303 return to_int(transaction_count) 

304 

305 @property 

306 def awards(self) -> "AwardsSearch": 

307 """Get an AwardsSearch instance pre-filtered to the current agency as a top-tier "Awarding" agency. 

308  

309 Returns: 

310 AwardsSearch instance 

311 """ 

312 from ..queries.filters import AgencyTier, AgencyType 

313 return self._client.awards.search().for_agency(self.name,AgencyType.AWARDING,AgencyTier.TOPTIER) 

314 

315 @property 

316 def subagencies(self) -> List["SubTierAgency"]: 

317 """Get list of sub-agencies for this agency. 

318  

319 Returns: 

320 List of SubTierAgency instances 

321 """ 

322 # Get toptier code 

323 toptier_code = self.code 

324 if not toptier_code: 

325 logger.error("Cannot fetch sub-agencies without agency code.") 

326 return [] 

327 

328 try: 

329 from ..queries.sub_agency_query import SubAgencyQuery 

330 from .subtier_agency import SubTierAgency 

331 

332 query = SubAgencyQuery(self._client) 

333 

334 # Use fiscal year from this agency if available 

335 fiscal_year = self.fiscal_year 

336 

337 response = query.get_subagencies( 

338 toptier_code=toptier_code, 

339 fiscal_year=fiscal_year, 

340 limit=100 # Default to maximum 

341 ) 

342 

343 # Transform results into SubTierAgency objects 

344 subagencies = [] 

345 results = response.get("results", []) 

346 for result in results: 

347 if isinstance(result, dict): 

348 subagency = SubTierAgency(result, self._client) 

349 subagencies.append(subagency) 

350 

351 return subagencies 

352 

353 except Exception as e: 

354 logger.debug( 

355 f"Could not fetch sub-agencies for {toptier_code}: {e}" 

356 ) 

357 return [] 

358 

359 def get_obligations( 

360 self, 

361 fiscal_year: Optional[int] = None, 

362 agency_type: str = "awarding", 

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

364 ) -> Optional[float]: 

365 """Get obligations for this agency, optionally filtered. 

366  

367 Args: 

368 fiscal_year: Return obligation totals for a given fiscal year (defaults to current FY) 

369 agency_type: "awarding" or "funding" 

370 award_type_codes: Optional list of award type codes to filter 

371  

372 Returns: 

373 Obligations amount or None if unavailable 

374 """ 

375 

376 # Fetch from award summary API 

377 summary = self._get_award_summary( 

378 award_type_codes=award_type_codes, 

379 fiscal_year=fiscal_year, 

380 agency_type=agency_type 

381 ) 

382 return to_float(summary.get("obligations")) if summary else None 

383 

384 @cached_property 

385 def contract_obligations(self) -> Optional[float]: 

386 """Get contract obligations for this agency. 

387 

388 Returns: 

389 Contract obligations amount or None if unavailable 

390 """ 

391 summary = self._get_award_summary( 

392 award_type_codes=list(CONTRACT_CODES) 

393 ) 

394 return to_float(summary.get("obligations")) if summary else None 

395 

396 @cached_property 

397 def grant_obligations(self) -> Optional[float]: 

398 """Get grant obligations for this agency in the current fiscal year. 

399  

400 Returns: 

401 Grant obligations amount or None if unavailable 

402 """ 

403 summary = self._get_award_summary( 

404 award_type_codes=list(GRANT_CODES) 

405 ) 

406 return to_float(summary.get("obligations")) if summary else None 

407 

408 @cached_property 

409 def idv_obligations(self) -> Optional[float]: 

410 """Get Indefinite Delivery Vehicle (IDV) obligations for this agency in the current fiscal year. 

411  

412 Returns: 

413 IDV obligations amount or None if unavailable 

414 """ 

415 summary = self._get_award_summary( 

416 award_type_codes=list(IDV_CODES) 

417 ) 

418 return to_float(summary.get("obligations")) if summary else None 

419 

420 @cached_property 

421 def loan_obligations(self) -> Optional[float]: 

422 """Get loan obligations for this agency for the current fiscal year 

423 

424 Returns: 

425 Loan obligations amount or None if unavailable 

426 """ 

427 summary = self._get_award_summary( 

428 award_type_codes=list(LOAN_CODES) 

429 ) 

430 return to_float(summary.get("obligations")) if summary else None 

431 

432 @cached_property 

433 def direct_payment_obligations(self) -> Optional[float]: 

434 """Get direct payment obligations for this agency for the current fiscal year. 

435  

436 Returns: 

437 Direct payment obligations amount or None if unavailable 

438 """ 

439 summary = self._get_award_summary( 

440 award_type_codes=list(DIRECT_PAYMENT_CODES) 

441 ) 

442 return to_float(summary.get("obligations")) if summary else None 

443 

444 @cached_property 

445 def other_obligations(self) -> Optional[float]: 

446 """Get other assistance obligations for this agency. 

447  

448 Returns: 

449 Other assistance obligations amount or None if unavailable 

450 """ 

451 summary = self._get_award_summary( 

452 award_type_codes=list(OTHER_CODES) 

453 ) 

454 return to_float(summary.get("obligations")) if summary else None 

455 

456 def get_transaction_count( 

457 self, 

458 fiscal_year: Optional[int] = None, 

459 agency_type: str = "awarding", 

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

461 ) -> Optional[int]: 

462 """Get transaction count for this agency, optionally filtered. 

463  

464 Args: 

465 fiscal_year: Override the agency's fiscal year (None uses self.fiscal_year) 

466 agency_type: "awarding" or "funding" 

467 award_type_codes: Optional list of award type codes to filter 

468  

469 Returns: 

470 Transaction count or None if unavailable 

471 """ 

472 # If no filters and we have existing data, return it 

473 if not any([fiscal_year, award_type_codes]) and agency_type == "awarding": 

474 existing = self._lazy_get("transaction_count") 

475 if existing is not None: 

476 return to_int(existing) 

477 

478 # Fetch from award summary API 

479 summary = self._get_award_summary( 

480 award_type_codes=award_type_codes, 

481 fiscal_year=fiscal_year, 

482 agency_type=agency_type 

483 ) 

484 return to_int(summary.get("transaction_count")) if summary else None 

485 

486 def __repr__(self) -> str: 

487 """String representation of Agency.""" 

488 name = self.name or "?" 

489 code = self.code or "?" 

490 return f"<Agency {code}: {name}>"