Dedup connectors and cut pipeline compute time

Deduplication:
- Extract shared USGS retry/backoff and truncation checks into a composed
  USGSRequester (dependency-injected http client + warn), removing two
  near-identical inline retry blocks in the USGS connector.
- Add a default __repr__ to BaseSource and remove 21 trivial per-connector
  overrides that only returned the class name.

Performance:
- Fetch a source's site chunks concurrently in _site_wrapper via a
  ThreadPoolExecutor, processing results in chunk order so output,
  site_limit, and rollback stay deterministic. Gated by the new
  config.fetch_workers knob (default 4; set 1 for legacy serial).
- Tune httpx.Client connection pooling (keepalive reuse).
- Stop sleeping after the final retry attempt in the base and USGS retry
  loops; switch USGS from linear to capped exponential backoff.
- Fail fast on non-retryable client errors (4xx) instead of burning the
  full backoff schedule.

All 360 tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: jirhiker

backend/config.py

@@ -152,6 +152,12 @@ class Config:
     site_limit: int = 0
     dry: bool = False
 
+    # Number of chunks fetched concurrently per source (network-bound I/O).
+    # 1 = serial (legacy behavior). Higher values speed up multi-chunk sources
+    # but issue more simultaneous requests, which can trip per-source API rate
+    # limits (e.g. USGS) — tune down if you see 429s.
+    fetch_workers: int = 4
+
     # date
     start_date: str = ""
     end_date: str = ""

backend/connectors/bor/source.py

@@ -38,9 +38,6 @@ class BORSiteSource(BaseSiteSource):
     def __init__(self):
         super().__init__(transformer=BORSiteTransformer())
 
-    def __repr__(self):
-        return "BORSiteSource"
-
     def health(self):
         try:
             resp = self.get_records()
@@ -66,9 +63,6 @@ def __init__(self):
         super().__init__(transformer=BORAnalyteTransformer())
     _source_parameter_name = None
 
-    def __repr__(self):
-        return "BORAnalyteSource"
-
     def _extract_parameter_record(self, record):
         record[PARAMETER_NAME] = self.config.parameter
         record[PARAMETER_VALUE] = record["attributes"]["result"]

backend/connectors/ckan/source.py

@@ -98,9 +98,6 @@ def __init__(self, resource_id, **kw):
         elif resource_id == ROSWELL_RESOURCE_ID:
             self.bounding_polygon = OSE_ROSWELL_ROSWELL_BOUNDING_POLYGON
 
-    def __repr__(self):
-        return "NMOSERoswellSiteSource"
-
     def health(self):
         params = self._get_params()
         params["limit"] = 1
@@ -123,9 +120,6 @@ def __init__(self, resource_id=None, **kw):
         kw.setdefault("transformer", OSERoswellWaterLevelTransformer())
         super().__init__(resource_id, **kw)
 
-    def __repr__(self):
-        return "NMOSERoswellWaterLevelSource"
-
     def get_records(self, site_record):
         return self._parse_response(site_record, self.get_response())
 

backend/connectors/isc_seven_rivers/source.py

@@ -69,9 +69,6 @@ class ISCSevenRiversSiteSource(BaseSiteSource):
     def __init__(self):
         super().__init__(transformer=ISCSevenRiversSiteTransformer())
 
-    def __repr__(self):
-        return "ISCSevenRiversSiteSource"
-
     def health(self):
         try:
             resp = self.get_records()
@@ -94,9 +91,6 @@ class ISCSevenRiversAnalyteSource(BaseAnalyteSource):
     def __init__(self):
         super().__init__(transformer=ISCSevenRiversAnalyteTransformer())
 
-    def __repr__(self):
-        return "ISCSevenRiversAnalyteSource"
-
     def _get_analyte_id_and_name(self, analyte):
         """ """
         if self._analyte_ids is None:

backend/connectors/nmbgmr/source.py

@@ -64,9 +64,6 @@ class NMBGMRSiteSource(BaseSiteSource):
     def __init__(self):
         super().__init__(transformer=NMBGMRSiteTransformer())
 
-    def __repr__(self):
-        return "NMBGMRSiteSource"
-
     def health(self):
         try:
             resp = self._execute_json_request(
@@ -122,9 +119,6 @@ class NMBGMRAnalyteSource(BaseAnalyteSource):
     def __init__(self):
         super().__init__(transformer=NMBGMRAnalyteTransformer())
 
-    def __repr__(self):
-        return "NMBGMRAnalyteSource"
-
     def get_records(self, site_record):
         analyte = get_analyte_search_param(
             self.config.parameter, NMBGMR_ANALYTE_MAPPING
@@ -185,9 +179,6 @@ class NMBGMRWaterLevelSource(BaseWaterLevelSource):
     def __init__(self):
         super().__init__(transformer=NMBGMRWaterLevelTransformer())
 
-    def __repr__(self):
-        return "NMBGMRWaterLevelSource"
-
     def _clean_records(self, records):
         # remove records with no depth to water value
         return [

backend/connectors/nmenv/source.py

@@ -34,17 +34,13 @@
 URL = "https://nmenv.newmexicowaterdata.org/FROST-Server/v1.1/"
 
 
-
 class DWBSiteSource(STSiteSource):
     url = URL
     bounding_polygon = NM_STATE_BOUNDING_POLYGON
 
     def __init__(self):
         super().__init__(transformer=DWBSiteTransformer())
 
-    def __repr__(self):
-        return "DWBSiteSource"
-
     def health(self):
         try:
             resp = self.get_records(top=10, analyte=TDS)
@@ -115,9 +111,6 @@ class DWBAnalyteSource(STAnalyteSource):
     def __init__(self):
         super().__init__(transformer=DWBAnalyteTransformer())
 
-    def __repr__(self):
-        return "DWBAnalyteSource"
-
     def _parse_result(
         self, result, result_dt=None, result_id=None, result_location=None
     ):

backend/connectors/st2/source.py

@@ -184,48 +184,33 @@ class NMOSERoswellWaterLevelSource(ST2WaterLevelSource):
     def __init__(self):
         super().__init__(transformer=NMOSERoswellWaterLevelTransformer())
 
-    def __repr__(self):
-        return "NMOSERoswellWaterLevelSource"
-
 
 class PVACDWaterLevelSource(ST2WaterLevelSource):
     agency = "PVACD"
 
     def __init__(self):
         super().__init__(transformer=PVACDWaterLevelTransformer())
 
-    def __repr__(self):
-        return "PVACDWaterLevelSource"
-
 
 class EBIDWaterLevelSource(ST2WaterLevelSource):
     agency = "EBID"
 
     def __init__(self):
         super().__init__(transformer=EBIDWaterLevelTransformer())
 
-    def __repr__(self):
-        return "EBIDWaterLevelSource"
-
 
 class BernCoWaterLevelSource(ST2WaterLevelSource):
     agency = "BernCo"
 
     def __init__(self):
         super().__init__(transformer=BernCoWaterLevelTransformer())
 
-    def __repr__(self):
-        return "BernCoWaterLevelSource"
-
 
 class CABQWaterLevelSource(ST2WaterLevelSource):
     agency = "CABQ"
 
     def __init__(self):
         super().__init__(transformer=CABQWaterLevelTransformer())
 
-    def __repr__(self):
-        return "CABQWaterLevelSource"
-
 
 # ============= EOF =============================================