Coverage for src/usaspending/models/award.py: 78%
293 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 17:15 -0700
1"""Award model for USASpending data."""
3from __future__ import annotations
4from typing import Dict, Any, Optional, List, TYPE_CHECKING
5from functools import cached_property
6from datetime import datetime
8from .lazy_record import LazyRecord
9from .recipient import Recipient
10from .location import Location
11from .period_of_performance import PeriodOfPerformance
12from .agency import Agency
13from .subtier_agency import SubTierAgency
14from .download import AwardType, FileFormat
16from ..exceptions import ValidationError
17from ..logging_config import USASpendingLogger
18from ..utils.formatter import smart_sentence_case, to_float, to_date
20if TYPE_CHECKING:
21 from ..client import USASpending
22 from ..queries.transactions_search import TransactionsSearch
23 from ..queries.funding_search import FundingSearch
24 from ..queries.subawards_search import SubAwardsSearch
25 from ..download.job import DownloadJob
27logger = USASpendingLogger.get_logger(__name__)
29class Award(LazyRecord):
30 """Rich wrapper around a USAspending award record."""
32 # Base fields common to all award types
33 SEARCH_FIELDS = [
34 "Award ID",
35 "recipient_id",
36 "Recipient Name",
37 "Recipient DUNS Number",
38 "Recipient UEI",
39 "Recipient Location",
40 "Awarding Agency",
41 "Awarding Agency Code",
42 "Awarding Sub Agency",
43 "Awarding Sub Agency Code",
44 "Funding Agency",
45 "Funding Agency Code",
46 "Funding Sub Agency",
47 "Funding Sub Agency Code",
48 "Place of Performance City Code",
49 "Place of Performance State Code",
50 "Place of Performance Country Code",
51 "Place of Performance Zip5",
52 "Description",
53 "Last Modified Date",
54 "Base Obligation Date",
55 "prime_award_recipient_id",
56 "generated_internal_id",
57 "def_codes",
58 "COVID-19 Obligations",
59 "COVID-19 Outlays",
60 "Infrastructure Obligations",
61 "Infrastructure Outlays",
62 "Primary Place of Performance",
63 ]
65 def __init__(
66 self, data_or_id: Dict[str, Any] | str, client: USASpending
67 ):
68 """Initialize Award instance.
70 Args:
71 data_or_id: Award data dictionary or unique award ID string
72 client: Optional USASpending client instance
73 """
74 if isinstance(data_or_id, dict):
75 raw = data_or_id.copy()
76 elif isinstance(data_or_id, str):
77 raw = {"generated_unique_award_id": data_or_id}
78 else:
79 raise ValidationError("Award expects a dict or an award_id string")
80 super().__init__(raw, client)
82 def _fetch_details(self) -> Optional[Dict[str, Any]]:
83 """Fetch full award details from the awards resource."""
84 award_id = self.generated_unique_award_id
85 if not award_id:
86 raise ValidationError(
87 "Cannot lazy-load Award data. Property `generated_unique_award_id` is required to fetch details."
88 )
89 try:
90 # Use the awards resource to get full award data
91 full_award = self._client.awards.find_by_generated_id(award_id)
92 full_data = full_award.raw
94 # If we're a base Award class and now have type information,
95 # convert to appropriate subclass
96 if full_data and self.__class__ == Award:
97 from .award_factory import create_award
99 new_instance = create_award(full_data, self._client)
100 if new_instance.__class__ != Award:
101 # Copy state from new instance to self
102 self.__class__ = new_instance.__class__
103 # Merge the data
104 self._data.update(full_data)
105 return full_data
107 return full_data
108 except Exception:
109 logger.error(
110 f"Failed to fetch full details for Award ID {award_id}. "
111 "Check if the ID is valid and the client is configured correctly."
112 )
113 raise
115 # Core Award properties
116 @property
117 def id(self) -> Optional[int]:
118 """Internal USASpending database ID for this award."""
119 return self._lazy_get("id","internal_id")
121 @property
122 def generated_unique_award_id(self) -> Optional[str]:
123 """USASpending-generated unique award identifier."""
124 # This cannot be lazy-loaded since it's required to fetch details
125 return self.get_value(["generated_unique_award_id", "generated_internal_id"])
127 @property
128 def award_identifier(self) -> str:
129 """General-purpose award identifier, type-agnostic.
131 Award-type specific values are implemented in subclasses.
133 Returns:
134 (PIID, FAIN, URI): str
135 """
136 return str(self._lazy_get("Award ID", "piid", "fain", "uri", default=""))
138 @property
139 def category(self) -> str:
140 """Plain english description of the award type.
142 Returns:
143 One of "contract", "grant", "idv", "loan", or "other" if unknown
144 """
145 return self._lazy_get("category", default="")
147 @property
148 def type(self) -> Optional[str]:
149 """
150 The subtype award code (e.g. "A", "B", "C", etc. for contracts.
151 See the Config file for mappings.
152 """
153 return self._lazy_get("type", default="")
155 @property
156 def award_type_code(self) -> Optional[str]:
157 """ More expressive property name for `type` to avoid confusion with Python built-in."""
158 return self.type
160 @property
161 def type_description(self) -> Optional[str]:
162 """The plain text description of the type of the award"""
163 return self._lazy_get("type_description", "Contract Award Type", "Award Type", default="")
165 @property
166 def description(self) -> str:
167 """
168 A brief, plain English summary of the award.
169 """
170 desc = self._lazy_get("description", "Description")
171 if isinstance(desc, str):
172 return smart_sentence_case(desc)
173 return ""
175 @property
176 def total_obligation(self) -> float:
177 """The amount of money the government is obligated to pay for the award
179 This is a system generated element providing the sum of all the amounts
180 entered in the "Action Obligation" field for a particular PIID and Agency.
182 Example: Contract has 9 Modifications under "Transaction Number" as '1'
183 and 9 modifications with the same PIID under "Transaction Number" as '2'.
184 The base contracts and all the modifications have "Action Obligation" as $10
185 each. The value for the field "Total Obligated Amount" when the either of
186 the bases or the modification is retrieved through atom feeds will be $200
187 ($100 under Transaction Number 1 + $100 under Transaction Number 2).
188 "Total Obligated Amount" is generated irrespective of the "Transaction Number"
189 on the Awards.
191 """
192 return to_float(self._lazy_get("total_obligation", "Award Amount")) or 0.0
194 @property
195 def subaward_count(self) -> int:
196 """The number of subawards associated with this award."""
197 return int(self._lazy_get("subaward_count", default=0))
199 @property
200 def total_subaward_amount(self) -> Optional[float]:
201 """The total amount of subawards for this award."""
202 return to_float(self._lazy_get("total_subaward_amount", default=None))
204 @property
205 def date_signed(self) -> Optional[datetime]:
206 """The date the award was signed by the Government or a binding agreement was reached."""
207 return to_date(self._lazy_get("date_signed", "Base Obligation Date", default=None))
209 @property
210 def base_obligation_date(self) -> Optional[datetime]:
211 return self.date_signed
213 @property
214 def total_account_outlay(self) -> Optional[float]:
215 """The total amount of money that has been paid out for the award from the associated federal accounts"""
216 return to_float(
217 self._lazy_get("total_account_outlay", default=None)
218 )
220 @property
221 def total_account_obligation(self) -> Optional[float]:
222 """Total amount obligated for this award."""
223 return to_float(self._lazy_get("total_account_obligation", default=None))
225 @property
226 def total_outlay(self) -> Optional[float]:
227 return self._lazy_get("total_outlay", "Total Outlays", default=None)
229 @property
230 def total_account_obligation(self) -> Optional[float]:
231 """Total amount obligated for this award."""
232 return to_float(self._lazy_get("total_account_obligation", default=None))
234 @property
235 def account_outlays_by_defc(self) -> List[Dict[str, Any]]:
236 """Outlays broken down by Disaster Emergency Fund Code (DEFC)."""
237 return self._lazy_get("account_outlays_by_defc", default=[])
239 @property
240 def account_obligations_by_defc(self) -> List[Dict[str, Any]]:
241 """Obligations broken down by Disaster Emergency Fund Code (DEFC)."""
242 return self._lazy_get("account_obligations_by_defc", default=[])
244 @cached_property
245 def parent_award(self) -> Optional[Award]:
246 """Reference to parent award for child awards."""
247 data = self._lazy_get("parent_award")
248 from .award_factory import create_award
249 return create_award(data, self._client) if data else None
251 @cached_property
252 def executive_details(self) -> Optional[Dict[str, Any]]:
253 """Executive compensation details for the award recipient."""
254 return self._lazy_get("executive_details")
256 @property
257 def recipient_uei(self) -> Optional[str]:
258 """Recipient Unique Entity Identifier (UEI)."""
259 uei = self._lazy_get("recipient_uei", "Recipient UEI")
260 if not uei:
261 # Try nested recipient object if available
262 if self.recipient and self.recipient.uei:
263 uei = self.recipient.uei
264 return uei
266 @property
267 def covid19_obligations(self) -> float:
268 """COVID-19 related obligations amount."""
269 return to_float(
270 self._lazy_get("covid19_obligations", "COVID-19 Obligations", default=0)
271 )
273 @property
274 def covid19_outlays(self) -> float:
275 """COVID-19 related outlays amount."""
276 return to_float(
277 self._lazy_get("covid19_outlays", "COVID-19 Outlays", default=0)
278 )
280 @property
281 def infrastructure_obligations(self) -> float:
282 """Infrastructure related obligations amount."""
283 return to_float(
284 self._lazy_get(
285 "infrastructure_obligations", "Infrastructure Obligations", default=0
286 )
287 )
289 @property
290 def infrastructure_outlays(self) -> float:
291 """Infrastructure related outlays amount."""
292 return to_float(
293 self._lazy_get(
294 "infrastructure_outlays", "Infrastructure Outlays", default=0
295 )
296 )
299 # Helper properties properties. These often map to field names returned by
300 # the spending_by_award/Award Search results, or provide general access methods
301 # that are common across award types.
303 @property
304 def award_amount(self) -> float:
305 """General helper total obligated/loaned amount."""
306 return to_float(self._lazy_get("Award Amount", "Loan Amount", "total_obligation","total_funding")) or 0.0
308 @property
309 def start_date(self) -> Optional[datetime]:
310 start_date = self.get_value(["Start Date", "Base Obligation Date", "Period of Performance Start Date"])
311 if not start_date:
312 if self.period_of_performance and self.period_of_performance.start_date:
313 start_date = self.period_of_performance.start_date
314 return start_date
316 @property
317 def end_date(self) -> Optional[datetime]:
318 end_date = self.get_value(["End Date", "Period of Performance End Date"])
319 if not end_date:
320 if self.period_of_performance and self.period_of_performance.end_date:
321 end_date = self.period_of_performance.end_date
322 return end_date
324 @property
325 def usa_spending_url(self) -> str:
326 """Return the USASpending.gov public URL for this award."""
327 award_id = self.generated_unique_award_id
328 if award_id and isinstance(award_id, str):
329 return f"https://www.usaspending.gov/award/{award_id}/"
330 else:
331 return ""
333 # Properties that return complex objects and related award data
334 #
335 # Currently implemented are:
336 #
337 # Belongs To (one-to-one relationships):
338 # - parent_award (Award object: parent award if this is a child award)
339 # - recipient (Recipient object: details about the award recipient)
340 # - funding_agency (Agency object: details about the funding agency)
341 # - awarding_agency (Agency object: details about the awarding agency)
342 # - funding_subtier_agency (SubTierAgency object: details about the funding subtier agency)
343 # - awarding_subtier_agency (SubTierAgency object: details about the awarding subtier agency)
344 #
345 # Has One (one-to-one relationships):
346 # - period_of_performance (PlaceOfPerformance object: Start and End dates for award)
347 # - place_of_performance (Location object: location where the work is performed)
348 #
349 # Has Many (one-to-many relationships):
350 # - transactions (TransactionsSearch object: query builder for transactions associated with the award)
351 # - funding (FundingSearch object: query builder for treasury funding records (outlay and obligation) associated with the award)
352 # - subawards (SubAwardsSearch object: query builder for subawards associated with the award)
355 @cached_property
356 def period_of_performance(self) -> Optional[PeriodOfPerformance]:
357 """Award period of performance dates."""
358 if "period_of_performance" in self.raw and isinstance(self.raw.get("period_of_performance"), dict):
359 return PeriodOfPerformance(self.raw.get("period_of_performance"))
361 # Award search results return Period of Performance information in a flat structure
362 # We need to assign these values to a PeriodOfPerformance object
363 # to maintain consistency.
364 date_keys = ["Start Date", "End Date", "Last Modified Date"]
365 if any(k in self._data for k in date_keys):
366 return PeriodOfPerformance(
367 {
368 "start_date": self.get_value(["Start Date", "Base Obligation Date", "Period of Performance Start Date"]),
369 "end_date": self.get_value(["End Date", "Period of Performance Current End Date"]),
370 "last_modified_date": self.get_value("Last Modified Date"),
371 }
372 )
374 # If no data, trigger fetch
375 self._ensure_details()
376 return PeriodOfPerformance(self.get_value("period_of_performance"))
379 @cached_property
380 def place_of_performance(self) -> Optional[Location]:
381 """Award place of performance location."""
382 data = self._lazy_get(
383 "place_of_performance", "Primary Place of Performance", default=None
384 )
385 if not isinstance(data, dict) or not data:
386 return None
388 # Check if all values in the dict are None/null (common for IDV awards)
389 if all(v is None for v in data.values()):
390 return None
392 return Location(data, self._client)
394 @cached_property
395 def recipient(self) -> Optional[Recipient]:
396 """Award recipient with lazy loading."""
397 # First check if we already have a nested recipient object
398 if "recipient" in self._data and isinstance(self._data["recipient"], dict):
399 return Recipient(self._data["recipient"], self._client)
401 # Then, check for flat recipient fields from search results
402 recipient_keys = ["Recipient Name", "recipient_id", "Recipient Location"]
403 if any(key in self._data for key in recipient_keys):
404 recipient_data = {
405 "recipient_name": self._data.get("Recipient Name"),
406 "recipient_unique_id": self._data.get("Recipient DUNS Number"),
407 "recipient_id": self._data.get("recipient_id"),
408 "recipient_hash": self._data.get("recipient_hash"),
409 "recipient_uei": self._data.get("Recipient UEI"),
410 }
411 recipient = Recipient(recipient_data, self._client)
412 if "Recipient Location" in self._data and isinstance(self._data["Recipient Location"], dict):
413 recipient.location = Location(self._data["Recipient Location"], self._client)
414 return recipient
416 # If no recipient data is available locally, trigger a fetch
417 self._ensure_details()
418 if "recipient" in self._data and isinstance(self._data["recipient"], dict):
419 return Recipient(self._data["recipient"], self._client)
421 return None
423 def _load_agency_data(self, agency_type: str) -> Optional[Dict[str, Any]]:
424 """Load agency data from either nested or flat structure.
426 Args:
427 agency_type: Either "funding" or "awarding"
429 Returns:
430 Processed agency data dict or None if not available
431 """
432 if agency_type not in ["funding", "awarding"]:
433 raise ValueError(f"Invalid agency_type: {agency_type}")
435 # Define field mappings based on agency type
436 if agency_type == "funding":
437 nested_key = "funding_agency"
438 flat_keys = ["Funding Agency", "Funding Agency Code",
439 "Funding Sub Agency", "Funding Sub Agency Code"]
440 name_key = "Funding Agency"
441 code_key = "Funding Agency Code"
442 sub_name_key = "Funding Sub Agency"
443 sub_code_key = "Funding Sub Agency Code"
444 # No funding_agency_id available in search results
445 id_key = None
446 else: # awarding
447 nested_key = "awarding_agency"
448 flat_keys = ["Awarding Agency", "Awarding Agency Code",
449 "Awarding Sub Agency", "Awarding Sub Agency Code"]
450 name_key = "Awarding Agency"
451 code_key = "Awarding Agency Code"
452 sub_name_key = "Awarding Sub Agency"
453 sub_code_key = "Awarding Sub Agency Code"
454 id_key = "awarding_agency_id"
456 # First check if we have nested agency data (from full award details)
457 if self.raw.get(nested_key):
458 return self.raw.get(nested_key)
460 # Then check for flat agency fields (from search results)
461 if any(key in self.raw for key in flat_keys):
462 data = {
463 "toptier_agency": {
464 "name": self.raw.get(name_key),
465 "code": self.raw.get(code_key), # Agency code
466 "abbreviation": self.raw.get(code_key),
467 },
468 "subtier_agency": {
469 "name": self.raw.get(sub_name_key),
470 "code": self.raw.get(sub_code_key), # Subtier code
471 "abbreviation": self.raw.get(sub_code_key),
472 },
473 "id": self.raw.get(id_key) if id_key else None,
474 "has_agency_page": False, # Not available in search results
475 "office_agency_name": None, # Not available in search results
476 }
477 return data
479 # Finally try lazy loading
480 return self._lazy_get(nested_key)
482 @cached_property
483 def funding_agency(self) -> Optional[Agency]:
484 """Funding agency information."""
485 data = self._load_agency_data("funding")
487 if not data:
488 return None
490 # Extract toptier data and merge with top-level agency fields
491 toptier_data = data.get("toptier_agency", {})
492 agency_data = {
493 "agency_id": data.get("id"),
494 "has_agency_page": data.get("has_agency_page"),
495 "office_agency_name": data.get("office_agency_name"),
496 **toptier_data # Merge toptier fields (name, code, abbreviation, slug)
497 }
499 subtier_data = data.get("subtier_agency")
500 return Agency(agency_data, self._client, subtier_data)
502 @cached_property
503 def awarding_agency(self) -> Optional[Agency]:
504 """Awarding agency information."""
505 data = self._load_agency_data("awarding")
507 if not data:
508 return None
510 # Extract toptier data and merge with top-level agency fields
511 toptier_data = data.get("toptier_agency", {})
512 agency_data = {
513 "agency_id": data.get("id"),
514 "has_agency_page": data.get("has_agency_page"),
515 "office_agency_name": data.get("office_agency_name"),
516 **toptier_data # Merge toptier fields (name, code, abbreviation, slug)
517 }
519 subtier_data = data.get("subtier_agency")
520 return Agency(agency_data, self._client, subtier_data)
522 @cached_property
523 def funding_subtier_agency(self) -> Optional["SubTierAgency"]:
524 """Funding subtier agency information."""
525 data = self._load_agency_data("funding")
527 if not data:
528 return None
530 subtier_data = data.get("subtier_agency")
531 if not subtier_data:
532 return None
534 # Create a copy and add office_agency_name if available
535 enhanced_subtier_data = subtier_data.copy()
536 office_name = data.get("office_agency_name")
537 if office_name:
538 enhanced_subtier_data["office_agency_name"] = office_name
540 from .subtier_agency import SubTierAgency
541 return SubTierAgency(enhanced_subtier_data, self._client)
543 @cached_property
544 def awarding_subtier_agency(self) -> Optional["SubTierAgency"]:
545 """Awarding subtier agency information."""
546 data = self._load_agency_data("awarding")
548 if not data:
549 return None
551 subtier_data = data.get("subtier_agency")
552 if not subtier_data:
553 return None
555 # Create a copy and add office_agency_name if available
556 enhanced_subtier_data = subtier_data.copy()
557 office_name = data.get("office_agency_name")
558 if office_name:
559 enhanced_subtier_data["office_agency_name"] = office_name
561 from .subtier_agency import SubTierAgency
562 return SubTierAgency(enhanced_subtier_data, self._client)
564 @property
565 def transactions(self) -> "TransactionsSearch":
566 """Get transactions query builder for this award.
568 Returns a TransactionsSearch object that can be further filtered and chained.
570 Examples:
571 >>> award.transactions.count() # Get count without loading all data
572 >>> award.transactions.limit(10).all() # Get first 10 transactions
573 >>> list(award.transactions) # Iterate through all transactions
574 """
575 return self._client.transactions.for_award(self.generated_unique_award_id)
577 @property
578 def funding(self) -> "FundingSearch":
579 """Get funding query builder for this award.
581 Returns a FundingSearch object that can be further filtered and chained.
583 Examples:
584 >>> award.funding.count() # Get count without loading all data
585 >>> award.funding.order_by("fiscal_date", "asc").all() # Get all funding records sorted by date
586 >>> list(award.funding.limit(10)) # Iterate through first 10 funding records
587 """
588 return self._client.funding.for_award(self.generated_unique_award_id)
590 @property
591 def subawards(self) -> "SubAwardsSearch":
592 """Get subawards query builder for this award.
594 Returns a SubAwardsSearch object that can be further filtered and chained.
595 Implemented in subclasses
596 """
597 raise NotImplementedError()
600 # Downloading detailed award data
601 @property
602 def _download_type(self) -> Optional[AwardType]:
603 """
604 The type required by the download API ('contract' or 'assistance').
605 This must be implemented by model subclasses
606 """
607 from .contract import Contract
608 from .grant import Grant
609 from .idv import IDV
611 if isinstance(self, Contract):
612 return "contract"
613 elif isinstance(self, Grant):
614 return "assistance"
615 elif isinstance(self, IDV):
616 return "idv"
617 else:
618 raise(NotImplementedError)
621 def download(self, file_format: FileFormat = "csv", destination_dir: Optional[str] = None) -> "DownloadJob":
622 """
623 Queue a download job for this award's detailed data.
625 This utilizes the USASpending bulk download API, which queues the request
626 and processes it asynchronously.
628 Args:
629 file_format: The format of the file(s) in the zip file containing the data
630 destination_dir: Directory where the file will be saved (defaults to CWD).
632 Returns:
633 A DownloadJob object. Use job.wait_for_completion() to block until finished.
635 Raises:
636 ConfigurationError: If the Award instance lacks a client reference.
637 ValidationError: If the award ID or download type is missing/invalid.
639 Example:
640 >>> contract = client.awards.find_by_generated_id("CONT_AWD_123...")
641 >>> job = contract.download(destination_dir="./data")
642 >>> print(f"Job queued: {job.file_name}. Waiting...")
643 >>> extracted_files = job.wait_for_completion(timeout=600)
644 >>> print(f"Download complete. Files: {extracted_files}")
645 """
647 award_id = self.generated_unique_award_id
649 if not award_id:
650 # If we don't have an award ID, we cannot proceed
651 raise ValidationError("Cannot download award data without a 'generated_unique_award_id'. Ensure the award object is fully loaded.")
653 download_type = self._download_type
655 if not download_type:
656 # Safety check in case a subclass doesn't implement _download_type or the implementation returns None
657 raise ValidationError(f"Download is not supported or implemented for award type: {self.__class__.__name__}.")
659 # Access the DownloadManager via the client's download resource.
660 # We route the call through the appropriate method on the resource.
661 if download_type == "contract":
662 return self._client.downloads.contract(award_id, file_format, destination_dir)
663 elif download_type == "assistance":
664 return self._client.downloads.assistance(award_id, file_format, destination_dir)
665 elif download_type == "idv":
666 return self._client.downloads.idv(award_id, file_format, destination_dir)
667 else:
668 raise NotImplementedError
670 def __repr__(self) -> str:
671 """String representation of Award."""
672 recipient_name = self.recipient.name if self.recipient else "?"
673 award_id = self.award_identifier or self.generated_unique_award_id or "?"
674 return f"<Award {award_id} → {recipient_name}>"