Python Equivalents to R Vignette Examples
[1]:
from dataretrieval import nwis, waterdata, wqp
The dataretrieval Python package was created as an equivalent to the R dataRetrieval package.
The following shows Python equivalents for the methods outlined in the R dataRetrieval vignette, with the equivalent R code in comments.
[2]:
"""
{r getSite, echo=TRUE, eval=FALSE}
siteNumbers <- c("01491000","01645000")
siteINFO <- readNWISsite(siteNumbers)
"""
siteNumbers = ["USGS-01491000", "USGS-01645000"]
siteINFO, md = waterdata.get_monitoring_locations(
monitoring_location_id=siteNumbers, skip_geometry=True
)
Retrieving: monitoring-locations · 1 page · 2 rows
No API key detected — register for higher rate limits at https://api.waterdata.usgs.gov/signup/
[3]:
"""
# Continuing from the previous example:
# This pulls out just the daily, mean data:
dailyDataAvailable <- whatNWISdata(siteNumbers,
service="dv", statCd="00003")
"""
dailyDataAvailable, md = waterdata.get_time_series_metadata(
monitoring_location_id=siteNumbers, statistic_id="00003", skip_geometry=True
)
Retrieving: time-series-metadata · 1 page · 11 rows
[4]:
"""
# Choptank River near Greensboro, MD:
siteNumber <- "01491000"
parameterCd <- "00060" # Discharge
startDate <- "2009-10-01"
endDate <- "2012-09-30"
discharge <- readNWISdv(siteNumber, parameterCd, startDate, endDate)
"""
# Choptank River near Greensboro, MD:
siteNumber = "USGS-01491000"
parameterCd = "00060" # Discharge
discharge, md = waterdata.get_daily(
monitoring_location_id=siteNumber,
parameter_code=parameterCd,
time="2009-10-01/2012-09-30",
)
Retrieving: daily · 1 page · 1,096 rows
[5]:
"""
siteNumber <- "01491000"
parameterCd <- c("00010","00060") # Temperature and discharge
statCd <- c("00001","00003") # Maximum and mean
startDate <- "2012-01-01"
endDate <- "2012-05-01"
temperatureAndFlow <- readNWISdv(siteNumber, parameterCd, startDate, endDate, statCd=statCd)
"""
siteNumber = "USGS-01491000"
parameterCd = ["00010", "00060"] # Temperature and discharge
statisticId = ["00001", "00003"] # Maximum and mean
temperatureAndFlow, md = waterdata.get_daily(
monitoring_location_id=siteNumber,
parameter_code=parameterCd,
statistic_id=statisticId,
time="2012-01-01/2012-05-01",
)
---------------------------------------------------------------------------
RateLimited Traceback (most recent call last)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:615, in FanOut._run.<locals>.track(index, item)
614 """One chunk (with retry) + result-store + progress update."""
--> 615 result = await _retry(
616 lambda: self.fetch(item), self.retry_policy, gate=semaphore
617 )
618 self._chunks[index] = result
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:291, in retry_async(afn, policy, gate)
290 try:
--> 291 return await attempt_once()
292 except Exception as exc:
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:287, in retry_async.<locals>.attempt_once()
286 credit_wait(time.monotonic() - started)
--> 287 return await afn()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:416, in _fetch_once(args, build_request, geopd, include_geometry, row_cap)
415 req = build_request(**args)
--> 416 return await _walk_pages(
417 geopd=geopd,
418 req=req,
419 include_geometry=include_geometry,
420 row_cap=row_cap,
421 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:203, in _walk_pages(geopd, req, client, include_geometry, row_cap)
201 return await sess.request(method, cursor, headers=headers, content=content)
--> 203 return await paginate(
204 req,
205 parse_response=functools.partial(
206 _ogc_parse_response,
207 geopd=geopd,
208 include_geometry=include_geometry,
209 ),
210 follow_up=follow_up,
211 client=client,
212 raise_for_status=_raise_for_non_200,
213 row_cap=row_cap,
214 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/pagination.py:112, in paginate(initial_req, parse_response, follow_up, raise_for_status, client, row_cap)
111 response = await session.send(initial_req)
--> 112 raise_for_status(response)
113 initial_response = response
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/errors.py:139, in _raise_for_non_200(resp)
138 return
--> 139 raise error_for_status(
140 status,
141 _error_body(resp) + _url_suffix(resp),
142 retry_after=_parse_retry_after(resp.headers.get("Retry-After")),
143 )
RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/ogcapi/v0/collections/daily/items?monitoring_location_id=USGS-01491000¶meter_code=00010%2C00060&statistic_id=00001%2C00003&time=2012-01-01%2F2012-05-01&limit=50000)
The above exception was the direct cause of the following exception:
QuotaExhausted Traceback (most recent call last)
Cell In[5], line 14
10 siteNumber = "USGS-01491000"
11 parameterCd = ["00010", "00060"] # Temperature and discharge
12 statisticId = ["00001", "00003"] # Maximum and mean
13
---> 14 temperatureAndFlow, md = waterdata.get_daily(
15 monitoring_location_id=siteNumber,
16 parameter_code=parameterCd,
17 statistic_id=statisticId,
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/time_series.py:257, in get_daily(monitoring_location_id, parameter_code, statistic_id, properties, time_series_id, daily_id, approval_status, unit_of_measure, qualifier, value, last_modified, skip_geometry, time, bbox, limit, filter, filter_lang, convert_type, max_rows, **queryables)
254 # Build argument dictionary, omitting None values
255 args = _get_args(locals(), exclude={"max_rows"})
--> 257 return get_ogc_data(args, collection, max_rows=max_rows)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/utils.py:240, in get_ogc_data(args, collection, output_id, max_rows, cql_body, spatial)
238 if output_id is None:
239 output_id = _OUTPUT_ID_BY_COLLECTION[collection]
--> 240 return _facade_get_ogc_data(
241 args,
242 collection,
243 output_id,
244 max_rows=max_rows,
245 # The endpoint is resolved from the active ContextVar at request time;
246 # the documented ``OGC_API_URL`` constant remains the default-value
247 # compatibility path rather than a production request destination.
248 base_url=ogc_api_url(),
249 spatial=spatial,
250 extra_id_cols=_EXTRA_ID_COLS,
251 dialect=WATERDATA_DIALECT,
252 cql_body=cql_body,
253 # Which settings table these calls read. Declared here, in the one
254 # wrapper every Water Data getter goes through, rather than derived
255 # from ``base_url``: NGWMN is served from the same host, so a URL
256 # does not distinguish the two adapters (ADR 0010).
257 adapter="waterdata",
258 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:390, in get_ogc_data(args, collection, output_id, base_url, spatial, max_rows, extra_id_cols, dialect, cql_body, adapter)
385 run = chunking.multi_value_chunked(build_request=build_request, adapter=adapter)(
386 fetch
387 )
388 # No progress block here: the executor that emits the events owns the line
389 # (see :meth:`~dataretrieval.transport.fanout.FanOut.resume`).
--> 390 return run(args, finalize=finalize)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/chunking.py:261, in multi_value_chunked.<locals>.decorator.<locals>.wrapper(args, finalize)
247 retry_policy = RetryPolicy.from_configuration(adapter=adapter)
248 # The concurrency cap is resolved inside ``resume()`` through the
249 # configuration chain; ``1`` is a sequential gather,
250 # ``total <= 1`` a one-element gather — no special branch.
251 return ChunkedCall(
252 plan,
253 fetch,
254 retry_policy,
255 finalize,
256 canonical_url=plan.canonical_url,
257 # The collection name, for the progress line the executor
258 # opens. ``get_ogc_data`` puts it in ``args``.
259 service=args.get("collection"),
260 adapter=adapter,
--> 261 ).resume()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:494, in FanOut.resume(self)
486 concurrency = _configuration.concurrency(
487 self.default_concurrent, adapter=self.adapter
488 )
489 with start_blocking_portal() as portal:
490 # ``portal.call`` returns ``Any`` because ``functools.partial``
491 # erases ``_run``'s return type; restore the declared tuple.
492 return cast(
493 "tuple[pd.DataFrame, Any]",
--> 494 portal.call(functools.partial(self._run, concurrency)),
495 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:340, in BlockingPortal.call(self, func, *args)
325 def call(
326 self,
327 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
328 *args: Unpack[PosArgsT],
329 ) -> T_Retval:
330 """
331 Call the given function in the event loop thread.
332
(...) 338
339 """
--> 340 return cast(T_Retval, self.start_task_soon(func, *args).result())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:460, in Future.result(self, timeout)
458 raise CancelledError()
459 elif self._state == FINISHED:
--> 460 return self.__get_result()
461 else:
462 raise TimeoutError()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:402, in Future.__get_result(self)
400 if self._exception is not None:
401 try:
--> 402 raise self._exception
403 finally:
404 # Break a reference cycle with the exception in self._exception
405 self = None
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:265, in BlockingPortal._call_func(self, func, args, kwargs, future)
263 with CancelScope() as scope:
264 future.add_done_callback(callback)
--> 265 retval = await retval_or_awaitable
266 else:
267 retval = retval_or_awaitable
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:633, in FanOut._run(self, max_concurrent)
625 # Dispatch every pending chunk concurrently; the
626 # semaphore (held by ``_retry`` per attempt) is the only throttle.
627 # ``return_exceptions`` keeps completed pairs after a sibling
628 # fails, so partial state stays recoverable via :meth:`resume`.
629 results = await asyncio.gather(
630 *(track(index, item) for index, item in self._pending()),
631 return_exceptions=True,
632 )
--> 633 self._handle_gather_failures(results)
635 return self.finalize(*self._combine_raw())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:540, in FanOut._handle_gather_failures(self, results)
537 if interrupted is None:
538 # Unreachable: classified as transient above.
539 raise self._normalize_failure(first_transient)
--> 540 raise interrupted from first_transient
QuotaExhausted: HTTP 429 after 0/1 chunks; catch QuotaExhausted (or FanOutInterrupted) to access .partial_frame or .call.resume() once the rate-limit window has reset. Cause: RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/ogcapi/v0/collections/daily/items?monitoring_location_id=USGS-01491000¶meter_code=00010%2C00060&statistic_id=00001%2C00003&time=2012-01-01%2F2012-05-01&limit=50000)
[6]:
"""
parameterCd <- "00060" # Discharge
startDate <- "2012-05-12"
endDate <- "2012-05-13"
dischargeUnit <- readNWISuv(siteNumber, parameterCd, startDate, endDate)
"""
siteNumber = "USGS-01491000"
parameterCd = "00060" # Discharge
dischargeUnit, md = waterdata.get_continuous(
monitoring_location_id=siteNumber,
parameter_code=parameterCd,
time="2012-05-12/2012-05-13",
)
---------------------------------------------------------------------------
RateLimited Traceback (most recent call last)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:615, in FanOut._run.<locals>.track(index, item)
614 """One chunk (with retry) + result-store + progress update."""
--> 615 result = await _retry(
616 lambda: self.fetch(item), self.retry_policy, gate=semaphore
617 )
618 self._chunks[index] = result
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:291, in retry_async(afn, policy, gate)
290 try:
--> 291 return await attempt_once()
292 except Exception as exc:
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:287, in retry_async.<locals>.attempt_once()
286 credit_wait(time.monotonic() - started)
--> 287 return await afn()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:416, in _fetch_once(args, build_request, geopd, include_geometry, row_cap)
415 req = build_request(**args)
--> 416 return await _walk_pages(
417 geopd=geopd,
418 req=req,
419 include_geometry=include_geometry,
420 row_cap=row_cap,
421 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:203, in _walk_pages(geopd, req, client, include_geometry, row_cap)
201 return await sess.request(method, cursor, headers=headers, content=content)
--> 203 return await paginate(
204 req,
205 parse_response=functools.partial(
206 _ogc_parse_response,
207 geopd=geopd,
208 include_geometry=include_geometry,
209 ),
210 follow_up=follow_up,
211 client=client,
212 raise_for_status=_raise_for_non_200,
213 row_cap=row_cap,
214 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/pagination.py:112, in paginate(initial_req, parse_response, follow_up, raise_for_status, client, row_cap)
111 response = await session.send(initial_req)
--> 112 raise_for_status(response)
113 initial_response = response
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/errors.py:139, in _raise_for_non_200(resp)
138 return
--> 139 raise error_for_status(
140 status,
141 _error_body(resp) + _url_suffix(resp),
142 retry_after=_parse_retry_after(resp.headers.get("Retry-After")),
143 )
RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/ogcapi/v0/collections/continuous/items?monitoring_location_id=USGS-01491000¶meter_code=00060&time=2012-05-12%2F2012-05-13&limit=50000)
The above exception was the direct cause of the following exception:
QuotaExhausted Traceback (most recent call last)
Cell In[6], line 10
6 """
7 siteNumber = "USGS-01491000"
8 parameterCd = "00060" # Discharge
9
---> 10 dischargeUnit, md = waterdata.get_continuous(
11 monitoring_location_id=siteNumber,
12 parameter_code=parameterCd,
13 time="2012-05-12/2012-05-13",
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/time_series.py:455, in get_continuous(monitoring_location_id, parameter_code, statistic_id, properties, time_series_id, continuous_id, approval_status, unit_of_measure, qualifier, value, last_modified, time, limit, filter, filter_lang, convert_type, max_rows, **queryables)
452 # Build argument dictionary, omitting None values
453 args = _get_args(locals(), exclude={"max_rows"})
--> 455 return get_ogc_data(args, collection, max_rows=max_rows)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/utils.py:240, in get_ogc_data(args, collection, output_id, max_rows, cql_body, spatial)
238 if output_id is None:
239 output_id = _OUTPUT_ID_BY_COLLECTION[collection]
--> 240 return _facade_get_ogc_data(
241 args,
242 collection,
243 output_id,
244 max_rows=max_rows,
245 # The endpoint is resolved from the active ContextVar at request time;
246 # the documented ``OGC_API_URL`` constant remains the default-value
247 # compatibility path rather than a production request destination.
248 base_url=ogc_api_url(),
249 spatial=spatial,
250 extra_id_cols=_EXTRA_ID_COLS,
251 dialect=WATERDATA_DIALECT,
252 cql_body=cql_body,
253 # Which settings table these calls read. Declared here, in the one
254 # wrapper every Water Data getter goes through, rather than derived
255 # from ``base_url``: NGWMN is served from the same host, so a URL
256 # does not distinguish the two adapters (ADR 0010).
257 adapter="waterdata",
258 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:390, in get_ogc_data(args, collection, output_id, base_url, spatial, max_rows, extra_id_cols, dialect, cql_body, adapter)
385 run = chunking.multi_value_chunked(build_request=build_request, adapter=adapter)(
386 fetch
387 )
388 # No progress block here: the executor that emits the events owns the line
389 # (see :meth:`~dataretrieval.transport.fanout.FanOut.resume`).
--> 390 return run(args, finalize=finalize)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/chunking.py:261, in multi_value_chunked.<locals>.decorator.<locals>.wrapper(args, finalize)
247 retry_policy = RetryPolicy.from_configuration(adapter=adapter)
248 # The concurrency cap is resolved inside ``resume()`` through the
249 # configuration chain; ``1`` is a sequential gather,
250 # ``total <= 1`` a one-element gather — no special branch.
251 return ChunkedCall(
252 plan,
253 fetch,
254 retry_policy,
255 finalize,
256 canonical_url=plan.canonical_url,
257 # The collection name, for the progress line the executor
258 # opens. ``get_ogc_data`` puts it in ``args``.
259 service=args.get("collection"),
260 adapter=adapter,
--> 261 ).resume()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:494, in FanOut.resume(self)
486 concurrency = _configuration.concurrency(
487 self.default_concurrent, adapter=self.adapter
488 )
489 with start_blocking_portal() as portal:
490 # ``portal.call`` returns ``Any`` because ``functools.partial``
491 # erases ``_run``'s return type; restore the declared tuple.
492 return cast(
493 "tuple[pd.DataFrame, Any]",
--> 494 portal.call(functools.partial(self._run, concurrency)),
495 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:340, in BlockingPortal.call(self, func, *args)
325 def call(
326 self,
327 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
328 *args: Unpack[PosArgsT],
329 ) -> T_Retval:
330 """
331 Call the given function in the event loop thread.
332
(...) 338
339 """
--> 340 return cast(T_Retval, self.start_task_soon(func, *args).result())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:460, in Future.result(self, timeout)
458 raise CancelledError()
459 elif self._state == FINISHED:
--> 460 return self.__get_result()
461 else:
462 raise TimeoutError()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:402, in Future.__get_result(self)
400 if self._exception is not None:
401 try:
--> 402 raise self._exception
403 finally:
404 # Break a reference cycle with the exception in self._exception
405 self = None
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:265, in BlockingPortal._call_func(self, func, args, kwargs, future)
263 with CancelScope() as scope:
264 future.add_done_callback(callback)
--> 265 retval = await retval_or_awaitable
266 else:
267 retval = retval_or_awaitable
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:633, in FanOut._run(self, max_concurrent)
625 # Dispatch every pending chunk concurrently; the
626 # semaphore (held by ``_retry`` per attempt) is the only throttle.
627 # ``return_exceptions`` keeps completed pairs after a sibling
628 # fails, so partial state stays recoverable via :meth:`resume`.
629 results = await asyncio.gather(
630 *(track(index, item) for index, item in self._pending()),
631 return_exceptions=True,
632 )
--> 633 self._handle_gather_failures(results)
635 return self.finalize(*self._combine_raw())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:540, in FanOut._handle_gather_failures(self, results)
537 if interrupted is None:
538 # Unreachable: classified as transient above.
539 raise self._normalize_failure(first_transient)
--> 540 raise interrupted from first_transient
QuotaExhausted: HTTP 429 after 0/1 chunks; catch QuotaExhausted (or FanOutInterrupted) to access .partial_frame or .call.resume() once the rate-limit window has reset. Cause: RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/ogcapi/v0/collections/continuous/items?monitoring_location_id=USGS-01491000¶meter_code=00060&time=2012-05-12%2F2012-05-13&limit=50000)
[7]:
"""
# Dissolved Nitrate parameter codes:
parameterCd <- c("00618","71851")
startDate <- "1985-10-01"
endDate <- "2012-09-30"
dfLong <- read_USGS_samples(monitoringLocationIdentifier=sprintf("USGS-%s", siteNumber),
usgsPCode=parameterCd, activityStartDateLower=startDate, activityStartDateUpper=endDate)
"""
siteNumber = "USGS-01491000"
parameterCd = ["00618", "71851"]
dfLong, md = waterdata.get_samples(
monitoringLocationIdentifier=siteNumber,
usgsPCode=parameterCd,
activityStartDateLower="1985-10-01",
activityStartDateUpper="2012-09-30",
)
/tmp/ipykernel_6565/2092726829.py:12: DeprecationWarning: The 'activityStartDateLower' argument is deprecated and will be removed from `dataretrieval` in a future release; use 'activity_start_date_lower' instead.
dfLong, md = waterdata.get_samples(
/tmp/ipykernel_6565/2092726829.py:12: DeprecationWarning: The 'activityStartDateUpper' argument is deprecated and will be removed from `dataretrieval` in a future release; use 'activity_start_date_upper' instead.
dfLong, md = waterdata.get_samples(
/tmp/ipykernel_6565/2092726829.py:12: DeprecationWarning: The 'usgsPCode' argument is deprecated and will be removed from `dataretrieval` in a future release; use 'usgs_pcode' instead.
dfLong, md = waterdata.get_samples(
/tmp/ipykernel_6565/2092726829.py:12: DeprecationWarning: The 'monitoringLocationIdentifier' argument is deprecated and will be removed from `dataretrieval` in a future release; use 'monitoring_location_id' instead.
dfLong, md = waterdata.get_samples(
[8]:
"""
siteNumber <- '01594440'
peakData <- readNWISpeak(siteNumber)
"""
peakData, md = waterdata.get_peaks(
monitoring_location_id="USGS-01594440", parameter_code="00060"
)
---------------------------------------------------------------------------
RateLimited Traceback (most recent call last)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:615, in FanOut._run.<locals>.track(index, item)
614 """One chunk (with retry) + result-store + progress update."""
--> 615 result = await _retry(
616 lambda: self.fetch(item), self.retry_policy, gate=semaphore
617 )
618 self._chunks[index] = result
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:291, in retry_async(afn, policy, gate)
290 try:
--> 291 return await attempt_once()
292 except Exception as exc:
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:287, in retry_async.<locals>.attempt_once()
286 credit_wait(time.monotonic() - started)
--> 287 return await afn()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:416, in _fetch_once(args, build_request, geopd, include_geometry, row_cap)
415 req = build_request(**args)
--> 416 return await _walk_pages(
417 geopd=geopd,
418 req=req,
419 include_geometry=include_geometry,
420 row_cap=row_cap,
421 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:203, in _walk_pages(geopd, req, client, include_geometry, row_cap)
201 return await sess.request(method, cursor, headers=headers, content=content)
--> 203 return await paginate(
204 req,
205 parse_response=functools.partial(
206 _ogc_parse_response,
207 geopd=geopd,
208 include_geometry=include_geometry,
209 ),
210 follow_up=follow_up,
211 client=client,
212 raise_for_status=_raise_for_non_200,
213 row_cap=row_cap,
214 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/pagination.py:112, in paginate(initial_req, parse_response, follow_up, raise_for_status, client, row_cap)
111 response = await session.send(initial_req)
--> 112 raise_for_status(response)
113 initial_response = response
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/errors.py:139, in _raise_for_non_200(resp)
138 return
--> 139 raise error_for_status(
140 status,
141 _error_body(resp) + _url_suffix(resp),
142 retry_after=_parse_retry_after(resp.headers.get("Retry-After")),
143 )
RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/ogcapi/v0/collections/peaks/items?monitoring_location_id=USGS-01594440¶meter_code=00060&limit=50000)
The above exception was the direct cause of the following exception:
QuotaExhausted Traceback (most recent call last)
Cell In[8], line 5
1 """
2 siteNumber <- '01594440'
3 peakData <- readNWISpeak(siteNumber)
4 """
----> 5 peakData, md = waterdata.get_peaks(
6 monitoring_location_id="USGS-01594440", parameter_code="00060"
7 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/measurements.py:368, in get_peaks(monitoring_location_id, parameter_code, time_series_id, unit_of_measure, time, last_modified, water_year, year, month, day, peak_since, properties, skip_geometry, bbox, limit, filter, filter_lang, convert_type, max_rows, **queryables)
364 collection = "peaks"
366 args = _get_args(locals(), exclude={"max_rows"})
--> 368 return get_ogc_data(args, collection, max_rows=max_rows)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/utils.py:240, in get_ogc_data(args, collection, output_id, max_rows, cql_body, spatial)
238 if output_id is None:
239 output_id = _OUTPUT_ID_BY_COLLECTION[collection]
--> 240 return _facade_get_ogc_data(
241 args,
242 collection,
243 output_id,
244 max_rows=max_rows,
245 # The endpoint is resolved from the active ContextVar at request time;
246 # the documented ``OGC_API_URL`` constant remains the default-value
247 # compatibility path rather than a production request destination.
248 base_url=ogc_api_url(),
249 spatial=spatial,
250 extra_id_cols=_EXTRA_ID_COLS,
251 dialect=WATERDATA_DIALECT,
252 cql_body=cql_body,
253 # Which settings table these calls read. Declared here, in the one
254 # wrapper every Water Data getter goes through, rather than derived
255 # from ``base_url``: NGWMN is served from the same host, so a URL
256 # does not distinguish the two adapters (ADR 0010).
257 adapter="waterdata",
258 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/engine.py:390, in get_ogc_data(args, collection, output_id, base_url, spatial, max_rows, extra_id_cols, dialect, cql_body, adapter)
385 run = chunking.multi_value_chunked(build_request=build_request, adapter=adapter)(
386 fetch
387 )
388 # No progress block here: the executor that emits the events owns the line
389 # (see :meth:`~dataretrieval.transport.fanout.FanOut.resume`).
--> 390 return run(args, finalize=finalize)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/chunking.py:261, in multi_value_chunked.<locals>.decorator.<locals>.wrapper(args, finalize)
247 retry_policy = RetryPolicy.from_configuration(adapter=adapter)
248 # The concurrency cap is resolved inside ``resume()`` through the
249 # configuration chain; ``1`` is a sequential gather,
250 # ``total <= 1`` a one-element gather — no special branch.
251 return ChunkedCall(
252 plan,
253 fetch,
254 retry_policy,
255 finalize,
256 canonical_url=plan.canonical_url,
257 # The collection name, for the progress line the executor
258 # opens. ``get_ogc_data`` puts it in ``args``.
259 service=args.get("collection"),
260 adapter=adapter,
--> 261 ).resume()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:494, in FanOut.resume(self)
486 concurrency = _configuration.concurrency(
487 self.default_concurrent, adapter=self.adapter
488 )
489 with start_blocking_portal() as portal:
490 # ``portal.call`` returns ``Any`` because ``functools.partial``
491 # erases ``_run``'s return type; restore the declared tuple.
492 return cast(
493 "tuple[pd.DataFrame, Any]",
--> 494 portal.call(functools.partial(self._run, concurrency)),
495 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:340, in BlockingPortal.call(self, func, *args)
325 def call(
326 self,
327 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
328 *args: Unpack[PosArgsT],
329 ) -> T_Retval:
330 """
331 Call the given function in the event loop thread.
332
(...) 338
339 """
--> 340 return cast(T_Retval, self.start_task_soon(func, *args).result())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:460, in Future.result(self, timeout)
458 raise CancelledError()
459 elif self._state == FINISHED:
--> 460 return self.__get_result()
461 else:
462 raise TimeoutError()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:402, in Future.__get_result(self)
400 if self._exception is not None:
401 try:
--> 402 raise self._exception
403 finally:
404 # Break a reference cycle with the exception in self._exception
405 self = None
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:265, in BlockingPortal._call_func(self, func, args, kwargs, future)
263 with CancelScope() as scope:
264 future.add_done_callback(callback)
--> 265 retval = await retval_or_awaitable
266 else:
267 retval = retval_or_awaitable
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:633, in FanOut._run(self, max_concurrent)
625 # Dispatch every pending chunk concurrently; the
626 # semaphore (held by ``_retry`` per attempt) is the only throttle.
627 # ``return_exceptions`` keeps completed pairs after a sibling
628 # fails, so partial state stays recoverable via :meth:`resume`.
629 results = await asyncio.gather(
630 *(track(index, item) for index, item in self._pending()),
631 return_exceptions=True,
632 )
--> 633 self._handle_gather_failures(results)
635 return self.finalize(*self._combine_raw())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:540, in FanOut._handle_gather_failures(self, results)
537 if interrupted is None:
538 # Unreachable: classified as transient above.
539 raise self._normalize_failure(first_transient)
--> 540 raise interrupted from first_transient
QuotaExhausted: HTTP 429 after 0/1 chunks; catch QuotaExhausted (or FanOutInterrupted) to access .partial_frame or .call.resume() once the rate-limit window has reset. Cause: RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/ogcapi/v0/collections/peaks/items?monitoring_location_id=USGS-01594440¶meter_code=00060&limit=50000)
[9]:
"""
ratingData <- readNWISrating(siteNumber, "base")
attr(ratingData, "RATING")
"""
# get_ratings returns a dict keyed by "<id>.<file_type>.rdb"
ratings_data = waterdata.get_ratings(
monitoring_location_id="USGS-01594440", file_type="base"
)
list(ratings_data.keys())
---------------------------------------------------------------------------
RateLimited Traceback (most recent call last)
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:615, in FanOut._run.<locals>.track(index, item)
614 """One chunk (with retry) + result-store + progress update."""
--> 615 result = await _retry(
616 lambda: self.fetch(item), self.retry_policy, gate=semaphore
617 )
618 self._chunks[index] = result
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:291, in retry_async(afn, policy, gate)
290 try:
--> 291 return await attempt_once()
292 except Exception as exc:
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/retry.py:287, in retry_async.<locals>.attempt_once()
286 credit_wait(time.monotonic() - started)
--> 287 return await afn()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/pagination.py:192, in run_paginated.<locals>.fetch(request)
191 async def fetch(request: httpx.Request) -> tuple[pd.DataFrame, httpx.Response]:
--> 192 return await paginate(
193 request,
194 parse_response=parse_response,
195 follow_up=follow_up,
196 client=client,
197 raise_for_status=raise_for_status,
198 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/pagination.py:112, in paginate(initial_req, parse_response, follow_up, raise_for_status, client, row_cap)
111 response = await session.send(initial_req)
--> 112 raise_for_status(response)
113 initial_response = response
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/ogc/errors.py:139, in _raise_for_non_200(resp)
138 return
--> 139 raise error_for_status(
140 status,
141 _error_body(resp) + _url_suffix(resp),
142 retry_after=_parse_retry_after(resp.headers.get("Retry-After")),
143 )
RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/stac/v0/search?limit=10000&filter=monitoring_location_id+IN+%28%27USGS-01594440%27%29+AND+file_type+%3D+%27base%27)
The above exception was the direct cause of the following exception:
QuotaExhausted Traceback (most recent call last)
Cell In[9], line 6
2 ratingData <- readNWISrating(siteNumber, "base")
3 attr(ratingData, "RATING")
4 """
5 # get_ratings returns a dict keyed by "<id>.<file_type>.rdb"
----> 6 ratings_data = waterdata.get_ratings(
7 monitoring_location_id="USGS-01594440", file_type="base"
8 )
9 list(ratings_data.keys())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/ratings.py:190, in get_ratings(monitoring_location_id, file_type, file_path, time, bbox, limit, download_and_parse, ssl_check)
187 server_file_type = file_types[0] if len(file_types) == 1 else None
188 filter_str = _build_filter(monitoring_location_id, server_file_type)
--> 190 features = _search(filter_str, time_str, bbox, limit, ssl_check)
192 if not download_and_parse:
193 return features
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/waterdata/ratings.py:307, in _search(filter_str, time_str, bbox, limit, ssl_check)
304 async def follow_up(cursor: str, sess: httpx.AsyncClient) -> httpx.Response:
305 return await sess.get(cursor, headers=_default_headers(cursor))
--> 307 df, _ = run_paginated(
308 [req],
309 parse_response=parse_response,
310 follow_up=follow_up,
311 raise_for_status=_raise_for_non_200,
312 client_options={"verify": ssl_check},
313 service="ratings",
314 adapter="waterdata",
315 )
316 # Every page frame is built with a ``feature`` column, and the combine
317 # helpers preserve it, so the empty case needs no special branch.
318 return list(df["feature"])
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/pagination.py:212, in run_paginated(requests, parse_response, follow_up, raise_for_status, service, finalize, client, client_options, default_concurrent, canonical_url, adapter)
200 if canonical_url is None and requests:
201 canonical_url = str(requests[0].url)
202 return FanOut(
203 requests,
204 fetch,
205 RetryPolicy.from_configuration(adapter=adapter),
206 finalize=finalize,
207 client_options=client_options,
208 default_concurrent=default_concurrent,
209 canonical_url=canonical_url,
210 service=service,
211 adapter=adapter,
--> 212 ).resume()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:494, in FanOut.resume(self)
486 concurrency = _configuration.concurrency(
487 self.default_concurrent, adapter=self.adapter
488 )
489 with start_blocking_portal() as portal:
490 # ``portal.call`` returns ``Any`` because ``functools.partial``
491 # erases ``_run``'s return type; restore the declared tuple.
492 return cast(
493 "tuple[pd.DataFrame, Any]",
--> 494 portal.call(functools.partial(self._run, concurrency)),
495 )
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:340, in BlockingPortal.call(self, func, *args)
325 def call(
326 self,
327 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
328 *args: Unpack[PosArgsT],
329 ) -> T_Retval:
330 """
331 Call the given function in the event loop thread.
332
(...) 338
339 """
--> 340 return cast(T_Retval, self.start_task_soon(func, *args).result())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:460, in Future.result(self, timeout)
458 raise CancelledError()
459 elif self._state == FINISHED:
--> 460 return self.__get_result()
461 else:
462 raise TimeoutError()
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/concurrent/futures/_base.py:402, in Future.__get_result(self)
400 if self._exception is not None:
401 try:
--> 402 raise self._exception
403 finally:
404 # Break a reference cycle with the exception in self._exception
405 self = None
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/anyio/from_thread.py:265, in BlockingPortal._call_func(self, func, args, kwargs, future)
263 with CancelScope() as scope:
264 future.add_done_callback(callback)
--> 265 retval = await retval_or_awaitable
266 else:
267 retval = retval_or_awaitable
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:633, in FanOut._run(self, max_concurrent)
625 # Dispatch every pending chunk concurrently; the
626 # semaphore (held by ``_retry`` per attempt) is the only throttle.
627 # ``return_exceptions`` keeps completed pairs after a sibling
628 # fails, so partial state stays recoverable via :meth:`resume`.
629 results = await asyncio.gather(
630 *(track(index, item) for index, item in self._pending()),
631 return_exceptions=True,
632 )
--> 633 self._handle_gather_failures(results)
635 return self.finalize(*self._combine_raw())
File /opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/transport/fanout.py:540, in FanOut._handle_gather_failures(self, results)
537 if interrupted is None:
538 # Unreachable: classified as transient above.
539 raise self._normalize_failure(first_transient)
--> 540 raise interrupted from first_transient
QuotaExhausted: HTTP 429 after 0/1 chunks; catch QuotaExhausted (or FanOutInterrupted) to access .partial_frame or .call.resume() once the rate-limit window has reset. Cause: RateLimited: 429: Too many requests made. Please obtain an API token or try again later. (URL: https://api.waterdata.usgs.gov/stac/v0/search?limit=10000&filter=monitoring_location_id+IN+%28%27USGS-01594440%27%29+AND+file_type+%3D+%27base%27)
[10]:
"""
discharge_stats <- readNWISstat(siteNumbers=c("02319394"),
parameterCd=c("00060"),
statReportType="annual")
"""
discharge_stats, md = waterdata.get_stats_date_range(
monitoring_location_id="USGS-02319394",
parameter_code="00060",
computation_type="arithmetic_mean",
)
Retrieving: observationIntervals · 1 page · 344 rows
[11]:
# R: readNWISdata(service="dv", stateCd="WI", parameterCd="00060",
# drainAreaMin="50", statCd="00003")
#
# The Water Data API serves daily values per monitoring location. To assemble a
# state-wide set, first find the locations (optionally filtering by drainage
# area) with waterdata.get_monitoring_locations(state_name="Wisconsin", ...),
# then pass their ids to waterdata.get_daily(parameter_code="00060",
# statistic_id="00003").
[12]:
# '''
# sitesNJ <- whatWQPsites(statecode="US:34",
# characteristicName="Chloride")
# '''
# sitesNJ, md = wqp.what_sites(statecode="US:34", characteristicName="Chloride")
[13]:
# '''
# dataPH <- readWQPdata(statecode="US:55",
# characteristicName="pH")
# '''
# dataPH, md = wqp.what_sites(statecode="US:55", characteristicName="pH")
[14]:
# '''
# type <- "Stream"
# sites <- whatWQPdata(countycode="US:55:025",siteType=type)
# '''
# streamType = "Stream"
# sites, md = wqp.get_results(countycode="US:55:025", siteType=streamType)
[15]:
"""site <- whatWQPsamples(siteid="USGS-01594440")"""
site, md = wqp.what_sites(siteid="USGS-01594440")
/opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/wqp.py:661: DataCurrencyWarning: This function call will return the legacy WQX format, which means USGS data have not been updated since March 2024. Please review the dataretrieval-python documentation for more information on updated WQX3.0 profiles. Setting `legacy=False` will remove this warning.
_warn_legacy_use()
[16]:
"""
type <- "Stream"
sites <- whatWQPmetrics(countycode="US:55:025",siteType=type)
"""
streamType = "Stream"
sites, md = wqp.what_sites(countycode="US:55:025", siteType=streamType)
/opt/hostedtoolcache/Python/3.13.15/x64/lib/python3.13/site-packages/dataretrieval/wqp.py:661: DataCurrencyWarning: This function call will return the legacy WQX format, which means USGS data have not been updated since March 2024. Please review the dataretrieval-python documentation for more information on updated WQX3.0 profiles. Setting `legacy=False` will remove this warning.
_warn_legacy_use()
Embedded Metadata
Most waterdata and wqp service methods return a tuple of the requested data (a pandas DataFrame) and a metadata object.
md is an object with the following attributes:
Metadata
url # the URL used to query the service
query_time # how long the query took
header # the response headers
Note: USGS water use data has no Water Data API equivalent yet. The legacy nwis.get_water_use() service has been decommissioned and now raises a “defunct” error, so there is currently no runnable way to retrieve water-use data through dataretrieval.