Coverage for src/usaspending/models/recipient.py: 100%
95 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 17:15 -0700
1from __future__ import annotations
2from typing import Dict, Any, Optional, List, TYPE_CHECKING
3from functools import cached_property
4import re
6from .lazy_record import LazyRecord
7from .location import Location
8from ..utils.formatter import to_float, contracts_titlecase
10from ..exceptions import ValidationError
11from ..logging_config import USASpendingLogger
13logger = USASpendingLogger.get_logger(__name__)
15if TYPE_CHECKING:
16 from ..client import USASpending
18# FUTURE: Add logic to self-categorize recipient type based on FPDS categories
19# This would enhance the Recipient model by automatically determining the recipient
20# type (corporation, university, government, etc.) based on FPDS category codes
23class Recipient(LazyRecord):
24 # compiled once at import time
25 _LIST_SUFFIX_RE = re.compile(
26 r"""
27 ^(?P<base>.+?) # everything before the dash (non-greedy)
28 -\[\s*(?P<body>[^\]]+)\] # -[ ... ]
29 $ # end of string
30 """,
31 re.VERBOSE,
32 )
34 def __init__(
35 self, data_or_id: Dict[str, Any] | str, client: Optional[USASpending] = None
36 ):
37 if isinstance(data_or_id, dict):
38 raw = data_or_id.copy()
39 rid = raw.get("recipient_id") or raw.get("recipient_hash")
40 if rid:
41 raw["recipient_id"] = self._clean_recipient_id(rid)
42 elif isinstance(data_or_id, str):
43 raw = {"recipient_id": self._clean_recipient_id(data_or_id)}
44 else:
45 raise ValidationError("Recipient expects dict or recipient_id/hash string")
46 super().__init__(raw, client)
48 def _fetch_details(self) -> Optional[Dict[str, Any]]:
49 """Fetch full recipient details from the API."""
50 recipient_id = self.recipient_id
51 if not recipient_id:
52 logger.error(
53 "Cannot lazy-load Recipient data. Property `recipient_id` is required to fetch details."
54 )
55 return None
56 try:
57 # Make direct API call to avoid circular dependency
58 endpoint = f"/recipient/{recipient_id}/"
59 response = self._client._make_request("GET", endpoint)
60 return response
61 except Exception as e:
62 # If fetch fails, return None to avoid breaking the application
63 logger.error(f"Failed to fetch recipient details for {recipient_id}: {e}")
64 return None
66 @staticmethod
67 def _clean_recipient_id(rid: str) -> str:
68 """
69 Normalise list-annotated recipient IDs.
70 Sometimes these look like "abc123-['C','R']".
71 This will select the first letter after the dash,
72 """
73 if not isinstance(rid, str):
74 return rid # defensive; shouldn't happen
76 rid = rid.strip().rstrip("/") # drop accidental trailing slash
78 m = Recipient._LIST_SUFFIX_RE.match(rid)
79 if not m:
80 return rid # already in normal form
82 base = m.group("base")
83 body = m.group("body")
85 # turn "'C','R'" or "'R'" etc. into a list of clean tokens
86 tokens = [
87 tok.strip().strip("'\"").upper() for tok in body.split(",") if tok.strip()
88 ]
90 letter = tokens[0]
91 return f"{base}-{letter}" if letter else base
93 @property
94 def recipient_id(self) -> Optional[str]:
95 return self.get_value(["recipient_id", "recipient_hash"], default=None)
97 @property
98 def name(self) -> Optional[str]:
99 return contracts_titlecase(
100 self._lazy_get("name", "recipient_name", "Recipient Name", default=None)
101 )
103 @property
104 def duns(self) -> Optional[str]:
105 return self._lazy_get(
106 "duns", "recipient_unique_id", "Recipient DUNS Number", default=None
107 )
109 @property
110 def uei(self) -> Optional[str]:
111 return self._lazy_get("uei", "recipient_uei")
113 @cached_property
114 def parent(self) -> Optional["Recipient"]:
115 pid = self._lazy_get("parent_id")
117 # Don't load a parent if parent id is missing or
118 # the parent recipient_id is theh same as the current one
119 if not pid or pid == self.recipient_id:
120 return None
121 else:
122 return Recipient(
123 {
124 "recipient_id": pid,
125 "name": self.get_value("parent_name"),
126 "duns": self.get_value("parent_duns"),
127 "uei": self.get_value("parent_uei"),
128 },
129 client=self._client,
130 )
132 @cached_property
133 def parents(self) -> List["Recipient"]:
134 plist = []
135 for p in self.get_value("parents", default=[]):
136 if isinstance(p, dict):
137 # Skip if parent_id is missing or the same as current recipient_id
138 if not p.get("parent_id") or p.get("parent_id") == self.recipient_id:
139 continue
140 plist.append(
141 Recipient(
142 {
143 "recipient_id": p.get("parent_id"),
144 "name": p.get("parent_name"),
145 "duns": p.get("parent_duns"),
146 "uei": p.get("parent_uei"),
147 },
148 client=self._client,
149 )
150 )
151 return plist
153 @property
154 def business_types(self) -> List[str]:
155 return self._lazy_get("business_types", "business_categories", default=[])
157 @cached_property
158 def location(self) -> Optional[Location]:
159 """Get recipient location - shares same client."""
160 data = self._lazy_get("location")
161 return Location(data, self._client) if data else None
163 @property
164 def total_transaction_amount(self):
165 return to_float(self._lazy_get("total_transaction_amount"))
167 @property
168 def total_transactions(self):
169 return self._lazy_get("total_transactions")
171 @property
172 def total_face_value_loan_amount(self):
173 return to_float(self._lazy_get("total_face_value_loan_amount"))
175 @property
176 def total_face_value_loan_transactions(self):
177 return self._lazy_get("total_face_value_loan_transactions")
179 def __repr__(self) -> str:
180 return f"<Recipient {self.name or '?'} ({self.recipient_id})>"