diff --git a/README.md b/README.md index 88b076ab..c1333d69 100644 --- a/README.md +++ b/README.md @@ -42,15 +42,41 @@ pip install git+https://github.com/DOI-USGS/dataretrieval-python.git Access USGS water-monitoring data. **Important:** Users are strongly encouraged to obtain an API key for higher -rate limits. [Register for an API key](https://api.waterdata.usgs.gov/signup/) -and set it as an environment variable: +rate limits. [Register for an API key](https://api.waterdata.usgs.gov/signup/), +then supply it in whichever of these ways suits you. They are listed from +highest to lowest precedence, so an explicit block or deployment environment +can override a file without editing it: ```python -import os +# 1. a configure() block - for one call, an interactive prompt, or when +# different threads/tasks need different credentials. +from getpass import getpass -os.environ["API_USGS_PAT"] = "your_api_key_here" +import dataretrieval +from dataretrieval import waterdata + +with dataretrieval.configure(api_key=getpass("USGS API key: ")): + df, metadata = waterdata.get_daily(monitoring_location_id="USGS-01646500") +``` + +```bash +# 2. an environment variable (the R dataRetrieval package uses the same +# variable, so one export serves both) +export API_USGS_PAT="your_api_key_here" +``` + +```toml +# 3. ~/.dataretrieval/config.toml - keeps the key out of your shell +# environment, where every process you start inherits it. +# Restrict it afterwards: chmod 600 ~/.dataretrieval/config.toml +api_key = "your_api_key_here" ``` +`dataretrieval.show_config()` reports what is in effect and where each setting +came from, without printing the key. Concurrency, retries, and the progress +line are configured the same way — see the +[configuration guide](https://doi-usgs.github.io/dataretrieval-python/userguide/configuration.html). + The following example retrieves daily streamflow data for a specific monitoring location. The `/` in the `time` argument separates the start and end of the desired range: @@ -112,12 +138,14 @@ By default the getters split a multi-value request only as far as the server's ~8 KB URL limit forces — the fewest sub-requests. For a **large, paginated** pull that is needlessly conservative: every sub-request pages through its own results, so dividing the query into more, smaller sub-requests lets those pages -be fetched **in parallel**. `parallel_chunks(n)` opts a single call into that -finer split, fanning it out into `n` sub-requests. It pays off only when the -result is large enough to span many pages *and* the query has a multi-value -argument to divide (such as a list of monitoring locations); on a small query — -or one with nothing to split — it just adds requests, so it is a deliberate, -scoped `with` block, never the default. +be fetched **in parallel**. `parallel_chunks(n)` opts a single call into finer +optional splitting, up to `n` sub-requests when the input divides that way. It +pays off only when the result spans many pages and the query has a multi-value +argument to divide (such as a list of monitoring locations). A query with +nothing to split remains one request. On a small but splittable query, extra +chunks only spend quota, so the scoped block is the recommended usage. A +deliberate config-file or profile baseline is also available for repeated large +pulls. ```python from dataretrieval import waterdata @@ -126,7 +154,7 @@ from dataretrieval import waterdata # enough to span many pages, so it profits from a finer split. sites, _ = waterdata.get_monitoring_locations(state="Ohio", site_type_code="ST") -with waterdata.parallel_chunks(32): # fan out into 32 sub-requests +with waterdata.parallel_chunks(32): # request up to 32 optional chunks df, md = waterdata.get_daily( monitoring_location_id=sites["monitoring_location_id"], parameter_code="00060", # discharge @@ -134,11 +162,13 @@ with waterdata.parallel_chunks(32): # fan out into 32 sub-requests ) ``` -`n` is the number of sub-requests to fan the call out into. It is capped by how -many values there are to split, and each sub-request costs a request against -your hourly [rate limit](https://api.waterdata.usgs.gov/signup/); since how many -run *at once* is capped separately by `API_USGS_CONCURRENT` (default 32), the -useful range is roughly `2` up to that value. +`n` is the ceiling for optional refinement, not a hard ceiling on URL-safety +chunking: an oversized request may already require more than `n` sub-requests, +while indivisible inputs may produce fewer. Each sub-request costs a request +against your hourly [rate limit](https://api.waterdata.usgs.gov/signup/). How +many run *at once* is controlled separately by the effective `concurrency` +setting (`API_USGS_CONCURRENT`), so the useful optional range is +roughly `2` up to that value. Benchmark — a fixed 271-site subset of Ohio stream gages (`get_daily`, `parameter_code="00060"`), with a small fixed page size @@ -146,17 +176,17 @@ Benchmark — a fixed 271-site subset of Ohio stream gages the effect of parallelism). Each `n` was run against its own cold 1-year time window so no run is served from the server's data-window cache: -| `n` | parallelism | pages | wall-clock | speedup | -| ---- | ----------- | ----- | ----------------------- | ------- | -| off | 1 | ~30 | 9.5 s / 9.1 s (2 runs) | 1× | -| `8` | 8 | ~32 | 2.2 s / 1.9 s | ~4.5× | -| `32` | 32 | 54 | 1.2 s | ~8× | +| `n` | optional fan-out | pages | wall-clock | speedup | +| ---- | ---------------- | ----- | ---------------------- | ------- | +| off | 1 | ~30 | 9.5 s / 9.1 s (2 runs) | 1× | +| `8` | 8 | ~32 | 2.2 s / 1.9 s | ~4.5× | +| `32` | 32 | 54 | 1.2 s | ~8× | The gain comes from overlapping each sub-request's per-page latency and server-side work, so the exact multiplier scales with how many pages the pull -spans — a larger pull (more pages) has more parallelism to exploit. The extra -sub-requests each cost quota, so reserve a large `n` for pulls you know are -large. +spans — a larger pull gives the executor more independent chunks to schedule. +The extra sub-requests each cost quota, so reserve a large `n` for pulls you +know are large. Visit the [API Reference](https://doi-usgs.github.io/dataretrieval-python/reference/waterdata.html) diff --git a/dataretrieval/__init__.py b/dataretrieval/__init__.py index 469fe0f5..5e302d23 100644 --- a/dataretrieval/__init__.py +++ b/dataretrieval/__init__.py @@ -17,6 +17,12 @@ ``nldi`` requires geopandas (``pip install dataretrieval[nldi]``) and is imported on demand: ``from dataretrieval import nldi``. +Settings -- the Water Data API key, fan-out concurrency, retries, the progress +line -- resolve through :mod:`dataretrieval.config`: a +``with dataretrieval.configure(...)`` block, then the ``API_USGS_*`` environment +variables, then ``~/.dataretrieval/config.toml``. ``dataretrieval.show_config()`` +reports what is in effect and where each value came from. + A failed request raises a subclass of :class:`dataretrieval.DataRetrievalError` (the taxonomy lives in ``dataretrieval.exceptions``); connection-level failures (timeouts, DNS) are wrapped as :class:`dataretrieval.NetworkError`. A large @@ -31,6 +37,10 @@ except PackageNotFoundError: __version__ = "version-unknown" +# Layered configuration: a ``with configure(...)`` block, the environment, then +# the config file. The canonical home is ``dataretrieval.config``; +# the callable is named ``configure`` so it doesn't shadow that module. +from dataretrieval.config import ConfigError, configure, show_config from dataretrieval.exceptions import ( DataRetrievalError, HTTPError, @@ -62,6 +72,7 @@ ) from . import ( + config, exceptions, ngwmn, nwis, @@ -73,6 +84,11 @@ ) __all__ = [ + # layered configuration (canonical home: ``dataretrieval.config``) + "config", + "configure", + "show_config", + "ConfigError", # service modules "ngwmn", "nwis", diff --git a/dataretrieval/config.py b/dataretrieval/config.py new file mode 100644 index 00000000..5421b65e --- /dev/null +++ b/dataretrieval/config.py @@ -0,0 +1,1022 @@ +"""Layered configuration resolution for ``dataretrieval``. + +Every tunable setting -- the Water Data API key, the fan-out concurrency cap, +the retry count, and the progress line -- resolves through one ordered chain so +a caller never has to mutate ``os.environ`` to configure a single call. + +Sources, highest precedence first: + +1. An active :func:`configure` block -- a :class:`~contextvars.ContextVar`, so a + setting applies to the current thread or asyncio task and cannot leak into + another one. +2. The environment variable for that setting (``API_USGS_PAT``, + ``API_USGS_CONCURRENT``, ``API_USGS_RETRIES``, ``API_USGS_PROGRESS``). +3. The configuration file (TOML): ``~/.dataretrieval/config.toml``, or the path + in ``DATARETRIEVAL_CONFIG``. Top-level keys are the defaults; a + ``[profiles.]`` table layers over them when that profile is selected. +4. The built-in default. + +Precedence applies **per setting**, not per source: an environment that sets only +``API_USGS_PAT`` leaves a file-provided ``concurrency`` fully in effect. Putting +the environment above the file follows common deployment conventions and keeps +the original environment-variable interface authoritative (see ADR 0006). + +This module is a leaf: it imports only the standard library plus the Python 3.10 +``tomli`` backport, so any module can depend on it without an import cycle or +pulling in httpx or pandas. It centralizes each setting's parser while retaining +legacy environment behavior and stricter validation for the new Python/TOML +surfaces. +""" + +from __future__ import annotations + +import os +import stat +import sys +import warnings +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass, field +from functools import partial +from numbers import Integral +from pathlib import Path +from types import MappingProxyType +from typing import Any, TextIO + +from dataretrieval.exceptions import ConfigError + +if sys.version_info >= (3, 11): + import tomllib +else: # pragma: no cover - exercised only on Python 3.10 + import tomli as tomllib + +# ``ConfigError`` is re-exported; its canonical home and rationale are in +# :mod:`dataretrieval.exceptions`. +__all__ = ["configure", "show_config", "config_path", "ConfigError"] + + +#: The settings this module resolves, in display order. +SETTINGS: tuple[str, ...] = ( + "api_key", + "concurrency", + "retries", + "progress", + "parallel_chunks", +) + +#: Environment variable backing a setting (precedence step 2). +#: +#: Not every setting has one. ``parallel_chunks`` is deliberately absent: it +#: fans a query into more sub-requests, each of which spends rate-limit quota, +#: and ``dataretrieval.parallel_chunks`` documents why that must stay a +#: deliberate choice rather than a process-wide default. An environment +#: variable is the wrong shape for it -- exported once in a shell profile, +#: inherited by every subprocess, invisible at the call site. A config-file +#: entry is written deliberately and shows up in :func:`show_config`, so the +#: file and :func:`configure` block are the only sources for it. +ENV_VARS: dict[str, str] = { + "api_key": "API_USGS_PAT", + "concurrency": "API_USGS_CONCURRENT", + "retries": "API_USGS_RETRIES", + "progress": "API_USGS_PROGRESS", +} + +#: Environment variable holding an explicit path to the configuration file. +CONFIG_PATH_ENV = "DATARETRIEVAL_CONFIG" + +#: Environment variable selecting a ``[profiles.]`` table. +PROFILE_ENV = "DATARETRIEVAL_PROFILE" + +#: TOML table holding the named profiles. +_PROFILES_TABLE = "profiles" + +#: Label for the file's top-level table, where keys are the defaults. +_TOP_LEVEL = "top level" + +# Built-in defaults (precedence step 4). ``concurrency`` and ``retries`` keep the +# values the environment-only implementation used, so behavior is unchanged for +# anyone who configures nothing. +DEFAULT_CONCURRENCY = 32 +DEFAULT_RETRIES = 4 +DEFAULT_PARALLEL_CHUNKS = 1 +CONCURRENCY_UNBOUNDED = "unbounded" + +# Values that turn the progress line off. Blank counts: ``API_USGS_PROGRESS=`` +# has always meant "off", not "unset" -- unlike the numeric knobs, where blank +# falls through to the default. +_PROGRESS_FALSEY = frozenset({"", "0", "false", "no", "off"}) + +# Settings for which a *blank* environment variable is a value rather than an +# absence. ``API_USGS_PROGRESS=`` has always meant "off". For every other +# setting a blank variable is what container and CI tooling produces when it +# has nothing to pass (``docker run -e API_USGS_PAT``, a workflow secret that +# is absent on a fork), so treating it as configured would let it shadow the +# config file and silently drop the user's API key. Keeping this a property of +# the setting -- rather than a second, lower visit to the environment -- keeps +# the chain at the three tiers the docstring and ADR 0006 describe. +_BLANK_MEANS_SET = frozenset({"progress"}) + +# Warnings about the config file report the file, not a call site: settings are +# resolved lazily from wherever a getter first needs one, so the user frame is +# a different depth every time and no fixed ``stacklevel`` can name it. Pointing +# at this module consistently at least makes the warnings filterable by module, +# and every message names the offending path and setting. +_WARN_STACKLEVEL = 2 +_PROGRESS_TRUTHY = frozenset({"1", "true", "yes", "on"}) + + +class _Unset: + """Sentinel that distinguishes an omitted override from explicit ``None``.""" + + __slots__ = () + + def __repr__(self) -> str: + return "" + + +# Typed as Any so public annotations describe accepted caller values without +# exposing this private implementation detail in generated signatures. +_UNSET: Any = _Unset() +_ConfigValue = str | None + +# Overrides from the innermost active ``configure`` block, as raw strings so that +# every source shares one parser and one set of error messages. The selected +# profile rides in the same mapping under ``_PROFILE_KEY`` -- it is not a +# setting, so it never resolves as one, but it inherits the same nesting and +# restore-on-exit for free. The default is an immutable empty mapping: a +# ``configure`` block always replaces the mapping wholesale rather than +# mutating it, and a read-only default makes that impossible to get wrong. +_PROFILE_KEY = "\0profile" +_NO_OVERRIDES: Mapping[str, _ConfigValue] = MappingProxyType({}) +_scope: ContextVar[Mapping[str, _ConfigValue]] = ContextVar( + "dataretrieval_config", default=_NO_OVERRIDES +) + +# Resolved config-file path, memoized on the raw ``DATARETRIEVAL_CONFIG`` +# value (see :func:`config_path`). +_path_cache: tuple[str | None, object | None, Path] | None = None + +# Parsed configuration file, keyed by file identity, change metadata, and raw +# content. POSIX ctime makes metadata hits reliable; Windows ctime is creation +# time, so cache hits there compare content before reusing the parsed result. +_FileStamp = tuple[int, int, int, int, int, int] +_file_cache: tuple[Path, _FileStamp, bytes, _ParsedFile] | None = None + +# Paths already warned about for loose permissions, so the warning fires once. +_permission_warned: set[Path] = set() + + +@dataclass(frozen=True) +class _ParsedFile: + """A parsed configuration file: top-level defaults plus named profiles. + + ``exists`` distinguishes "the file is there and defines nothing" from "there + is no file", which decides whether selecting an undefined profile is a typo + worth raising on (see :func:`_file_settings`). + """ + + base: dict[str, str] = field(default_factory=dict) + #: Raw, *unvalidated* TOML tables -- see :func:`_interpret`. + profiles: dict[str, dict[str, Any]] = field(default_factory=dict) + exists: bool = False + + +# --- public API ---------------------------------------------------------- + + +@contextmanager +def configure( + *, + api_key: str | None = _UNSET, + concurrency: int | str | None = _UNSET, + retries: int | None = _UNSET, + progress: bool | str | None = _UNSET, + parallel_chunks: int | None = _UNSET, + profile: str | None = _UNSET, +) -> Iterator[None]: + """Apply configuration for the duration of a ``with`` block. + + The highest-precedence source. Because it is backed by a + :class:`~contextvars.ContextVar`, a value set here applies to the current + thread and to asyncio tasks started inside the block, and cannot leak into + another thread, task, or unrelated call the way ``os.environ`` does -- + which is what makes it safe for a server or notebook handling several + users' credentials at once:: + + with dataretrieval.configure(api_key=secrets["usgs"]): + df, md = waterdata.get_daily(monitoring_location_id="USGS-05114000") + + Values are validated on entry, so a typo raises here rather than deep in a + later request. Blocks nest, and merge per setting -- an inner block that + sets only ``concurrency`` keeps the outer block's ``api_key``. + + Omitting a setting inherits it from an outer block or lower-precedence + source. Passing ``None`` explicitly suppresses those sources and restores + the built-in behavior for that setting (no key, automatic progress, and so + on). ``profile=None`` selects the file's top-level settings even when + ``DATARETRIEVAL_PROFILE`` is set. + + Parameters + ---------- + api_key : str, optional + Water Data API key, sent as ``X-Api-Key`` and only ever to + ``api.waterdata.usgs.gov``. Prefer reading it from a secret store or + the environment or configuration file over writing a literal into a + script. Pass ``None`` to make a call without an ambient key. + concurrency : int or str, optional + Cap on simultaneous sub-requests: a positive integer, or + ``"unbounded"`` to disable the cap. + retries : int, optional + Retries attempted after a transient failure; ``0`` disables retrying. + progress : bool or str, optional + Whether to draw the progress line. ``None`` leaves the automatic + behavior (on for a TTY or Jupyter kernel, off otherwise). + parallel_chunks : int, optional + Default optional fan-out for multi-value queries. It limits extra + refinement, but URL-byte safety may already require more sub-requests. + Sets the baseline that :func:`dataretrieval.parallel_chunks` overrides + per call. Each sub-request spends rate-limit quota, so raise it only + for pulls you know are large. + profile : str, optional + Name of a ``[profiles.]`` table in the configuration file to + layer over the file's top-level settings. Pass ``None`` to ignore an + environment-selected profile. + + Yields + ------ + None + + Examples + -------- + .. code-block:: python + + # credentials from a secret store, no environment mutation + with dataretrieval.configure(api_key=vault.read("usgs/pat")): + df, md = waterdata.get_daily(monitoring_location_id="USGS-05114000") + + # a big overnight pull, using a profile from the config file + with dataretrieval.configure(profile="bulk-pull"): + df, md = waterdata.get_daily(monitoring_location_id=many_sites) + + See Also + -------- + show_config : Report the effective configuration and where it came from. + """ + supplied = { + "api_key": api_key, + "concurrency": concurrency, + "retries": retries, + "progress": progress, + "parallel_chunks": parallel_chunks, + } + overrides = { + name: _normalize_override(name, value) + for name, value in supplied.items() + if value is not _UNSET + } + + # The selected profile rides in the same mapping as the settings, so + # nesting and per-key inheritance fall out of one merge. ``_PROFILE_KEY`` + # is not in ``SETTINGS``, so it is never resolved as one. + merged = {**_scope.get(), **overrides} + if profile is not _UNSET: + merged[_PROFILE_KEY] = _normalize_profile(profile) + token = _scope.set(merged) + try: + # An explicitly selected profile is a value supplied to this block, so + # validate its existence on entry rather than on a later request. + if profile is not _UNSET and profile is not None: + _file_settings() + yield + finally: + _scope.reset(token) + + +def show_config(*, stream: TextIO | None = None) -> None: + """Print the effective configuration and the source of each setting. + + A debugging aid for "why is this using my old key?". The API key is never + printed -- only whether one is set and where it came from. + + Parameters + ---------- + stream : file-like, optional + Where to write. Defaults to ``sys.stdout``. + + Examples + -------- + .. code-block:: text + + >>> dataretrieval.show_config() + config file /home/u/.dataretrieval/config.toml (found) + profile default + api_key /home/u/.dataretrieval/config.toml + concurrency 32 built-in default + retries 8 $API_USGS_RETRIES + progress auto built-in default + """ + out = sys.stdout if stream is None else stream + try: + path = config_path() + except ConfigError as exc: + # Resolution itself can fail (a relative override with the working + # directory removed). That is precisely a configuration a caller would + # run this to understand, so report it as the file row rather than + # raising out of the explainer. + print(f"config file ", file=out) + return + + # Nothing here raises. This function exists to explain a configuration, and + # the configurations most in need of explaining are the broken ones -- an + # unparseable file, a value that fails its grammar, a profile that no + # longer exists. Each distinct failure is printed once, in the first place + # it shows up; a repeat is collapsed, so one bad file does not bury the + # rows that did resolve under ten copies of the same message. + reported: str | None = None + + def cell(render: Callable[[], object]) -> str: + nonlocal reported + try: + value = render() + except ConfigError as exc: + if str(exc) == reported: + return "" + reported = str(exc) + return f"" + return "" if value is None else str(value) + + # Probing the whole file layer (not just the parse) means a bad profile -- + # which ``_file_settings`` raises, not ``_load_file`` -- is also reported + # here rather than in all five rows. + try: + _file_settings() + status = "found" if path.exists() else "not found" + except ConfigError as exc: + reported = str(exc) + status = f"ERROR: {exc}" + print(f"config file {path} ({status})", file=out) + print(f"profile {_active_profile() or 'default'}", file=out) + + rows = [ + (name, cell(_DISPLAYS[name]), cell(partial(_source_label, name))) + for name in SETTINGS + ] + name_width = max(len(name) for name, _value, _source in rows) + value_width = max(len(value) for _name, value, _source in rows) + for name, value, source in rows: + print(f"{name:<{name_width}} {value:<{value_width}} {source}", file=out) + + +def _source_label(name: str) -> str: + """The provenance label for one setting, for :func:`show_config`.""" + return _resolve(name)[1] + + +def config_path() -> Path: + """Path to the configuration file, honoring ``DATARETRIEVAL_CONFIG``. + + Memoized on the raw ``DATARETRIEVAL_CONFIG`` value, because this sits on + the per-request path via :func:`api_key` and building the default costs + more than the ``stat`` it leads to (``Path.home()`` alone dominates the + whole resolution). Returning a stable object also lets :func:`_load_file` + check its cache by identity instead of re-normalizing a fresh ``Path``. + + Returns + ------- + pathlib.Path + The explicit path from ``DATARETRIEVAL_CONFIG`` if set, otherwise + ``~/.dataretrieval/config.toml``. The file need not exist. + """ + global _path_cache + override = os.environ.get(CONFIG_PATH_ENV) + + # Probe the memo before doing any work: this runs once per request via + # ``api_key()``, so the hit path should be a dict lookup and a compare. + cached = _path_cache + if cached is not None and cached[0] == override: + cached_guard, path = cached[1], cached[2] + # The memo is only valid while whatever the path was *derived from* is + # unchanged, so each branch records its own guard. A relative override + # is anchored to the working directory (a later ``os.chdir`` in a + # per-job notebook or scheduler must not keep reading the previous + # job's file); the default branch is anchored to ``$HOME``. An absolute + # override depends on neither and guards with ``None``. ``stat(".")`` + # identifies the directory ~17x cheaper than ``getcwd()``, which + # reifies the whole path string. + if cached_guard is None or cached_guard == _path_guard(cached_guard): + return path + + expanded = ( + Path(override.strip()).expanduser() if override and override.strip() else None + ) + guard: object | None + if expanded is None: + path = _default_home_path() + guard = _home_id() + elif expanded.is_absolute(): + path = expanded + guard = None + else: + guard = _cwd_id() + path = _resolve_against_cwd(expanded) + _path_cache = (override, guard, path) + return path + + +def _default_home_path() -> Path: + """The default ``~/.dataretrieval/config.toml``, or an unusable path. + + ``Path.home()`` raises ``RuntimeError`` where no home can be resolved at all + -- a rootless container running as an arbitrary UID with no passwd entry and + no ``HOME``. That is not a misconfiguration to report: such a deployment + simply has no config file, and before settings were layered it worked fine + on the environment alone. So the unexpanded ``~/...`` form is returned + instead: it does not exist, which keeps the whole file layer inert rather + than failing every request from inside the header builder, and it still + reads correctly in :func:`show_config` output. + """ + try: + home = Path.home() + except (RuntimeError, OSError): + return Path("~") / ".dataretrieval" / "config.toml" + return home / ".dataretrieval" / "config.toml" + + +def _resolve_against_cwd(relative: Path) -> Path: + """Resolve a relative override, or report a working directory that is gone. + + A scratch-dir job that removes its own cwd cannot resolve a relative + ``DATARETRIEVAL_CONFIG`` at all. That surfaces as a :class:`ConfigError` + rather than a bare ``OSError`` escaping onto the request path -- the + taxonomy contract the rest of this module keeps. + """ + try: + return Path.cwd() / relative + except OSError as exc: + raise ConfigError( + f"cannot resolve the relative {CONFIG_PATH_ENV} path {str(relative)!r}: " + f"the working directory is unavailable ({exc})." + ) from exc + + +def _path_guard(previous: object) -> object: + """Re-read whichever guard the cached entry was built with.""" + return _cwd_id() if isinstance(previous, tuple) else _home_id() + + +def _cwd_id() -> tuple[int, int]: + """Identify the working directory without building its path string. + + Only identifies the directory; :func:`_resolve_against_cwd` is what turns a + missing cwd into a :class:`ConfigError`. Both are needed, because ``stat`` + on a *deleted* working directory still succeeds -- the process holds the + open handle -- while resolving its path does not. + """ + try: + st = os.stat(".") + except OSError as exc: + raise ConfigError( + f"cannot resolve the relative {CONFIG_PATH_ENV} path: the working " + f"directory is unavailable ({exc})." + ) from exc + return (st.st_dev, st.st_ino) + + +def _home_id() -> str: + """The home directory as the environment reports it. + + A plain environment read, not ``Path.home()``: this is on the per-request + path and only needs to detect a *change* (a test or notebook that + reassigns ``HOME`` after the first resolution), not to resolve the path. + """ + return os.environ.get("HOME") or os.environ.get("USERPROFILE") or "" + + +# --- resolved settings --------------------------------------------------- + + +def api_key() -> str | None: + """The Water Data API key, or ``None`` if none is configured. + + Surrounding whitespace is stripped, so a key read from a file with a + trailing newline works; a blank value resolves to ``None``. + """ + raw, _source = _resolve("api_key") + return raw.strip() or None if raw is not None else None + + +def concurrency() -> int | None: + """Cap on simultaneous sub-requests; ``None`` means unbounded.""" + raw, source = _resolve("concurrency") + if raw is None: + return DEFAULT_CONCURRENCY + return _parse_concurrency(raw, source) + + +def retries() -> int: + """Retries attempted after the first try; ``0`` disables retrying.""" + raw, source = _resolve("retries") + if raw is None: + return DEFAULT_RETRIES + return _parse_int(raw, source, default=DEFAULT_RETRIES, minimum=0) + + +def progress() -> bool | None: + """Explicit progress-line setting, or ``None`` to auto-detect. + + ``None`` means nothing configured it, so the caller applies its own + default (a TTY or Jupyter kernel gets the line, redirected output + doesn't). + """ + raw, source = _resolve("progress") + if raw is None: + return None + # Preserve the legacy environment behavior (any value outside the false + # set enables progress), while new block/file values are validated strictly. + strict = source != f"${ENV_VARS['progress']}" + return _parse_progress(raw, source, strict=strict) + + +def parallel_chunks() -> int: + """Configured default fan-out for multi-value queries. + + ``1`` (the default) means "chunk only as much as the URL byte limit + forces". This is the *baseline*; + :func:`dataretrieval.parallel_chunks` overrides it for one call. Shares + the name of that context manager because it is the same setting -- this + is the resolved value, not the scoping block. + """ + raw, source = _resolve("parallel_chunks") + if raw is None: + return DEFAULT_PARALLEL_CHUNKS + return _parse_int( + raw, source, default=DEFAULT_PARALLEL_CHUNKS, minimum=1, examples="2, 8, 32" + ) + + +# --- value grammar ------------------------------------------------------- +# +# One parser drives each setting's grammar, so a value means the same thing and +# reports the same way whichever source wrote it. Source-level adapters retain +# TOML types and reject Python API type errors before producing raw strings. + + +def _type_error(source: str, expected: str, value: object) -> ConfigError: + """Build a type error without rendering a possibly secret value.""" + return ConfigError(f"{source} must be {expected} (got {type(value).__name__}).") + + +def _coerce_typed(name: str, value: object, source: str, *, optional: str = "") -> str: + """Type-check one source-level value and render it as a raw string. + + Shared by the two *typed* surfaces -- ``configure()`` keyword arguments and + TOML scalars -- so a value accepted from one is accepted from the other and + a tightened rule cannot land on only half of them. (The environment is not + typed: it delivers strings, which go straight to :func:`_validate_raw`.) + + ``optional`` is the only thing that differs between them: the Python + surface accepts ``None`` and says so in its messages. (Integers are matched + as :class:`numbers.Integral` for both -- a numpy or pandas integer is a + legitimate count from Python, and ``tomllib`` only ever yields ``int``, so + the wider check cannot change a TOML outcome.) + """ + if name == "api_key": + if not isinstance(value, str): + raise _type_error(source, "a string" + optional, value) + return value + if name == "progress": + if isinstance(value, bool): + return str(value) + if isinstance(value, str): + return value + raise _type_error(source, "a bool or recognized string" + optional, value) + if name == "concurrency": + if isinstance(value, bool) or not isinstance(value, (Integral, str)): + raise _type_error(source, "an integer or 'unbounded'" + optional, value) + if isinstance(value, str) and value.strip().lower() != CONCURRENCY_UNBOUNDED: + raise ConfigError(f"{source} must be an integer or 'unbounded'.") + return str(value) + if isinstance(value, bool) or not isinstance(value, Integral): + raise _type_error(source, "an integer" + optional, value) + return str(value) + + +def _normalize_override(name: str, value: object) -> _ConfigValue: + """Validate and normalize one value supplied to :func:`configure`.""" + if value is None: + return None + source = f"{name}= in configure()" + raw = _coerce_typed(name, value, source, optional=", or None") + _validate_raw(name, raw, source) + return raw + + +def _normalize_profile(value: object) -> str | None: + """Validate and normalize a profile supplied to :func:`configure`.""" + if value is None: + return None + if not isinstance(value, str): + raise _type_error( + "profile= in configure()", "a non-empty string or None", value + ) + profile = value.strip() + if not profile: + raise ConfigError("profile= in configure() must not be blank.") + return profile + + +def _parse_int( + raw: str, + source: str, + *, + default: int, + minimum: int, + examples: str | None = None, +) -> int: + """Parse a bounded integer setting; blank falls through to *default*. + + Parameters + ---------- + raw : str + The value as written, from whichever source supplied it. + source : str + Human-readable origin, used as the subject of any error message. + default : int + Returned for a blank value, matching the environment-variable + behavior this replaced. + minimum : int + Smallest accepted value. + examples : str, optional + Illustrative values appended to the message (e.g. ``"2, 8, 32"``). + """ + value = raw.strip() + if value == "": + return default + expected = f"an integer >= {minimum}" + (f", e.g. {examples}" if examples else "") + try: + parsed = int(value) + except ValueError as exc: + raise ConfigError(f"{source} must be {expected} (got {raw!r}).") from exc + if parsed < minimum: + raise ConfigError(f"{source} must be {expected} (got {parsed}).") + return parsed + + +def _parse_concurrency(raw: str, source: str) -> int | None: + """Parse a concurrency cap: a positive int, or ``unbounded`` -> ``None``.""" + if raw.strip().lower() == CONCURRENCY_UNBOUNDED: + return None + try: + return _parse_int(raw, source, default=DEFAULT_CONCURRENCY, minimum=1) + except ConfigError as exc: + raise ConfigError( + f"{exc} Use '{CONCURRENCY_UNBOUNDED}' to disable the cap." + ) from exc + + +def _parse_progress(raw: str, source: str, *, strict: bool) -> bool: + """Parse a progress toggle, optionally preserving legacy env truthiness.""" + value = raw.strip().lower() + if strict and not value: + raise ConfigError(f"{source} must not be blank.") + if value in _PROGRESS_FALSEY: + return False + if value in _PROGRESS_TRUTHY: + return True + if not strict: + return True + expected = ", ".join(sorted(_PROGRESS_TRUTHY | _PROGRESS_FALSEY)) + raise ConfigError(f"{source} must be one of {expected} (got {raw!r}).") + + +#: Per-setting validators used for eager block and TOML validation. +_VALIDATORS: dict[str, Callable[[str, str], object]] = { + "concurrency": _parse_concurrency, + "retries": lambda raw, source: _parse_int( + raw, source, default=DEFAULT_RETRIES, minimum=0 + ), + "progress": lambda raw, source: _parse_progress(raw, source, strict=True), + "parallel_chunks": lambda raw, source: _parse_int( + raw, source, default=DEFAULT_PARALLEL_CHUNKS, minimum=1, examples="2, 8, 32" + ), +} + + +def _validate_raw(name: str, raw: str, source: str) -> None: + """Run a setting's grammar validator when it has one.""" + validate = _VALIDATORS.get(name) + if validate is not None: + validate(raw, source) + + +# --- resolution ---------------------------------------------------------- + + +def _resolve(name: str) -> tuple[str | None, str]: + """Return the raw value for *name* and a human-readable source label. + + Returns + ------- + tuple[str or None, str] + The raw string as written (parsing happens per setting, so each keeps + its own blank-value rule), and where it came from -- ``None`` with + ``"built-in default"`` when nothing configured it. + """ + scope = _scope.get() + if name in scope: + return scope[name], "configure() block" + + env = ENV_VARS.get(name) + if env is not None: + raw = os.environ.get(env) + if raw is not None and (raw.strip() or name in _BLANK_MEANS_SET): + return raw, f"${env}" + + from_file = _file_settings() + if name in from_file: + return from_file[name] + + return None, "built-in default" + + +def _active_profile() -> str | None: + """The selected profile name: a :func:`configure` block wins over the env.""" + scope = _scope.get() + if _PROFILE_KEY in scope: + return scope[_PROFILE_KEY] + env = os.environ.get(PROFILE_ENV) + return env.strip() if env and env.strip() else None + + +def _file_settings() -> Mapping[str, tuple[str, str]]: + """File-sourced settings, each with a label naming exactly where it came from. + + A selected ``[profiles.]`` table layers over the file's top-level + keys per setting, so a profile that only tunes ``concurrency`` still + inherits the top-level ``api_key`` -- and each value's label names the + table it actually came from, not merely the profile in effect. + + Selecting a profile the file doesn't define is a typo, and raises. But + with *no config file at all* there are no profiles to select from and the + whole file layer is inert, so a lingering ``DATARETRIEVAL_PROFILE`` export + is ignored rather than failing every request from inside + :func:`~dataretrieval.utils._default_headers`. + """ + path = config_path() + parsed = _load_file(path) + merged: dict[str, tuple[str, str]] = { + name: (value, str(path)) for name, value in parsed.base.items() + } + + profile = _active_profile() + if profile is None: + return merged + if profile not in parsed.profiles: + # With no file at all there are no profiles to select from. A lingering + # DATARETRIEVAL_PROFILE export is then ignored rather than failing every + # request -- but a name the caller just typed into configure() is a typo + # worth reporting at the ``with``, which is what its docstring promises. + if not parsed.exists and _PROFILE_KEY not in _scope.get(): + return merged + if not parsed.exists: + raise ConfigError( + f"profile {profile!r} cannot be selected: there is no " + f"configuration file at {path}." + ) + raise ConfigError( + f"profile {profile!r} is not defined in {path} " + f"(add a [{_PROFILES_TABLE}.{profile}] table)." + ) + label = f"{path} [{_PROFILES_TABLE}.{profile}]" + selected = _scalars( + parsed.profiles[profile], path, f"[{_PROFILES_TABLE}.{profile}]" + ) + merged.update({name: (value, label) for name, value in selected.items()}) + return merged + + +def _load_file(path: Path) -> _ParsedFile: + """Parse the configuration file at *path*, caching until it changes on disk.""" + global _file_cache + try: + st = path.stat() + except FileNotFoundError: + # No file is the normal case: continue to the built-in default. + return _ParsedFile() + except OSError as exc: + raise ConfigError(f"could not access {path}: {exc}") from exc + + if stat.S_ISDIR(st.st_mode): + raise ConfigError(f"configuration path {path} is a directory, not a file.") + + # Only a regular file is parsed. Anything else readable -- a character + # device, a FIFO -- is treated as *empty* configuration without being + # opened, which is what ``DATARETRIEVAL_CONFIG=/dev/null`` asks for and the + # only coherent answer for a stream: settings are re-resolved on every + # request, so a FIFO would hand its contents to the first getter and + # nothing to the rest, making the API key vanish mid-run. (It would also + # block on open until a writer appeared.) + if not stat.S_ISREG(st.st_mode): + return _ParsedFile(exists=True) + + # POSIX ``st_ctime_ns`` advances on any inode change, so the metadata stamp + # catches even a rewrite that restores the original mtime (``cp -p``, rsync + # ``--times``, an editor that preserves timestamps). Windows ctime is + # *creation* time, so there the stamp cannot see that class of edit and the + # content compare below is the only correct check -- worth the re-read, + # since serving a stale API key is the alternative. + # + # Dropping this gate (or dropping ctime from the stamp so Windows can use + # it) has been proposed repeatedly on the grounds that the re-read is + # wasteful. It is, but it is also the only thing standing between a + # timestamp-preserving write and a stale credential; a ctime-less stamp is + # identical across exactly that edit. ``test_file_edit_is_picked_up`` + # pins the behavior. Please do not "optimize" it without a Windows-safe + # change detector. + cached = _file_cache + if ( + os.name != "nt" + and cached is not None + and cached[0] is path + and cached[1] == _file_stamp(st) + ): + return cached[3] + + try: + with path.open("rb") as handle: + content = handle.read() + opened_st = os.fstat(handle.fileno()) + except OSError as exc: + raise ConfigError(f"could not read {path}: {exc}") from exc + + if cached is not None and cached[0] is path and cached[2] == content: + parsed = cached[3] + else: + try: + data = tomllib.loads(content.decode("utf-8")) + except UnicodeDecodeError as exc: + raise ConfigError(f"{path} is not valid UTF-8: {exc}") from exc + except tomllib.TOMLDecodeError as exc: + raise ConfigError(f"{path} is not valid TOML: {exc}") from exc + parsed = _interpret(data, path) + _warn_on_loose_permissions(path, opened_st, parsed) + _file_cache = (path, _file_stamp(opened_st), content, parsed) + return parsed + + +def _file_stamp(st: os.stat_result) -> _FileStamp: + """Metadata that changes with file replacement, content, or permissions.""" + return ( + st.st_dev, + st.st_ino, + st.st_mode, + st.st_size, + st.st_mtime_ns, + st.st_ctime_ns, + ) + + +def _interpret(data: dict[str, Any], path: Path) -> _ParsedFile: + """Validate a parsed TOML document into defaults plus profiles. + + Only the top-level table is validated here, because it always applies. + Profile tables are kept raw and validated in :func:`_file_settings` when + one is actually selected: a bad value in a profile nobody asked for must + not fail every request, the same blast-radius rule + :func:`~dataretrieval.utils._default_headers` follows for the key itself. + """ + top: dict[str, Any] = {} + profiles: dict[str, dict[str, Any]] = {} + + for key, value in data.items(): + if key == _PROFILES_TABLE: + if not isinstance(value, dict): + raise ConfigError( + f"{path}: [{_PROFILES_TABLE}] must be a table of profiles." + ) + for name, table in value.items(): + if not isinstance(table, dict): + raise ConfigError( + f"{path}: [{_PROFILES_TABLE}.{name}] must be a table." + ) + profiles[name] = table + continue + if isinstance(value, dict): + raise ConfigError( + f"{path}: unknown table [{key}]. Named profiles go under " + f"[{_PROFILES_TABLE}.{key}]; top-level keys are the defaults." + ) + top[key] = value + + return _ParsedFile(_scalars(top, path, _TOP_LEVEL), profiles, exists=True) + + +def _scalars(table: dict[str, Any], path: Path, where: str) -> dict[str, str]: + """Validate and normalize one table's recognized settings. + + ``tomllib`` returns typed scalars (``concurrency = 32`` is an ``int``, + ``concurrency = "unbounded"`` a ``str``), so types are checked here before + values pass through the same grammar used by the other sources. + Unrecognized keys warn rather than raise, so a file written for a newer + release still works. + """ + out: dict[str, str] = {} + for key, value in table.items(): + if key not in SETTINGS: + warnings.warn( + f"{path}: unknown setting {key!r} at {where} (ignored). " + f"Known settings: {', '.join(SETTINGS)}.", + UserWarning, + stacklevel=2, + ) + continue + if key == "parallel_chunks" and where == _TOP_LEVEL: + # The one setting that spends rate-limit quota, so a value left + # here applies to every splittable query in every process that + # reads the file. A profile is opt-in per run, which is the shape + # this setting wants. + warnings.warn( + f"{path}: 'parallel_chunks' at {where} applies to every query " + "in every process and spends rate-limit quota. Prefer a " + f"[{_PROFILES_TABLE}.] table selected per run, or the " + "dataretrieval.parallel_chunks(n) block for a single call.", + UserWarning, + stacklevel=_WARN_STACKLEVEL, + ) + source = f"{path}: {key!r} at {where}" + raw = _coerce_typed(key, value, source) + _validate_raw(key, raw, source) + out[key] = raw + return out + + +def _warn_on_loose_permissions( + path: Path, st: os.stat_result, parsed: _ParsedFile +) -> None: + """Warn once if a file holding an API key is readable by other users. + + Follows the ``~/.ssh`` and ``.netrc`` convention, but warns rather than + refusing -- shared filesystems on HPC clusters have their own conventions, + and refusing to read would strand those users. + """ + if os.name != "posix" or path in _permission_warned: + return + holds_key = "api_key" in parsed.base or any( + "api_key" in table for table in parsed.profiles.values() + ) + if not holds_key: + return + if stat.S_IMODE(st.st_mode) & 0o077: + _permission_warned.add(path) + warnings.warn( + f"{path} contains an API key and is readable by other users. " + f"Restrict it with: chmod 600 {path}", + UserWarning, + stacklevel=_WARN_STACKLEVEL, + ) + + +def _display_api_key() -> str: + """Render the key's presence, never its value.""" + return "" if api_key() else "" + + +def _display_concurrency() -> str: + value = concurrency() + return CONCURRENCY_UNBOUNDED if value is None else str(value) + + +def _display_progress() -> str: + setting = progress() + return "auto" if setting is None else ("on" if setting else "off") + + +#: How each setting renders in :func:`show_config`. Keyed by the same names as +#: :data:`SETTINGS`, and asserted to cover them, so a setting added to one +#: without the other fails loudly instead of silently printing a neighbour's +#: value in the one report whose whole job is to be trustworthy. +_DISPLAYS: dict[str, Callable[[], str]] = { + "api_key": _display_api_key, + "concurrency": _display_concurrency, + "retries": lambda: str(retries()), + "progress": _display_progress, + "parallel_chunks": lambda: str(parallel_chunks()), +} + +if set(_DISPLAYS) != set(SETTINGS): # pragma: no cover - guards a coding error + # Not an ``assert``: ``python -O`` strips those, and this guards the one + # report whose whole job is to be trustworthy about provenance. + raise RuntimeError( + "every setting needs a show_config renderer; " + f"missing={sorted(set(SETTINGS) - set(_DISPLAYS))} " + f"extra={sorted(set(_DISPLAYS) - set(SETTINGS))}" + ) + + +def _reset_file_cache() -> None: + """Drop the parsed-file cache. For tests that rewrite the file in place.""" + global _file_cache, _path_cache + _file_cache = None + _path_cache = None + _permission_warned.clear() diff --git a/dataretrieval/exceptions.py b/dataretrieval/exceptions.py index fefb62c5..cbe20bfd 100644 --- a/dataretrieval/exceptions.py +++ b/dataretrieval/exceptions.py @@ -12,6 +12,9 @@ aren't a plain status: :class:`RequestTooLarge` (with :class:`URLTooLong` / :class:`Unchunkable`), :class:`NetworkError` (a failed connection, per above), and :class:`NoSitesError`. :func:`error_for_status` maps a status to its type. +:class:`ConfigError` is the one member that is not a request failure at all -- +it reports unusable configuration, raised from wherever a setting is first +resolved. This module has no third-party runtime dependencies -- ``httpx`` is imported only for type checking -- so any module can import it without pulling in pandas / httpx @@ -27,6 +30,7 @@ __all__ = [ "DataRetrievalError", + "ConfigError", "HTTPError", "TransientError", "RateLimited", @@ -41,7 +45,15 @@ class DataRetrievalError(Exception): - """Base class for every failed-request error in ``dataretrieval``. + """Base class for every ``dataretrieval`` error. + + Almost every member is a failed request, and the read-anywhere fields below + describe one. The exception is :class:`ConfigError`, which reports a + configuration the library cannot use; it appears here because configuration + is resolved lazily on the request path, so it surfaces from inside a getter + and one ``except DataRetrievalError`` should cover it too. It carries no + status and is not retryable, so the branching idiom below routes it to the + final ``raise``. Catch it to handle any USGS or EPA service failure uniformly, and branch on the read-anywhere fields below without needing the concrete subclass:: @@ -97,6 +109,27 @@ def _new_error(cls: type[DataRetrievalError]) -> DataRetrievalError: return cls.__new__(cls) +# --- Configuration ------------------------------------------------------- + + +class ConfigError(DataRetrievalError, ValueError): + """A configuration value or file could not be used. + + Raised by :mod:`dataretrieval.config` for a malformed + ``~/.dataretrieval/config.toml``, a value that fails its grammar, or a + profile that the file does not define. + + It lives in the taxonomy -- rather than being a bare :class:`ValueError` -- + because configuration is resolved lazily, on the request path: a broken + config file surfaces from inside whichever getter runs first, so + ``except DataRetrievalError`` around a call must catch it like any other + failure of that call. It is *also* a :class:`ValueError`, so the + ``except ValueError`` that used to wrap the environment-variable knobs + keeps working whether the value came from the environment, a file, or a + :func:`dataretrieval.configure` block. + """ + + # --- HTTP status errors -------------------------------------------------- diff --git a/dataretrieval/ogc/chunking.py b/dataretrieval/ogc/chunking.py index 79037a6b..0453b3f2 100644 --- a/dataretrieval/ogc/chunking.py +++ b/dataretrieval/ogc/chunking.py @@ -12,8 +12,8 @@ Parallel chunks: the planner is conservative by default — it splits only as far as the byte limit forces. A caller who knows their result is large can opt into a finer split via the ``parallel_chunks(n)`` context manager, which fans the query -out into ``n`` parallel sub-requests; ``n`` drives :meth:`ChunkPlan._refine`. See -``parallel_chunks`` for the why and the when. +out toward ``n`` sub-requests; ``n`` drives :meth:`ChunkPlan._refine`. See +``parallel_chunks`` for the planning/execution distinction and when to use it. This module owns the *execution* half — the event loop and bounded concurrency that drive a plan to completion (``ChunkedCall``) plus the @@ -32,9 +32,8 @@ ``asyncio.Semaphore`` — not the client's connection pool, which is merely sized to match — caps the sub-requests in flight at ``N``; see :meth:`ChunkedCall._run` for why the gate must be the semaphore rather -than the pool. ``API_USGS_CONCURRENT`` resolves ``N``: an integer N > 1 -allows N sub-requests in flight; ``1`` forces sequential dispatch (one -request at a time); the literal ``unbounded`` lifts the cap. ``N`` +than the pool. The effective ``concurrency`` setting resolves ``N`` (see the +configuration guide for its values and sources). ``N`` bounds only how many of a chunked query's sub-requests are in flight at once — a client-side trade-off between open connections and fan-out latency. It does not affect the API rate limit: a chunked call issues @@ -48,9 +47,10 @@ Retries: each sub-request is retried on a transient failure (429, 5xx, connect/read timeout) with exponential backoff + full jitter, -honoring a server ``Retry-After`` when present. ``API_USGS_RETRIES`` -sets the cap (default 4; ``0`` disables). A ``Retry-After`` longer -than the per-call ceiling escalates to a resumable interruption. +honoring a server ``Retry-After`` when present. The effective ``retries`` +setting (with ``API_USGS_RETRIES`` as its environment source) sets the cap +(default 4; ``0`` disables). A ``Retry-After`` longer than the per-call +ceiling escalates to a resumable interruption. Interruption: any mid-stream transient failure — 429, 5xx, or a bare transport error (connect/read timeout, oversize follow-up URL) — surfaces @@ -73,7 +73,6 @@ import asyncio import functools -import os from collections.abc import Awaitable, Callable, Iterator from contextlib import contextmanager from contextvars import copy_context @@ -83,6 +82,7 @@ import pandas as pd from anyio.from_thread import start_blocking_portal +import dataretrieval.config as _config from dataretrieval.utils import HTTPX_ASYNC_DEFAULTS, Ambient, _require_positive_int from . import progress as _progress @@ -109,49 +109,6 @@ _OGC_URL_BYTE_LIMIT = 8000 -# Fan-out concurrency cap, read at call time (not import) so test -# ``monkeypatch.setenv`` applies. Value grammar in :func:`_read_concurrency_env`; -# the concurrency model is in the module docstring. -_CONCURRENCY_ENV = "API_USGS_CONCURRENT" -_CONCURRENCY_DEFAULT = 32 -_CONCURRENCY_UNBOUNDED = "unbounded" - - -def _read_concurrency_env() -> int | None: - """ - Resolve the ``API_USGS_CONCURRENT`` env var to a parallelism cap. - - Returns - ------- - int or None - ``1`` for sequential dispatch (one sub-request at a time); an - integer >1 for bounded concurrency; ``None`` to disable the - per-call cap entirely (``unbounded`` keyword). Unset → default - of ``_CONCURRENCY_DEFAULT``. - """ - raw = os.environ.get(_CONCURRENCY_ENV) - if raw is None: - return _CONCURRENCY_DEFAULT - raw = raw.strip() - if raw == "": - return _CONCURRENCY_DEFAULT - if raw.lower() == _CONCURRENCY_UNBOUNDED: - return None - try: - value = int(raw) - except ValueError as exc: - raise ValueError( - f"{_CONCURRENCY_ENV} must be a positive integer or " - f"'{_CONCURRENCY_UNBOUNDED}'; got {raw!r}." - ) from exc - if value < 1: - raise ValueError( - f"{_CONCURRENCY_ENV} must be >= 1 (got {value}); use " - f"'{_CONCURRENCY_UNBOUNDED}' to disable the cap." - ) - return value - - # Shared per-call ``httpx.AsyncClient``, scoped via ``with _chunked_client(c):`` # during ``ChunkedCall._run`` so paginated-loop helpers (``_walk_pages``) reuse # the same connection pool across every sub-request. ``None`` outside a chunked @@ -176,19 +133,10 @@ def get_active_client() -> httpx.AsyncClient | None: return _chunked_client.get() -# Parallel-chunks dial: opt-in to fan a query out *more finely* than the byte -# limit alone requires. Scoped to a ``with parallel_chunks(...):`` block (a -# ContextVar), deliberately NOT an env var (see :func:`parallel_chunks` for -# why). The ambient holds ``n`` — the requested cap on the plan's total -# sub-request count; ``1`` (the default, outside any block) means "off — chunk -# only as much as the byte limit needs, no extra fan-out". -_parallel_chunks: Ambient[int] = Ambient("ogc_parallel_chunks", 1) - - @contextmanager def parallel_chunks(n: int) -> Iterator[None]: """ - Fan the OGC getters' multi-value requests out into ``n`` parallel sub-requests. + Ask the OGC planner for up to ``n`` optional chunks for a multi-value request. By default the Water Data / NGWMN getters chunk a request only as much as the server's ~8 KB URL-byte limit forces — the fewest sub-requests that @@ -209,31 +157,36 @@ def parallel_chunks(n: int) -> Iterator[None]: would only burn quota), this is a *deliberate* per-call knob rather than an automatic behavior or a process-wide environment variable — scoping it to a ``with`` block keeps an aggressive setting from leaking into unrelated calls - and accidentally spending quota. Outside any block the getters use the - conservative default. Only the OGC getters (Water Data, NGWMN) read this; - wrapping a legacy NWIS call in the block is a harmless no-op. + and accidentally spending quota. For the same reason it is the one setting + :mod:`dataretrieval.config` exposes with no environment variable; outside + any block the getters use the configured baseline, which is the + conservative default of ``1`` unless a config file or a + ``dataretrieval.configure`` block raised it. Only the OGC getters (Water + Data, NGWMN) read this; wrapping a legacy NWIS call in the block is a + harmless no-op. + + This is sugar for ``dataretrieval.configure(parallel_chunks=n)`` — one + scoping mechanism, so the innermost block wins whichever of the two forms + wrote it, and :func:`dataretrieval.show_config` always reports the value + the chunker will actually use. Parameters ---------- n : int - The number of sub-requests to fan the whole call out into — a positive - integer such as ``2``, ``8``, or ``32``. It caps the plan's *total* - sub-request count (the cartesian product across every multi-value - argument combined, not per argument), so several multi-value arguments - cannot multiply past it. The cap is a ceiling, never exceeded: the - actual count is bounded below by what the ~8 KB URL limit already - forces and above by ``n``, so an ``n`` larger than the input allows - simply yields one sub-request per value, and with several multi-value - arguments the total may land somewhat below ``n`` because splits are - whole (the plan can't always divide evenly onto ``n``); ``n=1`` asks - for no extra fan-out. + Soft ceiling for optional refinement of the whole call — a positive + integer such as ``2``, ``8``, or ``32``. URL-byte safety is planned + first, so a request that already requires more than ``n`` sub-requests + keeps that mandatory plan. Otherwise optional splits approach ``n`` + without crossing it. Indivisible inputs and cartesian products can + leave the result below ``n``; ``n=1`` asks for no extra fan-out. Each sub-request fetches at least one page, so it costs at least one request against your hourly rate limit — a larger ``n`` spends more quota. And because how many sub-requests run *at once* is capped - separately by ``API_USGS_CONCURRENT`` (default 32), an ``n`` beyond that - adds quota without adding parallelism; the useful range is roughly ``2`` - up to ``API_USGS_CONCURRENT``. + separately by the effective ``concurrency`` setting + (``API_USGS_CONCURRENT``), an ``n`` beyond that adds quota without + adding parallelism; the useful range is roughly ``2`` up to the + concurrency cap. Yields ------ @@ -283,8 +236,12 @@ def parallel_chunks(n: int) -> Iterator[None]: """ # Fail loudly on a bad ``n`` at ``with`` entry, before any request. Shared # rules with ``max_rows`` via the helper (accepts numpy ints, rejects bool). + # Validate here rather than leaving it to the config parser: this is the + # Python API, so it is strict about *type* (a float, a bool, or even the + # numeric string "8" is a usage error), while TOML and environment values + # are normalized through the shared configuration grammar. _require_positive_int(n, "parallel_chunks(n)", examples="2, 8, 32") - with _parallel_chunks(n): + with _config.configure(parallel_chunks=n): yield @@ -563,11 +520,36 @@ def resume(self) -> tuple[pd.DataFrame, Any]: # active when the call was created reach the rebuilt sub-requests, # even when this is a resume fired long after the original ``with`` # blocks exited. - return self._ctx.run(self._resume_in_context) - - def _resume_in_context(self) -> tuple[pd.DataFrame, Any]: + # The concurrency cap is read *outside* the snapshot, from the caller's + # live context. It is a client-side dial, and the documented way to + # recover from a QuotaExhausted is to wait and retry more gently: + # + # with dataretrieval.configure(concurrency=2): + # exc.call.resume() + # + # Resolving it inside the snapshot would silently ignore that block + # (a ContextVar set after construction is invisible to a context copied + # before it), while the equivalent ``API_USGS_CONCURRENT`` export -- + # read from ``os.environ``, which snapshots don't capture -- kept + # working, so the two spellings of one setting disagreed. Everything + # else the rebuilt sub-requests need (base URL, dialect, row cap, + # progress reporter, and the credentials this call started with) still + # comes from the snapshot. + # + # The credential is deliberately *not* re-resolved: a resume continues + # the same logical call, and the documented recovery re-issues it after + # the block that authorized it may well have exited. Note the resulting + # asymmetry, which cannot be removed from this side: a + # ``configure(api_key=...)`` value is pinned here, while + # ``API_USGS_PAT`` is read live from ``os.environ`` on every rebuilt + # sub-request, because a context snapshot does not capture the + # environment. Rotating a key mid-call therefore takes effect for the + # environment spelling and not for the block spelling. + concurrency = _config.concurrency() + return self._ctx.run(self._resume_in_context, concurrency) + + def _resume_in_context(self, concurrency: int | None) -> tuple[pd.DataFrame, Any]: """Body of :meth:`resume`, run inside the captured context.""" - concurrency = _read_concurrency_env() with start_blocking_portal() as portal: # ``portal.call`` returns ``Any`` because ``functools.partial`` # erases ``_run``'s return type; restore the declared tuple. @@ -732,9 +714,9 @@ def multi_value_chunked( ``async def fetch(args) -> (df, response)``, and drives it to completion via :meth:`ChunkedCall.resume`. The plan splits multi-value list params and the cql-text filter so each sub-request URL fits the - byte limit; an already-fitting request is a one-step plan, unless an - active :func:`parallel_chunks` block asks the plan to fan out more - finely. See the module docstring for the concurrency model. + byte limit; an already-fitting request is a one-step plan unless the + effective ``parallel_chunks`` setting asks the plan to fan out more finely. + See the module docstring for the concurrency model. Parameters ---------- @@ -778,17 +760,18 @@ def wrapper( finalize: _Finalize = _passthrough_result, ) -> tuple[pd.DataFrame, Any]: limit = _OGC_URL_BYTE_LIMIT if url_limit is None else url_limit - # Read the parallel_chunks dial ``n`` from the ambient set by - # ``parallel_chunks`` (1 = off outside any such block; otherwise the - # requested total sub-request cap). It only affects *planning*, done - # here up front, so a later resume — which re-issues the - # already-planned sub-requests — needs no snapshot. + # Resolve the parallel_chunks dial ``n`` — an active + # ``parallel_chunks`` / ``configure`` block, else the configured + # baseline (1 = off; otherwise the optional-refinement ceiling). + # It only affects *planning*, done here up front, so a later + # resume — which re-issues the already-planned sub-requests — + # needs no snapshot. plan = ChunkPlan( - args, build_request, limit, max_chunks=_parallel_chunks.get() + args, build_request, limit, max_chunks=_config.parallel_chunks() ) - retry_policy = RetryPolicy.from_env() - # The concurrency cap is resolved inside ``resume()`` from - # ``API_USGS_CONCURRENT``; ``1`` is a sequential gather, + retry_policy = RetryPolicy.from_config() + # The effective concurrency cap is resolved inside ``resume()``; + # ``1`` is a sequential gather, # ``total <= 1`` a one-element gather — no special branch. return ChunkedCall(plan, fetch, retry_policy, finalize).resume() diff --git a/dataretrieval/ogc/engine.py b/dataretrieval/ogc/engine.py index 31b81f5f..a0f34201 100644 --- a/dataretrieval/ogc/engine.py +++ b/dataretrieval/ogc/engine.py @@ -526,7 +526,7 @@ async def _fetch_once( and iterates the cartesian product. With no chunkable inputs the decorator passes args through unchanged. The decorator gathers every sub-request over one shared :class:`httpx.AsyncClient` (concurrency - bounded by a semaphore, sized from ``API_USGS_CONCURRENT``) + bounded by a semaphore, sized from the effective ``concurrency`` setting) and returns a *synchronous* wrapper, so ``get_ogc_data`` keeps calling ``_fetch_once(args, finalize=...)`` synchronously. The return shape is ``(frame, response)``. diff --git a/dataretrieval/ogc/planning.py b/dataretrieval/ogc/planning.py index 15b397a0..d7ab175b 100644 --- a/dataretrieval/ogc/planning.py +++ b/dataretrieval/ogc/planning.py @@ -285,18 +285,15 @@ class ChunkPlan: Byte budget for the request (URL + body) — a hard ceiling every sub-request must fit. max_chunks : int, optional - Hard cap on the plan's total sub-request count (default ``1`` = off). - ``1`` chunks only as much as ``url_limit`` requires — the most - conservative plan, fewest sub-requests — so a fitting request is a - passthrough. A cap of ``2`` or more fans the plan out to up to - ``max_chunks`` sub-requests overall (the cartesian product across axes, - never fewer than the byte budget already forces) — capped as a whole, - not per axis, so several multi-value axes can't multiply past the cap. - The plan never exceeds the cap and may land below it when no whole - split lands on it exactly. ``max_chunks`` is a sub-request count, so a - value below ``1`` (``0`` or negative) is a caller error and raises - ``ValueError``. Set from the - :func:`~dataretrieval.ogc.chunking.parallel_chunks` ``n``; see + Soft ceiling for optional refinement (default ``1`` = off). ``1`` + chunks only as much as ``url_limit`` requires — the most conservative + plan, fewest sub-requests — so a fitting request is a passthrough. A + value of ``2`` or more refines the plan toward ``max_chunks`` total + sub-requests across all axes without crossing it. Mandatory URL-byte + splitting runs first and may already exceed this value; indivisible + inputs may leave optional refinement below it. A value below ``1`` + (``0`` or negative) is a caller error and raises ``ValueError``. Set + from the :func:`~dataretrieval.ogc.chunking.parallel_chunks` ``n``; see :meth:`_refine`. Attributes @@ -470,8 +467,8 @@ def _refine(self, max_chunks: int) -> None: the ``parallel_chunks`` dial (see :func:`~dataretrieval.ogc.chunking.parallel_chunks` for why a caller would want this, and :class:`ChunkPlan`'s ``max_chunks`` parameter for - the cap's contract: total-not-per-axis, a hard ceiling that may land - below the cap). + the cap's contract: total-not-per-axis, a soft target that may land + below the requested value or start above it after mandatory splitting). Implementation. Each split multiplies the plan by ``(k+1)/k`` for the chosen axis (adding ``total // k`` sub-requests, not one), so a split diff --git a/dataretrieval/ogc/progress.py b/dataretrieval/ogc/progress.py index 6177c30f..db1dc318 100644 --- a/dataretrieval/ogc/progress.py +++ b/dataretrieval/ogc/progress.py @@ -16,18 +16,21 @@ By default the line is shown for interactive use — an interactive terminal or a Jupyter/IPython kernel, like ``tqdm`` — while redirected logs and CI stay clean. -``API_USGS_PROGRESS`` forces it on (``1``/``true``) or off (``0``/``false``). +The ``progress`` setting forces it on (``1``/``true``) or off (``0``/``false``) +— via a ``dataretrieval.configure`` block, ``API_USGS_PROGRESS``, or the config +file. """ from __future__ import annotations import contextvars -import os import sys from collections.abc import Iterator from contextlib import contextmanager from typing import TextIO +import dataretrieval.config as _config + def _group_int(value: str) -> str: """Comma-group a plain ASCII integer string; pass anything else through. @@ -48,8 +51,8 @@ def _group_int(value: str) -> str: ) # Where to register for an API key. Surfaced once when a query runs without an -# API key configured (no API_USGS_PAT), since unauthenticated callers hit much -# lower rate limits (see the API_USGS_PAT note in the README). +# API key configured, since unauthenticated callers hit much lower rate limits +# (see the API key note in the README). SIGNUP_URL = "https://api.waterdata.usgs.gov/signup/" # Process-level latch so the "no API key" pointer is shown at most once. @@ -77,13 +80,15 @@ def _in_jupyter_kernel() -> bool: def _enabled_default(stream: TextIO) -> bool: """Whether to draw the line by default. - ``API_USGS_PROGRESS`` wins when set. Otherwise show it for interactive use — - a TTY or a Jupyter/IPython kernel — and stay quiet for redirected output, - logs, and CI. + An explicit setting wins — a ``dataretrieval.configure(progress=...)`` block, + ``API_USGS_PROGRESS``, or the config file (see + :mod:`dataretrieval.config` for the order). Otherwise show it for + interactive use — a TTY or a Jupyter/IPython kernel — and stay quiet for + redirected output, logs, and CI. """ - override = os.getenv("API_USGS_PROGRESS") + override = _config.progress() if override is not None: - return override.strip().lower() not in {"", "0", "false", "no", "off"} + return override if _in_jupyter_kernel(): return True return hasattr(stream, "isatty") and stream.isatty() @@ -225,7 +230,7 @@ def _render(self) -> None: def close(self) -> None: """Finalize the line with a trailing newline so it persists on screen. - If no API key is configured (no ``API_USGS_PAT``), append a one-time + If no API key is configured, append a one-time pointer to API-key registration, since unauthenticated callers hit much lower rate limits. """ @@ -250,7 +255,7 @@ def close(self) -> None: def _maybe_hint_api_key(self) -> None: global _api_key_hint_shown - if _api_key_hint_shown or os.getenv("API_USGS_PAT"): + if _api_key_hint_shown or _config.api_key(): return # Set the once-per-process latch only after a successful write, so a # failed write (broken pipe) doesn't silently burn the hint for every diff --git a/dataretrieval/ogc/retry.py b/dataretrieval/ogc/retry.py index bd45f275..09ffc452 100644 --- a/dataretrieval/ogc/retry.py +++ b/dataretrieval/ogc/retry.py @@ -11,7 +11,6 @@ from __future__ import annotations import asyncio -import os import random from collections.abc import Awaitable, Callable from dataclasses import dataclass @@ -19,6 +18,7 @@ import httpx import pandas as pd +import dataretrieval.config as _config import dataretrieval.ogc.progress as _progress from dataretrieval.exceptions import RateLimited, TransientError from dataretrieval.ogc.interruptions import ( @@ -30,11 +30,11 @@ # Retry-with-backoff defaults for transient sub-request failures (429 / # 5xx / connect-read timeouts): exponential backoff with full jitter, and # honor a server ``Retry-After`` up to the cap below before escalating -# to a resumable interruption instead. -_RETRIES_ENV = "API_USGS_RETRIES" - - -_RETRIES_DEFAULT = 4 +# to a resumable interruption instead. The retry count itself resolves at call +# time through :mod:`dataretrieval.config` (a ``configure()`` block, then +# ``API_USGS_RETRIES``, then the config file); the default below is the dataclass +# field default for hand-constructed policies. +_RETRIES_DEFAULT = _config.DEFAULT_RETRIES _RETRY_BASE_BACKOFF = 0.5 @@ -46,30 +46,6 @@ _RETRY_AFTER_CAP = 60.0 -def _read_retries_env() -> int: - """ - Resolve the ``API_USGS_RETRIES`` env var to a max-retry count. - - Returns - ------- - int - Number of retries after the first attempt; ``0`` disables - retrying. Unset/blank → ``_RETRIES_DEFAULT``. - """ - raw = os.environ.get(_RETRIES_ENV) - if raw is None or raw.strip() == "": - return _RETRIES_DEFAULT - try: - value = int(raw.strip()) - except ValueError as exc: - raise ValueError( - f"{_RETRIES_ENV} must be a non-negative integer (got {raw!r})." - ) from exc - if value < 0: - raise ValueError(f"{_RETRIES_ENV} must be >= 0 (got {value}).") - return value - - @dataclass(frozen=True) class RetryPolicy: """Bounded retry-with-backoff config for transient sub-request failures. @@ -114,14 +90,16 @@ def __post_init__(self) -> None: raise ValueError("retry backoff settings must be non-negative.") @classmethod - def from_env(cls) -> RetryPolicy: + def from_config(cls) -> RetryPolicy: """ - Build a policy from the module-level defaults, resolved now. + Build a policy from the effective configuration, resolved now. - Reads ``max_retries`` from ``API_USGS_RETRIES`` and the timing - knobs from the ``_RETRY_*`` module constants at call time — not - the dataclass field defaults (which freeze at class definition) - — so test ``monkeypatch.setattr`` on the constants takes effect. + Reads ``max_retries`` through :mod:`dataretrieval.config` — a + ``dataretrieval.configure(retries=...)`` block, then + ``API_USGS_RETRIES``, then the config file — and the timing knobs from the + ``_RETRY_*`` module constants at call time, not the dataclass field + defaults (which freeze at class definition), so test + ``monkeypatch.setattr`` on the constants takes effect. Returns ------- @@ -130,7 +108,7 @@ def from_env(cls) -> RetryPolicy: call time. """ return cls( - max_retries=_read_retries_env(), + max_retries=_config.retries(), base_backoff=_RETRY_BASE_BACKOFF, max_backoff=_RETRY_MAX_BACKOFF, retry_after_cap=_RETRY_AFTER_CAP, @@ -186,7 +164,7 @@ def backoff(self, attempt: int, retry_after: float | None) -> float: # Default for direct ``ChunkedCall`` / ``ChunkPlan.execute`` construction # (and tests): no retrying. The production decorator path explicitly passes -# ``RetryPolicy.from_env()`` so retries are on by default there. +# ``RetryPolicy.from_config()`` so retries are on by default there. _NO_RETRY = RetryPolicy(max_retries=0) diff --git a/dataretrieval/utils.py b/dataretrieval/utils.py index 7506a469..31b339d1 100644 --- a/dataretrieval/utils.py +++ b/dataretrieval/utils.py @@ -5,7 +5,6 @@ from __future__ import annotations import numbers -import os import warnings from collections.abc import Callable, Iterable, Iterator from contextlib import contextmanager @@ -17,6 +16,7 @@ import httpx import pandas as pd +import dataretrieval.config as _config from dataretrieval.codes import tz from dataretrieval.exceptions import ( NetworkError, @@ -114,10 +114,20 @@ def _default_headers(target_url: str | httpx.URL | None = None) -> dict[str, str """Build the default HTTP headers for a USGS web-API request. Always sets a descriptive ``User-Agent`` plus ``Accept`` / - ``Accept-Encoding`` and ``lang``. If the ``API_USGS_PAT`` environment - variable is set AND ``target_url`` points to the explicitly authorized - Water Data host (``api.waterdata.usgs.gov``), its value is added as the - ``X-Api-Key`` header. The key is never sent to other hosts. + ``Accept-Encoding`` and ``lang``. If an API key is configured AND + ``target_url`` points to the explicitly authorized Water Data host + (``api.waterdata.usgs.gov``), its value is added as the ``X-Api-Key`` + header. The key is never sent to other hosts. + + The key resolves through :mod:`dataretrieval.config` -- a + ``dataretrieval.configure(api_key=...)`` block, then ``API_USGS_PAT``, then + the configuration file -- so host scoping applies identically no matter + which source supplied it. Resolution happens *after* the host check, and + only for the authorized host: config resolution can raise + :class:`~dataretrieval.exceptions.ConfigError` (a malformed config file, a + profile the file no longer defines), and a Water Data configuration problem + must not break a legacy NWIS, WQP, or NGWMN call that would never receive + the key. Parameters ---------- @@ -138,14 +148,15 @@ def _default_headers(target_url: str | httpx.URL | None = None) -> dict[str, str "User-Agent": f"python-dataretrieval/{_PACKAGE_VERSION}", "lang": "en-US", } - token = os.getenv("API_USGS_PAT") - if token and target_url is not None: + if target_url is not None: try: host = httpx.URL(str(target_url)).host except (httpx.InvalidURL, TypeError): host = None if host == _AUTHORIZED_API_KEY_HOST: - headers["X-Api-Key"] = token + token = _config.api_key() + if token: + headers["X-Api-Key"] = token return headers diff --git a/dataretrieval/waterdata/utils.py b/dataretrieval/waterdata/utils.py index a6d91272..4ad807c1 100644 --- a/dataretrieval/waterdata/utils.py +++ b/dataretrieval/waterdata/utils.py @@ -123,6 +123,38 @@ "thresholds", } +# Credential-shaped kwargs must never reach the generic queryable passthrough: +# URLs are retained by clients, proxies, logs, and response metadata. +# +# Matched as *substrings* of the separator-stripped name, not as exact names: +# an exact-match list missed the spelling the library's own docs make most +# tempting -- ``x_api_key``, after the ``X-Api-Key`` header. +# +# This catches the plausible mistake; it is not a security control. Nothing +# inspects *values*, so a secret pasted into ``state_name=`` travels just the +# same, and the name space belongs to the server (``get_queryables``) rather +# than to us. The point is to answer the caller who reasonably guesses that a +# credential goes here, with a TypeError naming ``configure(api_key=...)`` +# instead of a token in a URL. It errs toward rejecting for that reason. +_FORBIDDEN_QUERYABLE_MARKERS = ( + "apikey", + "authorization", + "credential", + "password", + "passwd", + "secret", + "token", +) + +# Whole names that are credentials on their own but too short to match as +# substrings without catching legitimate queryables. +# +# ``session`` is deliberately absent from both lists: it carries no secret, so +# rejecting it with a credentials message told users the wrong thing, and as a +# substring it claimed part of a namespace the *server* owns -- any future +# queryable containing it would have been unreachable behind that message. +_FORBIDDEN_QUERYABLE_NAMES = frozenset({"auth", "key", "pat", "pw"}) + def _flatten_queryables(local_vars: dict[str, Any]) -> dict[str, Any]: """Merge a getter's ``**queryables`` passthrough kwargs -- collected by @@ -137,7 +169,21 @@ def _flatten_queryables(local_vars: dict[str, Any]) -> dict[str, Any]: popped, so this is a no-op on getters without the passthrough and idempotent if called twice. """ - local_vars.update(local_vars.pop("queryables", {})) + queryables = local_vars.pop("queryables", {}) + forbidden = set() + for name in queryables: + flat = name.replace("_", "").replace("-", "").casefold() + if flat in _FORBIDDEN_QUERYABLE_NAMES or any( + marker in flat for marker in _FORBIDDEN_QUERYABLE_MARKERS + ): + forbidden.add(name) + if forbidden: + names = ", ".join(f"{name}=" for name in sorted(forbidden)) + raise TypeError( + f"Credentials cannot be passed as query parameters ({names}); " + "use dataretrieval.configure(api_key=...) instead." + ) + local_vars.update(queryables) return local_vars diff --git a/docs/source/architecture/decisions/0006-layered-configuration.rst b/docs/source/architecture/decisions/0006-layered-configuration.rst new file mode 100644 index 00000000..4f5741b6 --- /dev/null +++ b/docs/source/architecture/decisions/0006-layered-configuration.rst @@ -0,0 +1,138 @@ +ADR 0006: Layered configuration resolution +========================================== + +Status +------ + +Accepted + +Context +------- + +Settings reached the library through one mechanism: process-global environment +variables (``API_USGS_PAT``, ``API_USGS_CONCURRENT``, ``API_USGS_RETRIES``, +``API_USGS_PROGRESS``), each with its own hand-rolled parser at its point of +use. Nothing could report the effective configuration, and the grammars were +free to drift apart. + +That mechanism cannot express a per-call credential. An application holding +keys in a secret store, a notebook pulling for two accounts, or a server +handling concurrent users must assign to ``os.environ`` — which is +process-global, so it races across threads and tasks (issue #352). + +The obvious fix, an ``api_key=`` parameter on the public getters, is unsafe +here. Every Water Data getter ends in ``_get_args(locals())`` with a +``**queryables`` catch-all that forwards unrecognized keywords to the API as +query parameters. A credential parameter missed in one of ~20 signatures would +be serialized into a URL. The maintainers also object to an ``api_key=`` +parameter on the separate ground that it invites keys pasted into shared +scripts. + +Decision +-------- + +Every setting resolves through one ordered chain, owned by a new +``dataretrieval.config`` module: + +1. An active ``dataretrieval.configure(...)`` block (a ``ContextVar``). +2. The setting's environment variable. +3. The configuration file: ``~/.dataretrieval/config.toml``, or the path in + ``DATARETRIEVAL_CONFIG``. Top-level keys are the defaults; a + ``[profiles.]`` table layers over them per setting when selected. +4. The built-in default. + +Supporting decisions: + +- **Precedence is per setting, not per source.** An environment that sets only + ``API_USGS_PAT`` leaves a file-provided ``concurrency`` in effect. A + *blank* environment variable does not count as set, so it cannot shadow the + file: container and CI tooling routinely materializes one. The exception is + ``progress``, where a blank ``API_USGS_PROGRESS`` has always meant "off" -- + so "does blank count as a value?" is a property of the setting + (``config._BLANK_MEANS_SET``) rather than an extra tier in the chain. +- **The environment ranks above the file.** This follows the established + precedence used by `pip + `_ + and `AWS + `_, + supports deployment-time overrides without editing mounted files, and keeps + the pre-existing ``API_USGS_*`` interface authoritative. +- **Omitted and explicitly cleared values differ.** An omitted + ``configure()`` argument inherits from lower sources. Explicit ``None`` is a + scoped reset to built-in behavior, so a server can guarantee an anonymous + call rather than accidentally falling through to its process credential. +- **No public getter grows a credential parameter.** ``configure`` is the only + programmatic path, and a fitness function asserts no getter accepts + ``api_key`` / ``session`` / ``token``. The generic ``**queryables`` path also + rejects those names before request construction so they cannot enter a URL. +- **The module owns each setting's parser.** ``unbounded``, bounds, and + rejection messages live in one place. ``tomllib`` returns typed scalars, so + the file and Python API validate source-level types before normalized values + pass through the shared parsers. Legacy environment-only forms, including a + blank numeric value and arbitrary non-empty progress value, remain compatible + without making the new surfaces equally permissive. +- **TOML, read with** ``tomllib``. Stdlib from Python 3.11; the ``tomli`` + backport is a marker-scoped dependency that disappears when + ``requires-python`` moves to ``>=3.11``. YAML was rejected because PyYAML is + a dependency at every Python version and the settings are flat. +- **Not every setting gets an environment variable.** ``parallel_chunks`` + spends rate-limit quota, and ADR-adjacent documentation on + ``dataretrieval.parallel_chunks`` argues it must stay a deliberate choice. + It does not add a new exported process-global knob; the file and ``configure`` + block are its only sources, with a scoped block as the recommended use. +- **Names distinguish execution capacity from planning granularity.** + ``concurrency`` is the noun for the maximum in-flight subrequests and maps to + the established ``API_USGS_CONCURRENT`` variable. ``parallel_chunks`` asks + the planner for optional extra chunks; it does not promise that many requests + execute simultaneously. The name is retained because the context manager is + already public. ``parallelism`` and ``chunk_parallelism`` were rejected + because they would conflate this planning hint with ``concurrency``. +- **Configuration errors are in the error taxonomy.** ``ConfigError`` is a + ``DataRetrievalError`` *and* a ``ValueError``. Configuration resolves lazily + on the request path, so a broken file surfaces from inside whichever getter + runs first; ``except DataRetrievalError`` around a call has to catch it like + any other failure of that call, while the ``ValueError`` base keeps the + handlers that predate the file layer working. +- **``parallel_chunks`` at the top level of the file warns.** It is the one + setting that spends rate-limit quota, so a value left there applies to every + splittable query in every process that reads the file. A + ``[profiles.]`` table is opt-in per run, which is the shape this + setting wants; the top-level form still works but says so. +- **``dataretrieval.config`` is a lightweight leaf.** It uses only the standard + library, the ``tomli`` backport on Python 3.10, and + ``dataretrieval.exceptions`` -- itself a dependency-free leaf, so this adds + no weight and cannot cycle. It is read by ``utils`` + (headers), ``ogc.chunking``, ``ogc.retry``, and ``ogc.progress``, so under ADR + 0003 it must import none of them. The public callable is named ``configure`` + rather than ``config`` so it does not shadow the module. It is a scoped + action, not a ``Configuration`` dataclass: a value object would imply + snapshot, equality, serialization, and representation contracts while + risking disclosure of the API key through generated helpers. + +Consequences +------------ + +- A credential can be supplied per thread or per task without touching + ``os.environ``, which is what issue #352 asked for. +- Host scoping is unchanged and unconditional: a key from any source is sent + only to ``api.waterdata.usgs.gov`` and is stripped on cross-host redirects. +- ``show_config()`` reports the effective value and provenance of each setting + without ever printing the key. +- Behavior is unchanged when no file exists and no block is active, so + existing environment-variable users are unaffected. +- A configuration file becomes a supported artifact whose format is now a + compatibility surface. +- The Python floor and the file format are coupled: raising + ``requires-python`` to ``>=3.11`` drops the ``tomli`` dependency with no + other change. + +Compliance +---------- + +``tests/architecture_test.py::test_config_is_a_standard_library_only_leaf`` +asserts the module imports nothing from ``dataretrieval`` other than the +``exceptions`` taxonomy leaf, and no third-party package other than the +``tomli`` backport. +``tests/config_test.py`` covers the precedence chain, per-setting merging, +thread and asyncio isolation, host scoping for file-sourced keys, redaction in +``show_config``, and rejection of credential parameters on public getters. diff --git a/docs/source/architecture/decisions/index.rst b/docs/source/architecture/decisions/index.rst index f11aa4ba..2098afe0 100644 --- a/docs/source/architecture/decisions/index.rst +++ b/docs/source/architecture/decisions/index.rst @@ -22,4 +22,5 @@ records sequentially. 0003-dependency-direction 0004-error-retry-resume 0005-legacy-nwis + 0006-layered-configuration template diff --git a/docs/source/architecture/index.rst b/docs/source/architecture/index.rst index 906c07a4..23686251 100644 --- a/docs/source/architecture/index.rst +++ b/docs/source/architecture/index.rst @@ -95,6 +95,14 @@ Public service facades Shared components ^^^^^^^^^^^^^^^^^ +``dataretrieval.config`` + Lightweight configuration leaf: standard library plus the ``tomli`` + backport on Python 3.10. It resolves scoped overrides, environment + variables, a TOML file with optional profiles, and built-in defaults in + that order. Service and protocol modules may depend on it; it must not + depend back on them. Scoped overrides use ``ContextVar`` so concurrent + threads and asyncio tasks can carry distinct credentials. + ``dataretrieval.ogc`` Protocol subsystem for Water Data and NGWMN. A small facade (``__init__.py``) exposes the service-adapter seam: ``OgcDialect``, @@ -115,9 +123,9 @@ Shared components be imported by every service without creating an infrastructure cycle. ``dataretrieval.utils`` - Shared metadata, data-shaping helpers, ambient context support, and the - legacy synchronous request path. Its broad responsibility is known debt; - new service-specific behavior should not be added there by default. + Shared metadata, data-shaping helpers, and the legacy synchronous request + path. Its broad responsibility is known debt; new service-specific behavior + should not be added there by default. ``dataretrieval.codes`` and ``dataretrieval.rdb`` State/time-zone code conversion and RDB parsing leaves. @@ -177,24 +185,28 @@ those public contracts and must not invent unsupported upstream capabilities. Resource and configuration view ------------------------------- -``API_USGS_PAT`` - Optional USGS API token. It is attached only to requests for - ``api.waterdata.usgs.gov``. Shared synchronous and asynchronous clients - re-check every redirected request and strip the token before following a - link to any other host, including external rating assets. - -``API_USGS_CONCURRENT`` - OGC subrequest concurrency cap; defaults to 32, ``1`` is sequential, and - ``unbounded`` removes the explicit cap. A semaphore, not pool waiting, is - the execution throttle. - -``API_USGS_RETRIES`` - Number of OGC retries after the first attempt; defaults to four. Backoff is - exponential with full jitter and honors bounded ``Retry-After`` values. - -``API_USGS_PROGRESS`` - Controls best-effort progress display. Reporting failures must never change - retrieval results. +Every setting resolves per key through an active ``configure()`` block, its +environment variable when one exists, the selected profile and top-level +values in ``~/.dataretrieval/config.toml``, then its built-in default. +``show_config()`` reports the effective source while redacting credentials. + +The settings themselves -- names, defaults, environment variables, and the +config-file format -- are catalogued once in the +:doc:`configuration guide `. What matters +architecturally is the behavior around them: + +* The API token is attached only to requests for ``api.waterdata.usgs.gov``. + Shared synchronous and asynchronous clients re-check every redirected request + and strip the token before following a link to any other host, including + external rating assets. +* A semaphore, not connection-pool waiting, is the execution throttle for + sub-request concurrency. +* Retry backoff is exponential with full jitter and honors bounded + ``Retry-After`` values. +* Progress reporting is best-effort: a reporting failure must never change + retrieval results. +* ``dataretrieval.config`` is a stdlib-only leaf, so any module may depend on + it without an import cycle. HTTP timeouts and connection limits are centralized for existing paths. ``wateruse`` currently has its own smaller fan-out cap. These differences must diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst new file mode 100644 index 00000000..751fabb5 --- /dev/null +++ b/docs/source/reference/config.rst @@ -0,0 +1,14 @@ +.. _config: + +dataretrieval.config +-------------------- + +Layered configuration: a ``dataretrieval.configure(...)`` block, then +the ``API_USGS_*`` environment variables, then +``~/.dataretrieval/config.toml``, then built-in defaults. See the +:doc:`configuration guide ` for the settings and +worked examples. + +.. automodule:: dataretrieval.config + :members: configure, show_config, config_path, ConfigError + :show-inheritance: diff --git a/docs/source/reference/index.rst b/docs/source/reference/index.rst index 13c44963..d5ff3e83 100644 --- a/docs/source/reference/index.rst +++ b/docs/source/reference/index.rst @@ -7,6 +7,7 @@ API reference .. toctree:: :maxdepth: 1 + config exceptions ngwmn nldi diff --git a/docs/source/userguide/configuration.rst b/docs/source/userguide/configuration.rst new file mode 100644 index 00000000..571ada21 --- /dev/null +++ b/docs/source/userguide/configuration.rst @@ -0,0 +1,298 @@ +.. _configuration: + +============= +Configuration +============= + +``dataretrieval`` has a handful of settings — most importantly your Water Data +API key. Each one resolves through the same ordered chain, so you can pick the +mechanism that suits how your code runs. + +.. contents:: + :local: + :depth: 1 + + +Settings +-------- + +.. list-table:: + :header-rows: 1 + :widths: 18 12 26 44 + + * - Setting + - Default + - Environment variable + - What it does + * - ``api_key`` + - none + - ``API_USGS_PAT`` + - Water Data API key. Raises your hourly request quota substantially; + `register for one `_. + * - ``concurrency`` + - ``32`` + - ``API_USGS_CONCURRENT`` + - Cap on sub-requests in flight at once for a chunked query. A positive + integer, ``1`` to run them one at a time, or ``"unbounded"`` to remove + the cap. Does not change how many requests are made, only how many run + simultaneously. + * - ``retries`` + - ``4`` + - ``API_USGS_RETRIES`` + - Retries after a transient failure (429, 5xx, timeout). ``0`` disables. + * - ``progress`` + - auto + - ``API_USGS_PROGRESS`` + - Whether to draw the status line. Auto means on for a terminal or + Jupyter kernel, off for redirected output and CI. + * - ``parallel_chunks`` + - ``1`` + - *(none — see below)* + - Default fan-out for multi-value queries. ``1`` means split only as far + as the URL byte limit forces. + + +Where settings come from +------------------------ + +Highest precedence first: + +1. An active ``dataretrieval.configure(...)`` block. +2. The environment variable for that setting. +3. The configuration file — ``~/.dataretrieval/config.toml``, or the path in + ``DATARETRIEVAL_CONFIG``. +4. The built-in default. + +Precedence applies **per setting**. An environment that sets only +``API_USGS_PAT`` leaves a file-provided ``concurrency`` fully in effect — +sources are merged, not replaced. + +A variable that is *set but empty* (``export API_USGS_PAT=``, or a CI secret +that resolves to nothing) does not count as configured, so an empty variable +your tooling happened to create cannot silently discard the key in your config +file. The one exception is ``API_USGS_PROGRESS``, where blank has always meant +"off" and so is treated as a real value. + +.. note:: + + The environment ranks above the file, matching common deployment tools and + preserving the existing ``API_USGS_*`` variables as authoritative runtime + overrides. The reasoning is in :doc:`ADR 0006 + `. + + +An environment variable +----------------------- + +Still fully supported, and the simplest option for a single key on one +machine: + +.. code-block:: bash + + export API_USGS_PAT="your_api_key_here" + +This is also the mechanism the `R dataRetrieval package +`_ uses, under the same variable +name, so one export serves both. + + +A configuration file +-------------------- + +Better when you would rather not have a credential in your shell environment, +where it is inherited by every process you start. Create +``~/.dataretrieval/config.toml``: + +.. code-block:: toml + + api_key = "your_api_key_here" + +Restrict it so other users on the machine cannot read it — ``dataretrieval`` +warns once if a file containing a key is group- or world-readable: + +.. code-block:: bash + + chmod 600 ~/.dataretrieval/config.toml + +Any setting can go in the file: + +.. code-block:: toml + + api_key = "your_api_key_here" + concurrency = 16 + retries = 8 + +Point ``DATARETRIEVAL_CONFIG`` at a different path to override the location — +useful for a container or a job scheduler that mounts secrets elsewhere. + + +Profiles +~~~~~~~~ + +A ``[profiles.]`` table layers over the top-level keys, so a profile +only states what differs: + +.. code-block:: toml + + # top level = the defaults every profile starts from + api_key = "your_api_key_here" + concurrency = 16 + + [profiles.bulk-pull] + # api_key is inherited from the top level; only the differences go here + concurrency = "unbounded" + parallel_chunks = 8 + + [profiles.polite] + # likewise inherits api_key + concurrency = 2 + +Select one for a run, or for a block: + +.. code-block:: bash + + DATARETRIEVAL_PROFILE=bulk-pull python overnight_job.py + +.. code-block:: python + + with dataretrieval.configure(profile="bulk-pull"): + df, md = waterdata.get_daily(monitoring_location_id=many_sites) + +Both profiles above inherit the top-level ``api_key`` — you write the key once. + + +A ``configure`` block +--------------------- + +The highest-precedence source, and the one to use when a setting must apply to +*this* call and no other: + +.. code-block:: python + + import dataretrieval + from dataretrieval import waterdata + + with dataretrieval.configure(api_key=secrets["usgs"]): + df, md = waterdata.get_daily( + monitoring_location_id="USGS-05114000", + parameter_code="00060", + time="P7D", + ) + +Because it is backed by a :class:`~contextvars.ContextVar`, the value applies +to the current thread and to asyncio tasks started inside the block, and +cannot leak into another thread or task. That is what makes it usable from a +web service or a notebook working with more than one account: + +.. code-block:: python + + # each thread keeps its own key; no os.environ mutation, no race + def fetch_for(user): + with dataretrieval.configure(api_key=vault.read(user.key_path)): + return waterdata.get_daily(monitoring_location_id=user.sites) + +Blocks nest and merge per setting, so an inner block that tunes one thing +keeps the rest: + +.. code-block:: python + + with dataretrieval.configure(api_key=key, concurrency=8): + ... + with dataretrieval.configure(concurrency=1): # api_key still applies + ... + +Values are validated on entry, so a typo raises at the ``with`` statement +rather than deep inside a later request. + +Omitted settings inherit from an outer block or a lower-precedence source. +Passing ``None`` explicitly suppresses those sources and restores built-in +behavior for that block. For example, ``configure(api_key=None)`` makes an +anonymous call even if ``API_USGS_PAT`` is set, while ``profile=None`` selects +the file's top-level settings instead of ``DATARETRIEVAL_PROFILE``. + +.. tip:: + + Prefer reading the key from a secret store, environment, or config file + over writing a literal into a script — a literal is what ends up committed + or pasted into a shared notebook. + + +Checking what is in effect +-------------------------- + +``show_config()`` reports each setting's effective value and where it came +from. It never prints the key itself: + +.. code-block:: python + + >>> dataretrieval.show_config() + config file /home/u/.dataretrieval/config.toml (found) + profile bulk-pull + api_key /home/u/.dataretrieval/config.toml + concurrency unbounded /home/u/.dataretrieval/config.toml [profiles.bulk-pull] + retries 8 $API_USGS_RETRIES + progress auto built-in default + parallel_chunks 8 /home/u/.dataretrieval/config.toml [profiles.bulk-pull] + +Each line names the exact source, including which table inside the file, which +is usually enough to answer "why is it still using my old key?". + +It never raises. A malformed file, a value that fails its grammar, or a +profile that no longer exists is reported in place — on the ``config file`` +line for a whole-file problem, or in that setting's own row — because a broken +configuration is exactly when you reach for this. + + +Why ``parallel_chunks`` has no environment variable +--------------------------------------------------- + +Every other setting can be set from the environment. ``parallel_chunks`` +cannot, on purpose. + +Raising it splits a query into more sub-requests, and *each sub-request spends +rate-limit quota*. Whether that is a good trade depends on the size of the +query — which the library cannot know in advance. The setting therefore does +not add another process-global environment knob that could be exported once +and inherited by every subprocess. + +Set it per call, which is almost always what you want: + +.. code-block:: python + + with waterdata.parallel_chunks(8): + df, md = waterdata.get_daily(monitoring_location_id=many_sites) + +or as a baseline in the config file — deliberately written, and visible in +``show_config()``. Put it in a ``[profiles.]`` table rather than at the +top level: a profile applies only to runs that select it, while a top-level +value applies to every query in every process that reads the file, which is +how a setting added for one bulk pull quietly exhausts an hourly quota months +later. ``dataretrieval`` warns if it finds one at the top level. + +The value limits optional refinement only. URL-byte safety can require more +sub-requests than the configured value, and an input with nothing to split +stays a single request. + +``parallel_chunks(n)`` is sugar for ``configure(parallel_chunks=n)``: one +scoping mechanism, so the innermost block wins whichever spelling set it, and +``show_config()`` always reports the value the chunker will actually use. + + +Keeping a key out of your environment entirely +---------------------------------------------- + +If your credentials live in a secret manager, nothing needs to touch +``os.environ``: + +.. code-block:: python + + import dataretrieval + import boto3 + from dataretrieval import waterdata + + with dataretrieval.configure(api_key=boto3.client("secretsmanager") + .get_secret_value(SecretId="usgs-pat")["SecretString"]): + df, md = waterdata.get_continuous(monitoring_location_id="USGS-05114000") + +Wherever the key comes from, it is sent only to ``api.waterdata.usgs.gov`` and +is stripped from any cross-host redirect, so it cannot leak to another host. diff --git a/docs/source/userguide/errors.rst b/docs/source/userguide/errors.rst index 28da515f..65a6ca1d 100644 --- a/docs/source/userguide/errors.rst +++ b/docs/source/userguide/errors.rst @@ -108,9 +108,9 @@ would; a split that leaves each sub-request only a page or two adds its partial final page). So if you *know* your pull is large you can ask for a finer split with ``parallel_chunks(n)`` -- trading roughly the same pages for more, smaller sub-requests, which gives smoother progress, more even concurrency, and a -smaller unit of retry/resume. It is a scoped ``with`` -block, so an aggressive setting can't leak into unrelated calls and -accidentally spend quota: +smaller unit of retry/resume. A scoped ``with`` block is the recommended use, +so an aggressive setting does not leak into unrelated calls and accidentally +spend quota: .. code-block:: python @@ -121,19 +121,17 @@ accidentally spend quota: monitoring_location_id=many_sites, parameter_code="00060" ) -``n`` is a positive integer (e.g. ``2``, ``8``, ``32``) -- the number of -sub-requests to fan the call out into; a non-integer or non-positive value -raises ``ValueError`` at the ``with``. It caps the *total* sub-request count -across every multi-value argument combined (not per argument), bounded below by -what the byte limit already forces and above by how many values there are to -split, so several multi-value arguments can't multiply past it and ``n=1`` asks -for no extra fan-out. Each sub-request costs a request against your hourly rate -limit, and because how many run *at once* is capped separately by -``API_USGS_CONCURRENT`` (default 32) an ``n`` beyond that adds quota without -adding parallelism -- the useful range is roughly ``2`` up to -``API_USGS_CONCURRENT``. There is no "off" level: simply don't enter the block -unless you already expect a large, multi-page result -- on a query that would -have fit in a single page, extra chunks only burn quota. +``n`` is a positive integer (e.g. ``2``, ``8``, ``32``); a non-integer or +non-positive value raises ``ValueError`` at the ``with``. It is a soft ceiling +for optional refinement across every multi-value argument combined. URL-byte +safety may already require more than ``n`` sub-requests, while indivisible +inputs may produce fewer; ``n=1`` asks for no extra fan-out. Each sub-request +costs a request against your hourly rate limit. How many run *at once* is +capped separately by the effective ``concurrency`` setting +(``API_USGS_CONCURRENT``), so an ``n`` beyond that adds quota +without adding parallelism. Omitting the block uses the configured baseline, +which defaults to ``1``; use a larger config-file baseline only for workloads +that consistently return large, multi-page results. The full taxonomy ================= diff --git a/docs/source/userguide/index.rst b/docs/source/userguide/index.rst index 3cc4748a..e200b0f0 100644 --- a/docs/source/userguide/index.rst +++ b/docs/source/userguide/index.rst @@ -13,6 +13,7 @@ Contents .. toctree:: :maxdepth: 1 + configuration errors timeconventions dataportals diff --git a/pyproject.toml b/pyproject.toml index cec99697..c99257fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,10 @@ dependencies = [ # Directly imported by ``waterdata`` (``anyio.from_thread.start_blocking_portal``), # so declared here rather than relied on transitively via httpx. "anyio>=4.0", + # ``dataretrieval.config`` reads a TOML config file. ``tomllib`` is stdlib + # from 3.11, so this marker installs the backport only on 3.10 and drops + # itself when ``requires-python`` moves to >=3.11. + "tomli>=1.1.0; python_version < '3.11'", ] dynamic = ["version"] diff --git a/tests/architecture_test.py b/tests/architecture_test.py index 8ba2d864..23cae680 100644 --- a/tests/architecture_test.py +++ b/tests/architecture_test.py @@ -136,6 +136,44 @@ def test_exceptions_has_no_runtime_third_party_dependency() -> None: ) +def test_config_is_a_standard_library_only_leaf() -> None: + """Configuration resolution must stay importable from anywhere. + + ``dataretrieval.config`` is read by ``utils`` (headers), ``ogc.chunking`` + (concurrency), ``ogc.retry``, and ``ogc.progress``. If it imported any of + them -- or any third-party package -- it would create a cycle or make the + cheapest module in the package expensive. ``tomli`` is the one allowed + third-party import: it is the ``tomllib`` backport, used only on Python + 3.10, and the ``sys.version_info`` guard means it is not even imported on + 3.11+. + + ``dataretrieval.exceptions`` is the one allowed first-party import: it is + the taxonomy leaf, itself free of first-party and runtime third-party + imports (asserted above), so depending on it adds no weight and cannot + cycle. ``ConfigError`` lives there because configuration resolves on the + request path, so a broken config file must be catchable as + ``except DataRetrievalError`` like any other failure of that call. + """ + imports = _runtime_imports(PACKAGE_ROOT / "config.py") + first_party = { + name + for name in imports + if name.startswith("dataretrieval") and name != "dataretrieval.exceptions" + } + assert not first_party, ( + "dataretrieval.config may only import dataretrieval.exceptions: " + f"{sorted(first_party)}" + ) + roots = {module.partition(".")[0] for module in imports} + # Static analysis sees both sides of the version guard. Python 3.10's + # stdlib inventory does not yet include the unreachable ``tomllib`` branch. + allowed = {"dataretrieval", "tomli", "tomllib"} + third_party = roots - sys.stdlib_module_names - allowed + assert not third_party, ( + f"dataretrieval.config gained third-party dependencies: {sorted(third_party)}" + ) + + def test_ogc_does_not_depend_on_service_adapters() -> None: """The reusable protocol subsystem must not point back to its callers.""" violations: list[str] = [] diff --git a/tests/config_test.py b/tests/config_test.py new file mode 100644 index 00000000..02c18895 --- /dev/null +++ b/tests/config_test.py @@ -0,0 +1,823 @@ +"""Tests for layered configuration resolution (``dataretrieval.config``).""" + +from __future__ import annotations + +import asyncio +import io +import os +import threading + +import pytest + +import dataretrieval +from dataretrieval import config +from dataretrieval.utils import _default_headers + +WATERDATA_URL = "https://api.waterdata.usgs.gov/ogcapi/v0/collections/daily/items" + + +@pytest.fixture +def config_file(tmp_path, monkeypatch): + """Write a config file and point ``DATARETRIEVAL_CONFIG`` at it.""" + + def write(text: str): + path = tmp_path / "config.toml" + path.write_text(text) + path.chmod(0o600) # keep the loose-permission warning out of the way + for env in config.ENV_VARS.values(): + monkeypatch.delenv(env, raising=False) + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(path)) + config._reset_file_cache() + return path + + return write + + +# --- precedence ---------------------------------------------------------- + + +def test_default_when_nothing_is_configured(monkeypatch): + for env in config.ENV_VARS.values(): + monkeypatch.delenv(env, raising=False) + assert config.api_key() is None + assert config.concurrency() == config.DEFAULT_CONCURRENCY + assert config.retries() == config.DEFAULT_RETRIES + assert config.parallel_chunks() == config.DEFAULT_PARALLEL_CHUNKS + assert config.progress() is None + + +def test_env_is_used_when_no_file_or_block(monkeypatch): + monkeypatch.setenv("API_USGS_PAT", "env-key") + monkeypatch.setenv("API_USGS_CONCURRENT", "4") + assert config.api_key() == "env-key" + assert config.concurrency() == 4 + + +def test_env_outranks_file(config_file, monkeypatch): + config_file('api_key = "file-key"\n') + monkeypatch.setenv("API_USGS_PAT", "env-key") + assert config.api_key() == "env-key" + + +def test_block_outranks_file_and_env(config_file, monkeypatch): + config_file('api_key = "file-key"\n') + monkeypatch.setenv("API_USGS_PAT", "env-key") + with dataretrieval.configure(api_key="block-key"): + assert config.api_key() == "block-key" + assert config.api_key() == "env-key" + + +def test_precedence_is_per_setting_not_per_source(config_file, monkeypatch): + """An environment key must not blank out file-provided settings.""" + config_file("concurrency = 16\n") + monkeypatch.setenv("API_USGS_PAT", "env-key") + monkeypatch.setenv("API_USGS_RETRIES", "9") + assert config.concurrency() == 16 # from the file + assert config.api_key() == "env-key" # still from the env + assert config.retries() == 9 # still from the env + + +# --- the configure() block ----------------------------------------------- + + +def test_blocks_nest_and_merge_per_setting(): + with dataretrieval.configure(api_key="outer", concurrency=4): + with dataretrieval.configure(concurrency=8): + assert config.concurrency() == 8 + assert config.api_key() == "outer" # inherited from the outer block + assert config.concurrency() == 4 # inner block restored on exit + + +def test_omitted_setting_inherits_lower_source(monkeypatch): + monkeypatch.setenv("API_USGS_PAT", "env-key") + with dataretrieval.configure(concurrency=2): + assert config.api_key() == "env-key" + + +def test_explicit_none_suppresses_lower_sources(monkeypatch): + monkeypatch.setenv("API_USGS_PAT", "env-key") + monkeypatch.setenv("API_USGS_CONCURRENT", "4") + monkeypatch.setenv("API_USGS_PROGRESS", "true") + with dataretrieval.configure(api_key=None, concurrency=None, progress=None): + assert config.api_key() is None + assert config.concurrency() == config.DEFAULT_CONCURRENCY + assert config.progress() is None + assert config.api_key() == "env-key" + assert config.concurrency() == 4 + assert config.progress() is True + + +def test_block_validates_eagerly(): + """A bad value raises at the ``with``, not inside a later request.""" + with pytest.raises(config.ConfigError): + with dataretrieval.configure(concurrency=0): + pass + with pytest.raises(config.ConfigError): + with dataretrieval.configure(retries=-1): + pass + with pytest.raises(config.ConfigError): + with dataretrieval.configure(parallel_chunks=0): + pass + with pytest.raises(config.ConfigError): + with dataretrieval.configure(progress="flase"): + pass + + +@pytest.mark.parametrize( + ("kwargs", "expected"), + [ + ({"api_key": 123}, "string"), + ({"concurrency": 1.5}, "integer"), + ({"concurrency": "8"}, "integer"), + ({"retries": "2"}, "integer"), + ({"progress": []}, "bool"), + ({"parallel_chunks": True}, "integer"), + ({"profile": 123}, "string"), + ], +) +def test_block_rejects_values_outside_annotated_types(kwargs, expected): + with pytest.raises(config.ConfigError, match=expected): + with dataretrieval.configure(**kwargs): + pass + + +def test_block_accepts_ints_and_strings(): + with dataretrieval.configure(concurrency="unbounded"): + assert config.concurrency() is None + with dataretrieval.configure(concurrency=8): + assert config.concurrency() == 8 + with dataretrieval.configure(progress=False): + assert config.progress() is False + with dataretrieval.configure(progress=True): + assert config.progress() is True + + +# --- isolation (the point of issue #352) --------------------------------- + + +def test_threads_do_not_leak_credentials_into_each_other(): + """Two threads in different blocks see different keys. + + This is the concurrency complaint in #352: ``os.environ`` is + process-global, so it cannot express this. + """ + seen: dict[str, str | None] = {} + started = threading.Barrier(2) + + def worker(name: str, key: str) -> None: + with dataretrieval.configure(api_key=key): + started.wait(timeout=5) # force the blocks to overlap in time + seen[name] = config.api_key() + + threads = [ + threading.Thread(target=worker, args=("a", "key-a")), + threading.Thread(target=worker, args=("b", "key-b")), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + assert seen == {"a": "key-a", "b": "key-b"} + + +def test_asyncio_tasks_do_not_leak_credentials_into_each_other(): + """Concurrent asyncio tasks each keep their own key.""" + + async def worker(key: str) -> str | None: + with dataretrieval.configure(api_key=key): + await asyncio.sleep(0) # yield, letting the other task interleave + return config.api_key() + + async def main() -> list[str | None]: + return list(await asyncio.gather(worker("key-a"), worker("key-b"))) + + assert asyncio.run(main()) == ["key-a", "key-b"] + + +# --- the file ------------------------------------------------------------ + + +def test_profile_layers_over_top_level(config_file, monkeypatch): + config_file( + 'api_key = "shared"\nconcurrency = 4\n\n' + '[profiles.bulk]\nconcurrency = "unbounded"\n' + ) + with dataretrieval.configure(profile="bulk"): + assert config.concurrency() is None # from the profile + assert config.api_key() == "shared" # inherited from the top level + assert config.concurrency() == 4 # outside the block, top level again + + +def test_profile_selected_by_env(config_file, monkeypatch): + config_file("concurrency = 4\n\n[profiles.bulk]\nconcurrency = 16\n") + monkeypatch.setenv(config.PROFILE_ENV, "bulk") + assert config.concurrency() == 16 + + +def test_block_profile_outranks_env_profile(config_file, monkeypatch): + config_file("[profiles.a]\nconcurrency = 2\n\n[profiles.b]\nconcurrency = 3\n") + monkeypatch.setenv(config.PROFILE_ENV, "a") + with dataretrieval.configure(profile="b"): + assert config.concurrency() == 3 + + +def test_none_profile_selects_top_level(config_file, monkeypatch): + config_file("concurrency = 4\n\n[profiles.bulk]\nconcurrency = 16\n") + monkeypatch.setenv(config.PROFILE_ENV, "bulk") + with dataretrieval.configure(profile=None): + assert config.concurrency() == 4 + assert config.concurrency() == 16 + + +def test_unknown_profile_raises(config_file): + config_file("concurrency = 4\n") + with pytest.raises(config.ConfigError, match="not defined"): + with dataretrieval.configure(profile="nope"): + pass + + +def test_selected_profile_is_ignored_when_there_is_no_file(tmp_path, monkeypatch): + """A lingering ``DATARETRIEVAL_PROFILE`` must not break every request. + + With no config file there are no profiles to select from and the whole + file layer is inert, so the selection is moot rather than a typo. Raising + here would surface from ``_default_headers`` on every call — including + legacy services that never read the config. + """ + monkeypatch.delenv("API_USGS_CONCURRENT") # pinned by the autouse fixture + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(tmp_path / "absent.toml")) + monkeypatch.setenv(config.PROFILE_ENV, "long-gone") + config._reset_file_cache() + + assert config.concurrency() == config.DEFAULT_CONCURRENCY + assert _default_headers(WATERDATA_URL)["User-Agent"].startswith( + "python-dataretrieval/" + ) + + +def test_profile_typed_into_configure_is_checked_even_with_no_file( + tmp_path, monkeypatch +): + """An explicitly named profile is a typo to report, not ambient state. + + The leniency above is for a *lingering export*, which the caller may not + even know is set. A name passed to ``configure`` was just typed, and + silently proceeding on defaults would drop exactly the settings the caller + asked for — so it raises at the ``with``, as that function documents. + """ + monkeypatch.delenv("API_USGS_CONCURRENT") # pinned by the autouse fixture + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(tmp_path / "absent.toml")) + monkeypatch.delenv(config.PROFILE_ENV, raising=False) + config._reset_file_cache() + + with pytest.raises(config.ConfigError, match="no configuration file"): + with dataretrieval.configure(profile="also-gone"): + pass + + +def test_missing_file_is_not_an_error(tmp_path, monkeypatch): + monkeypatch.delenv("API_USGS_CONCURRENT") # pinned by the autouse fixture + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(tmp_path / "absent.toml")) + config._reset_file_cache() + assert config.concurrency() == config.DEFAULT_CONCURRENCY + + +def test_malformed_file_raises_pointing_at_the_file(config_file): + path = config_file("api_key = \n") + with pytest.raises(config.ConfigError) as excinfo: + config.api_key() + assert "not valid TOML" in str(excinfo.value) + assert str(path) in str(excinfo.value) + + +def test_non_utf8_file_raises_config_error(config_file): + path = config_file("") + path.write_bytes(b'api_key = "\xff"\n') + with pytest.raises(config.ConfigError, match="not valid UTF-8"): + config.api_key() + + +def test_config_path_must_not_be_a_directory(tmp_path, monkeypatch): + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(tmp_path)) + monkeypatch.delenv("API_USGS_CONCURRENT", raising=False) + config._reset_file_cache() + with pytest.raises(config.ConfigError, match="directory"): + config.concurrency() + + +@pytest.mark.skipif(os.name != "posix", reason="needs /dev/null") +def test_dev_null_config_path_means_no_configuration(monkeypatch): + """``DATARETRIEVAL_CONFIG=/dev/null`` is how a run isolates itself. + + A character device (or the FIFO from process substitution) reads as empty, + which is exactly "no configuration". Rejecting it would raise from + ``_default_headers`` on every request -- the opposite of what the caller + asked for. + """ + monkeypatch.delenv("API_USGS_CONCURRENT", raising=False) + monkeypatch.delenv("API_USGS_PAT", raising=False) + monkeypatch.setenv(config.CONFIG_PATH_ENV, "/dev/null") + config._reset_file_cache() + assert config.api_key() is None + assert config.concurrency() == config.DEFAULT_CONCURRENCY + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX directory permissions") +def test_inaccessible_config_path_raises(tmp_path, monkeypatch): + parent = tmp_path / "blocked" + parent.mkdir() + path = parent / "config.toml" + path.write_text("concurrency = 4\n") + parent.chmod(0) + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(path)) + monkeypatch.delenv("API_USGS_CONCURRENT", raising=False) + config._reset_file_cache() + try: + try: + path.stat() + except PermissionError: + pass + else: # pragma: no cover - root or a filesystem that ignores mode bits + pytest.skip("filesystem does not enforce directory mode bits") + with pytest.raises(config.ConfigError, match="could not access"): + config.concurrency() + finally: + parent.chmod(0o700) + + +def test_unknown_setting_warns_but_is_ignored(config_file): + config_file('concurrency = 4\napi_kye = "typo"\n') + with pytest.warns(UserWarning, match="unknown setting"): + assert config.concurrency() == 4 + + +def test_unknown_table_raises(config_file): + """A profile written as ``[bulk]`` instead of ``[profiles.bulk]``.""" + config_file("[bulk]\nconcurrency = 4\n") + with pytest.raises(config.ConfigError, match="unknown table"): + config.concurrency() + + +def test_typed_toml_values_are_normalized(config_file): + """``tomllib`` returns typed values that normalize into shared parsers.""" + config_file("concurrency = 16\nretries = 0\nprogress = true\n") + assert config.concurrency() == 16 + assert config.retries() == 0 + assert config.progress() is True + + +@pytest.mark.parametrize( + "text", + [ + "api_key = true\n", + 'concurrency = "8"\n', + 'retries = "2"\n', + "progress = 17\n", + "parallel_chunks = true\n", + ], +) +def test_toml_rejects_wrong_scalar_types(config_file, text): + config_file(text) + with pytest.raises(config.ConfigError): + config.parallel_chunks() + + +def test_file_edit_is_picked_up(config_file, monkeypatch): + path = config_file("concurrency = 4\n") + assert config.concurrency() == 4 + original = path.stat() + path.write_text("concurrency = 8\n") + os.utime(path, ns=(original.st_atime_ns, original.st_mtime_ns)) + # Windows ctime is creation time, so unchanged metadata must fall back to + # comparing raw content before the parsed cache is reused. + monkeypatch.setattr(config.os, "name", "nt") + assert config.concurrency() == 8 + + +def test_explicit_config_path_is_expanded(monkeypatch): + monkeypatch.setenv(config.CONFIG_PATH_ENV, "~/somewhere/config.toml") + assert str(config.config_path()).startswith(os.path.expanduser("~")) + assert "~" not in str(config.config_path()) + + +def test_relative_config_path_follows_the_working_directory(tmp_path, monkeypatch): + """A relative ``DATARETRIEVAL_CONFIG`` is resolved against the *current* cwd. + + The path memo keys on the working directory for exactly this reason: a + scheduler or notebook that sets a relative path and chdirs per job would + otherwise keep serving the first job's credentials for the life of the + process, with ``show_config()`` reporting the stale path as current. + """ + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + (first / "config.toml").write_text("concurrency = 4\n") + (second / "config.toml").write_text("concurrency = 9\n") + monkeypatch.delenv("API_USGS_CONCURRENT", raising=False) + monkeypatch.setenv(config.CONFIG_PATH_ENV, "config.toml") + + monkeypatch.chdir(first) + config._reset_file_cache() + assert config.config_path() == first / "config.toml" + assert config.concurrency() == 4 + + monkeypatch.chdir(second) + assert config.config_path() == second / "config.toml" + assert config.concurrency() == 9 + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX file modes") +def test_world_readable_file_with_a_key_warns(tmp_path, monkeypatch): + path = tmp_path / "config.toml" + path.write_text('api_key = "secret"\n') + path.chmod(0o644) + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(path)) + monkeypatch.delenv("API_USGS_PAT", raising=False) + config._reset_file_cache() + with pytest.warns(UserWarning, match="readable by other users"): + assert config.api_key() == "secret" + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX file modes") +def test_permission_change_is_checked_on_cached_file(config_file): + path = config_file('api_key = "secret"\n') + assert config.api_key() == "secret" + path.chmod(0o644) + with pytest.warns(UserWarning, match="readable by other users"): + assert config.api_key() == "secret" + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX file modes") +def test_no_permission_warning_without_a_key(tmp_path, monkeypatch, recwarn): + path = tmp_path / "config.toml" + path.write_text("concurrency = 4\n") + path.chmod(0o644) + monkeypatch.setenv(config.CONFIG_PATH_ENV, str(path)) + monkeypatch.delenv("API_USGS_CONCURRENT", raising=False) + config._reset_file_cache() + assert config.concurrency() == 4 + assert not [w for w in recwarn if "readable by other users" in str(w.message)] + + +# --- value grammar ------------------------------------------------------- + + +def test_api_key_is_stripped_and_blank_means_none(monkeypatch): + monkeypatch.setenv("API_USGS_PAT", " key-with-newline\n") + assert config.api_key() == "key-with-newline" + monkeypatch.setenv("API_USGS_PAT", " ") + assert config.api_key() is None + + +def test_blank_numeric_env_falls_back_to_the_default(monkeypatch): + monkeypatch.setenv("API_USGS_CONCURRENT", "") + monkeypatch.setenv("API_USGS_RETRIES", "") + assert config.concurrency() == config.DEFAULT_CONCURRENCY + assert config.retries() == config.DEFAULT_RETRIES + + +def test_blank_progress_env_means_off_not_unset(monkeypatch): + """Preserved from the pre-config behavior: blank disables the line.""" + monkeypatch.setenv("API_USGS_PROGRESS", "") + assert config.progress() is False + + +@pytest.mark.parametrize("value", ["0", "false", "no", "off", "FALSE"]) +def test_progress_falsey_values(monkeypatch, value): + monkeypatch.setenv("API_USGS_PROGRESS", value) + assert config.progress() is False + + +@pytest.mark.parametrize("value", ["1", "true", "yes", "on"]) +def test_progress_truthy_values(monkeypatch, value): + monkeypatch.setenv("API_USGS_PROGRESS", value) + assert config.progress() is True + + +def test_legacy_unknown_progress_env_still_means_on(monkeypatch): + monkeypatch.setenv("API_USGS_PROGRESS", "legacy-nonempty-value") + assert config.progress() is True + + +@pytest.mark.parametrize("value", ["nope", "-1", "0"]) +def test_invalid_concurrency_raises(monkeypatch, value): + monkeypatch.setenv("API_USGS_CONCURRENT", value) + with pytest.raises(ValueError): # ConfigError is a ValueError + config.concurrency() + + +def test_unbounded_concurrency(monkeypatch): + monkeypatch.setenv("API_USGS_CONCURRENT", "unbounded") + assert config.concurrency() is None + + +def test_error_message_names_the_source(config_file, monkeypatch): + monkeypatch.setenv("API_USGS_CONCURRENT", "nope") + with pytest.raises(config.ConfigError, match=r"\$?API_USGS_CONCURRENT"): + config.concurrency() + monkeypatch.delenv("API_USGS_CONCURRENT") + path = config_file('concurrency = "nope"\n') + with pytest.raises(config.ConfigError, match=str(path)): + config.concurrency() + + +# --- security ------------------------------------------------------------ + + +def test_show_config_never_prints_the_key(monkeypatch): + monkeypatch.setenv("API_USGS_PAT", "super-secret-value") + out = io.StringIO() + dataretrieval.show_config(stream=out) + text = out.getvalue() + assert "super-secret-value" not in text + assert "" in text + assert "$API_USGS_PAT" in text # provenance is still reported + + +def test_show_config_reports_absent_key(monkeypatch): + monkeypatch.delenv("API_USGS_PAT", raising=False) + out = io.StringIO() + dataretrieval.show_config(stream=out) + assert "" in out.getvalue() + + +def test_file_sourced_key_is_still_host_scoped(config_file): + """A key from a file gets the same host scoping as one from the env.""" + config_file('api_key = "file-key"\n') + assert _default_headers(WATERDATA_URL)["X-Api-Key"] == "file-key" + assert "X-Api-Key" not in _default_headers("https://example.com/data") + assert "X-Api-Key" not in _default_headers( + "https://api.waterdata.usgs.gov.evil.com/x" + ) + + +def test_block_sourced_key_is_still_host_scoped(): + with dataretrieval.configure(api_key="block-key"): + assert _default_headers(WATERDATA_URL)["X-Api-Key"] == "block-key" + assert "X-Api-Key" not in _default_headers("https://example.com/data") + + +def test_no_public_getter_accepts_a_credential_parameter(): + """Guards the ``**queryables`` catch-all. + + Every Water Data getter forwards unknown keywords as OGC query + parameters, so a getter that grew an ``api_key`` or ``session`` + parameter could serialize a credential into a URL. Credentials must + arrive through ``dataretrieval.configure`` instead. + """ + import inspect + + from dataretrieval import waterdata + + offenders = [] + for name in waterdata.__all__: + obj = getattr(waterdata, name) + if not callable(obj) or inspect.isclass(obj): + continue + try: + params = inspect.signature(obj).parameters + except (TypeError, ValueError): # pragma: no cover - builtins + continue + for forbidden in ("api_key", "session", "token", "apikey"): + if forbidden in params: + offenders.append(f"{name}({forbidden}=)") + assert not offenders, ( + "public getters must not take credential parameters: " + ", ".join(offenders) + ) + + +@pytest.mark.parametrize("allowed", ["session", "session_id", "sampling_session"]) +def test_session_is_not_treated_as_a_credential(allowed): + """``session`` carries no secret, and the queryable namespace is the + server's — a substring rule would make any future field containing it + unreachable behind a credentials message that misstates the problem.""" + from dataretrieval.waterdata.utils import _flatten_queryables + + assert _flatten_queryables({"queryables": {allowed: 1}}) == {allowed: 1} + + +@pytest.mark.parametrize( + "forbidden", + ["api_key", "apikey", "apiKey", "API_KEY", "api-key", "token"], +) +def test_credential_keyword_cannot_enter_queryables(forbidden): + from dataretrieval import waterdata + + with pytest.raises(TypeError, match=forbidden): + waterdata.get_daily( + monitoring_location_id="USGS-01646500", **{forbidden: "secret"} + ) + + +# --- wiring into the rest of the package --------------------------------- + + +def test_retry_policy_reads_the_block(): + from dataretrieval.ogc.retry import RetryPolicy + + with dataretrieval.configure(retries=3): + assert RetryPolicy.from_config().max_retries == 3 + + +def test_parallel_chunks_baseline_comes_from_config(config_file): + from dataretrieval.ogc.chunking import parallel_chunks + + assert config.parallel_chunks() == 1 + config_file("parallel_chunks = 8\n") + assert config.parallel_chunks() == 8 + with parallel_chunks(2): # an explicit block still wins over the file + assert config.parallel_chunks() == 2 + assert config.parallel_chunks() == 8 + + +def test_parallel_chunks_and_configure_share_one_mechanism(): + """``parallel_chunks(n)`` is sugar for ``configure(parallel_chunks=n)``. + + They must not be two competing scopes: whichever block is innermost wins, + so ``show_config()`` always reports the value the chunker will use. + """ + from dataretrieval.ogc.chunking import parallel_chunks + + with parallel_chunks(2): + with dataretrieval.configure(parallel_chunks=8): + assert config.parallel_chunks() == 8 + assert config.parallel_chunks() == 2 + + with dataretrieval.configure(parallel_chunks=8): + with parallel_chunks(2): + assert config.parallel_chunks() == 2 + assert config.parallel_chunks() == 8 + + +def test_parallel_chunks_has_no_environment_variable(): + """It spends quota, so it is deliberately file/block-only (see ENV_VARS).""" + assert "parallel_chunks" not in config.ENV_VARS + assert "parallel_chunks" in config.SETTINGS + + +def test_progress_reporter_reads_the_block(): + from dataretrieval.ogc.progress import ProgressReporter + + with dataretrieval.configure(progress=True): + assert ProgressReporter(stream=io.StringIO()).enabled + with dataretrieval.configure(progress=False): + assert not ProgressReporter(stream=io.StringIO()).enabled + + +# --- review regressions -------------------------------------------------- + + +def test_blank_env_does_not_mask_the_config_file(config_file, monkeypatch): + """A blank-but-set env var must not shadow a configured file. + + Container and CI tooling routinely materializes one (``docker run -e + API_USGS_PAT`` with nothing to pass, a workflow secret absent on a fork). + Letting that outrank the file silently dropped the API key and sent every + request unauthenticated. + """ + config_file('api_key = "file-key"\nconcurrency = 4\nretries = 7\nprogress = true\n') + for env in config.ENV_VARS.values(): + monkeypatch.setenv(env, "") + + assert config.api_key() == "file-key" + assert config.concurrency() == 4 + assert config.retries() == 7 + # ``progress`` is the documented exception: a blank API_USGS_PROGRESS has + # always meant "off", so for that setting blank *is* a value and outranks + # the file. The asymmetry is declared once, in config._BLANK_MEANS_SET. + assert config.progress() is False + assert set(config._BLANK_MEANS_SET) == {"progress"} + + +def test_blank_progress_env_keeps_its_legacy_meaning(monkeypatch): + """With no file, blank keeps the environment-only meaning it always had.""" + monkeypatch.setenv("API_USGS_PROGRESS", "") + monkeypatch.setenv("API_USGS_CONCURRENT", "") + assert config.progress() is False # blank has always meant "off" + assert config.concurrency() == config.DEFAULT_CONCURRENCY + + +def test_config_error_is_in_the_error_taxonomy(): + """A broken config surfaces from inside a getter, so it must be catchable.""" + import dataretrieval.exceptions as exceptions + + assert issubclass(config.ConfigError, exceptions.DataRetrievalError) + assert issubclass(config.ConfigError, ValueError) # legacy handlers still work + assert config.ConfigError is exceptions.ConfigError + + +def test_show_config_reports_a_broken_file_instead_of_raising(config_file): + """The tool that explains a configuration must survive a broken one.""" + config_file("this is not = valid toml [[[\n") + out = io.StringIO() + dataretrieval.show_config(stream=out) # must not raise + text = out.getvalue() + assert "ERROR:" in text + # Every setting still gets a row rather than the report dying part-way. + for name in config.SETTINGS: + assert name in text + + +def test_show_config_reports_a_bad_value_in_its_own_row(monkeypatch): + monkeypatch.setenv("API_USGS_CONCURRENT", "nope") + out = io.StringIO() + dataretrieval.show_config(stream=out) + text = out.getvalue() + assert " list[str]: @pytest.fixture(autouse=True) -def _pin_chunker_env(monkeypatch): +def _pin_chunker_env(monkeypatch, tmp_path): """Pin every test to one connection and no retries. Production defaults ``API_USGS_CONCURRENT`` to 32 and @@ -73,6 +75,14 @@ def _pin_chunker_env(monkeypatch): ``API_USGS_RETRIES=0`` makes a single transient surface immediately rather than be retried. Concurrency and retry tests opt in by overriding the env inside their body. + + Also points ``DATARETRIEVAL_CONFIG`` at a path that does not exist, so a + developer's real ``~/.dataretrieval/config.toml`` (which may hold an API + key or a raised concurrency) can never influence a test run. Config tests + opt in by pointing the variable at a file they wrote. """ monkeypatch.setenv("API_USGS_CONCURRENT", "1") monkeypatch.setenv("API_USGS_RETRIES", "0") + monkeypatch.setenv("DATARETRIEVAL_CONFIG", str(tmp_path / "no-such-config.toml")) + monkeypatch.delenv("DATARETRIEVAL_PROFILE", raising=False) + config._reset_file_cache() diff --git a/tests/waterdata_chunking_test.py b/tests/waterdata_chunking_test.py index ddefafcc..6f21fbf6 100644 --- a/tests/waterdata_chunking_test.py +++ b/tests/waterdata_chunking_test.py @@ -31,6 +31,8 @@ import pandas as pd import pytest +import dataretrieval +from dataretrieval import config as _config from dataretrieval.exceptions import ( DataRetrievalError, RateLimited, @@ -44,7 +46,6 @@ from dataretrieval.ogc.chunking import ( ChunkedCall, _chunked_client, - _parallel_chunks, get_active_client, multi_value_chunked, parallel_chunks, @@ -721,6 +722,48 @@ async def fetch(args): assert sorted(df["id"].tolist()) == sorted(sites) +def test_resume_reads_concurrency_from_the_caller_not_the_snapshot(monkeypatch): + """A ``configure()`` block around a ``resume()`` must actually take effect. + + ``resume()`` drives the call inside the context snapshot taken at + construction, so a ContextVar set afterwards is invisible in there. The + concurrency cap is the one dial a caller adjusts precisely *when* retrying + -- the documented recovery from ``QuotaExhausted`` is to wait and re-issue + more gently -- so it is resolved outside the snapshot. Before that, only + the ``API_USGS_CONCURRENT`` spelling worked (``os.environ`` is not + captured by a context copy) and the ``configure()`` spelling was silently + ignored. + """ + state = {"calls": 0} + + async def fetch(args): + state["calls"] += 1 + if state["calls"] == 3: + raise RateLimited("429: Too many requests made.") + sites = list(args["sites"]) + return (pd.DataFrame({"id": sites}), _quota_response(500)) + + sites = ["S" * 10 + str(i) for i in range(16)] + decorated = multi_value_chunked(build_request=_fake_build, url_limit=240)(fetch) + with pytest.raises(QuotaExhausted) as excinfo: + decorated({"sites": sites}) + + # Spy on the real, decorator-built ChunkedCall rather than a hand-made one. + seen: list[int | None] = [] + original_run = _chunking.ChunkedCall._run + + async def spy_run(self, max_concurrent): + seen.append(max_concurrent) + return await original_run(self, max_concurrent) + + monkeypatch.setattr(_chunking.ChunkedCall, "_run", spy_run) + + with dataretrieval.configure(concurrency=2): + excinfo.value.call.resume() + + assert seen == [2], seen + + def test_chunker_passes_through_non_429_runtime_error(): """A non-429 ``RuntimeError`` (e.g. a 500) is not a quota signal; it must propagate unchanged so callers see the real cause.""" @@ -1414,8 +1457,8 @@ def test_iter_sub_args_passthrough_yields_a_copy(): # --- async fan-out path ---------------------------------------------------- # # Every sub-request is gathered over one ``httpx.AsyncClient`` and -# concurrency is bounded by an ``asyncio.Semaphore`` sized from -# ``API_USGS_CONCURRENT`` (the client's connection pool is sized to +# concurrency is bounded by an ``asyncio.Semaphore`` sized from the effective +# configuration (the client's connection pool is sized to # match, but the semaphore is the throttle — see ``ChunkedCall._run``). # The conftest's ``_pin_chunker_env`` autouse pins # ``API_USGS_CONCURRENT=1`` (sequential dispatch) for the whole suite; @@ -1643,6 +1686,21 @@ def test_fan_out_in_flight_high_water_mark_is_the_cap( assert in_flight["max"] == expected_high_water +def test_configure_concurrency_controls_dispatch(monkeypatch): + """The highest-precedence block setting reaches the execution semaphore.""" + monkeypatch.setenv("API_USGS_CONCURRENT", "1") + in_flight = {"now": 0, "max": 0} + fetch = multi_value_chunked(build_request=_fake_build, url_limit=240)( + _concurrency_probe(in_flight) + ) + + with dataretrieval.configure(concurrency=2): + df, _ = fetch({"sites": list(_EIGHT_SINGLETON_SITES)}) + + assert len(df) == len(_EIGHT_SINGLETON_SITES) + assert in_flight["max"] == 2 + + def test_fan_out_outlives_pool_timeout_on_real_transport(monkeypatch): """End-to-end regression for the pool-timeout starvation bug: the fan-out must survive every pooled connection staying busy past the @@ -1847,19 +1905,19 @@ def test_retry_policy_long_retry_after_escalates(): assert not policy.should_retry(attempt=1, retry_after=120.0) # escalates -def test_retry_policy_from_env(monkeypatch): +def test_retry_policy_from_config(monkeypatch): monkeypatch.setenv("API_USGS_RETRIES", "2") - assert RetryPolicy.from_env().max_retries == 2 + assert RetryPolicy.from_config().max_retries == 2 monkeypatch.setenv("API_USGS_RETRIES", "0") - assert RetryPolicy.from_env().max_retries == 0 + assert RetryPolicy.from_config().max_retries == 0 monkeypatch.delenv("API_USGS_RETRIES", raising=False) - assert RetryPolicy.from_env().max_retries == _RETRIES_DEFAULT + assert RetryPolicy.from_config().max_retries == _RETRIES_DEFAULT monkeypatch.setenv("API_USGS_RETRIES", "-1") with pytest.raises(ValueError): - RetryPolicy.from_env() + RetryPolicy.from_config() monkeypatch.setenv("API_USGS_RETRIES", "lots") with pytest.raises(ValueError): - RetryPolicy.from_env() + RetryPolicy.from_config() def test_retry_policy_rejects_invalid_settings(): @@ -1871,12 +1929,12 @@ def test_retry_policy_rejects_invalid_settings(): RetryPolicy(max_backoff=-1.0) -def test_retry_policy_from_env_honors_monkeypatched_constants(monkeypatch): +def test_retry_policy_from_config_honors_monkeypatched_constants(monkeypatch): # The timing knobs are read from the module constants at call time, so # monkeypatching them (as the module comment promises) takes effect. monkeypatch.setattr(_retry_mod, "_RETRY_MAX_BACKOFF", 0.0) monkeypatch.setattr(_retry_mod, "_RETRY_BASE_BACKOFF", 0.0) - policy = RetryPolicy.from_env() + policy = RetryPolicy.from_config() assert policy.max_backoff == 0.0 and policy.base_backoff == 0.0 @@ -2354,16 +2412,24 @@ def test_cap_does_not_mask_unchunkable(): ChunkPlan(args, _fake_build, url_limit=10, max_chunks=32) -def test_parallel_chunks_publishes_n_on_the_ambient(): - """The context manager publishes ``n`` on the ambient for the block and - restores the previous value on exit — including proper nesting.""" - assert _parallel_chunks.get() == 1 # default (off, = no extra fan-out) +def test_parallel_chunks_publishes_n_as_the_effective_setting(): + """The context manager sets ``n`` for the block and restores the previous + value on exit — including proper nesting. + + ``parallel_chunks(n)`` is sugar for ``configure(parallel_chunks=n)``, so + both forms share one scoping mechanism and the innermost block wins. + Outside any block the configured baseline applies, which is ``1`` — off — + unless a config file raised it.""" + assert _config.parallel_chunks() == 1 # default (off, = no extra fan-out) with parallel_chunks(32): - assert _parallel_chunks.get() == 32 + assert _config.parallel_chunks() == 32 with parallel_chunks(2): - assert _parallel_chunks.get() == 2 - assert _parallel_chunks.get() == 32 # outer restored - assert _parallel_chunks.get() == 1 # default (off) outside any block + assert _config.parallel_chunks() == 2 + assert _config.parallel_chunks() == 32 # outer restored + with dataretrieval.configure(parallel_chunks=4): # the other spelling + assert _config.parallel_chunks() == 4 + assert _config.parallel_chunks() == 32 + assert _config.parallel_chunks() == 1 # default (off) outside any block @pytest.mark.parametrize( @@ -2387,7 +2453,7 @@ def test_parallel_chunks_rejects_non_positive_int(bad): with pytest.raises(ValueError, match="must be a positive integer"): with parallel_chunks(bad): pass - assert _parallel_chunks.get() == 1 # default (off) — unchanged by a rejected call + assert _config.parallel_chunks() == 1 # unchanged by a rejected call def test_parallel_chunks_drives_end_to_end_fan_out(): diff --git a/tests/waterdata_utils_test.py b/tests/waterdata_utils_test.py index c39a8b19..9cf2c3dc 100644 --- a/tests/waterdata_utils_test.py +++ b/tests/waterdata_utils_test.py @@ -36,7 +36,12 @@ ) from dataretrieval.waterdata import get_stats_date_range, get_stats_por from dataretrieval.waterdata.stats import _handle_nesting, get_data -from dataretrieval.waterdata.utils import OGC_API_URL, _finalize_ogc, _get_args +from dataretrieval.waterdata.utils import ( + OGC_API_URL, + _finalize_ogc, + _flatten_queryables, + _get_args, +) _LOGGER_NAME = _utils_module.__name__ @@ -1085,3 +1090,24 @@ def fake_engine_get_ogc_data(args, service, output_id, **k): ): _utils_module.get_ogc_data({"state": "WI"}, "monitoring-locations") assert captured["args"] == {"state": "WI"} + + +@pytest.mark.parametrize( + "name", ["x_api_key", "x-api-key", "api_token", "access_token", "pat", "auth"] +) +def test_credential_shaped_queryables_are_rejected(name): + """The denylist matches spellings, not just a few exact names. + + ``x_api_key`` is the tempting one -- it mirrors the ``X-Api-Key`` header + the README documents -- and an exact-match list let it through into the + query string. + """ + with pytest.raises(TypeError, match="Credentials cannot be passed"): + _flatten_queryables({"queryables": {name: "SECRET"}}) + + +@pytest.mark.parametrize( + "name", ["state_name", "site_type_code", "monitoring_location_id", "qualifier"] +) +def test_real_queryables_still_pass_through(name): + assert _flatten_queryables({"queryables": {name: "v"}}) == {name: "v"}