From 23b8bc1a73e1820d05330e74f4dc22bf2ba0811c Mon Sep 17 00:00:00 2001 From: thodson-usgs Date: Tue, 15 Sep 2026 15:23:51 -0500 Subject: [PATCH] perf(ogc): vectorize feature-frame builds and free parsed page bodies The per-chunk parse runs on the fan-out's event loop, so shaping CPU is on every chunked call's critical path. Separately, each aggregate kept for resume shared its first page's decompressed body, so a one-page-per-chunk fan-out held the whole download in memory until the call finished. Feature frames now take two fast paths. Flat properties build through `pd.DataFrame` rather than `pd.json_normalize`, with nested values found by scanning only object-dtype columns. An all-2D-point page builds its geometry in one `geopandas.points_from_xy` call over two flat coordinate lists rather than per-feature `GeoDataFrame.from_features`, and keeps nested properties as raw dicts the way `from_features` does. Non-point or malformed geometry, null coordinates, and a `geometry` property column fall back to `from_features`; the plain path still normalizes nested properties. Transport frees a page body once it is parsed and reported, but only when it owns the client -- `client is None`, which covers the fan-out's ambient client. A caller-supplied client keeps its page bodies. An aggregate is always a copy with an empty body, so status, headers, URL and elapsed stay readable. ADR 0006 records both contracts. `utils.format_datetime` shaped qw-service responses, lost its last caller in 491eb5c3, and appears in no docs or demos. It now warns through `_deprecation.warn_deprecated` with a 2027-09-15 horizon. Behavior change: the raw aggregate reached through `partial_response` -- on `ChunkInterrupted`, `FanOutInterrupted`, and the `ChunkedCall` they carry -- no longer carries body bytes. What it carried was one arbitrary page's fragment, not the query's data; use the returned frame. `BaseMetadata` is unaffected: it keeps the url, elapsed and headers, and has never retained the response object. Measured on an M3 Pro, Python 3.12, pandas 3.0.5, geopandas 1.1.3, against a `main` worktree. Frame builds, median of seven runs, traced peak separately: - 100k flat point features: 548 -> 101 ms, peak 64.1 -> 20.8 MiB - 50k nested point features: 294 -> 50 ms, peak 33.7 -> 11.2 MiB - 100k features without geometry: 110 -> 84 ms, peak 58.8 -> 15.5 MiB - 50k polygon features: unchanged, as the fallback still handles them - 100k flat plain frames: 211 -> 75 ms Replaying a recorded 93k-row 12-chunk call at zero latency: 0.84 -> 0.42 s, traced peak 165.8 -> 114.0 MB, and 165.8 -> 149.9 MB when pages arrive concurrently. The unfanned peak is unchanged at 211 MB: it is one 40 MB page's decode spike, which neither change touches. Every fast path is verified to produce frames identical to the `from_features` output. 1200 offline tests at 99.02% branch coverage; mypy, ruff, xenon, complexipy and import-linter pass. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014ppUeFxCydyd4n1k6TFXuq --- NEWS.md | 15 ++ dataretrieval/_deprecation.py | 1 + dataretrieval/combining.py | 20 +- dataretrieval/ogc/shaping.py | 99 ++++++- dataretrieval/transport/pagination.py | 32 ++- dataretrieval/utils.py | 11 +- .../0006-service-neutral-transport.rst | 30 ++- tests/transport_test.py | 71 +++++ tests/utils_test.py | 9 +- tests/waterdata_utils_test.py | 246 ++++++++++++++++++ 10 files changed, 507 insertions(+), 27 deletions(-) diff --git a/NEWS.md b/NEWS.md index f4dbd033..7c5e306e 100644 --- a/NEWS.md +++ b/NEWS.md @@ -1,3 +1,18 @@ +**09/15/2026:** Water Data and NGWMN OGC responses build frames faster +for flat properties and point geometries, with unchanged results, +and a fan-out no longer holds each chunk's page body in memory +for the life of the call. +**Behavior change:** the raw aggregate response reached through `partial_response` +-- on `ChunkInterrupted`, `FanOutInterrupted`, and the `ChunkedCall` they carry -- +now has an empty body rather than one arbitrary page's bytes. +Status, headers, URL, and elapsed are unchanged; the data is the returned frame. +Response metadata is unaffected: it has never retained the response object. +The page-body lifetime contract is documented in +[ADR 0006](docs/source/architecture/decisions/0006-service-neutral-transport.rst). +**Deprecation:** `utils.format_datetime` now emits a `DeprecationWarning` +and will be removed on or after 2027-09-15. +Combine the columns with `pandas.to_datetime` directly. + **09/09/2026:** **Bug fix:** code and identifier columns keep their leading zeros. A bare `pandas.read_csv` infers a zero-padded code as a number, so `waterdata.get_samples()` returned parameter code `00060` as `60` and HUC12 `070700050502` as `70700050502`, and `nwis.get_info()` returned `huc_cd` `02060005` as `2060005`. One rule now decides what a code column is — a name ending in `code`, the RDB abbreviation `_cd`, or a name containing `identifier`, `huc`, or `fips` — and every delimited response is parsed through it: the Samples and WQP CSV readers, `rdb.read_rdb` (which reads the names from the RDB header rather than the caller listing them), and the Water Use CSV pages. **Behavior change:** these columns now hold strings. `waterdata.get_samples()`: `USGSpcode`, `Location_HUCEightDigitCode`, `Location_HUCTwelveDigitCode`, `SampleCollectionMethod_Identifier` (`get_samples_summary()` shares the parse; no column in its current profile was affected). `nwis.get_info()`, `nwis.what_sites()`, and `nwis.get_record(service="site")`: `huc_cd`, `state_cd`, `county_cd`, `district_cd`. A comparison against a number — `df["USGSpcode"] == 60` — or a merge onto a numeric key now matches nothing instead of raising, so compare against the padded string (`== "00060"`) or call `.astype(int)` where the number is what you want. **Behavior change:** a count whose name reads as an identifier is numeric again. WQP's `AlternateLocation_IdentifierCount` has been read as text since 05/31/2026 because "Identifier" appears in its name; a name ending in `count` is now excluded from the rule, so the same column has one dtype in every service that reports it. Measurement columns are unchanged, and the `waterdata` OGC getters and `ngwmn` were never affected: their JSON responses deliver codes as strings and numeric coercion there is limited to a fixed list of measurement columns. **Correction to the 1.2.0 notes:** the same fix was applied to the nine `wqp` getters on 05/31/2026 and never recorded here — `wqp.get_results()` and the `what_*` getters have returned HUCs, parameter codes, and FIPS codes as strings since that release. **09/01/2026:** **Announcement:** We at USGS Water Data for the Nation want your feedback! Tell us how we're doing by taking our quick [survey](https://usgswaterresources.gov1.qualtrics.com/jfe/form/SV_07gX8G1DeOtVrH8), available through September 2026. diff --git a/dataretrieval/_deprecation.py b/dataretrieval/_deprecation.py index 4c8f1968..428b4bb3 100644 --- a/dataretrieval/_deprecation.py +++ b/dataretrieval/_deprecation.py @@ -19,6 +19,7 @@ "waterdata.get_cql(service=)": "2027-08-09", "wateruse": "2027-08-11", "ogc.interruptions": "2027-08-25", + "utils.format_datetime": "2027-09-15", } diff --git a/dataretrieval/combining.py b/dataretrieval/combining.py index ae0dec2d..9ddff92d 100644 --- a/dataretrieval/combining.py +++ b/dataretrieval/combining.py @@ -70,6 +70,16 @@ def _set_response_url(response: httpx.Response, url: str | httpx.URL) -> None: response.request = httpx.Request(method=old.method, url=target, headers=old.headers) +def _drop_body(response: httpx.Response) -> None: + """Free a response's fetched body, keeping status/headers/URL readable. + + ``_content`` and ``_text`` are the slots httpx caches a read and a decoded + body in; both are emptied so the two accessors stay valid and agree. + """ + response._content = b"" + response.__dict__.pop("_text", None) + + def _lowest_remaining(responses: list[httpx.Response]) -> httpx.Response: """The response reporting the lowest ``x-ratelimit-remaining``. @@ -102,13 +112,15 @@ def _merge_response( The copy's ``.headers`` are rebuilt as a new ``httpx.Headers`` from ``headers_from``, ``.elapsed`` is set to ``elapsed``, and ``.url`` is - overridden when ``url`` is given. ``base`` and ``headers_from`` are never - mutated, and the new ``httpx.Headers`` means downstream mutations don't - back-propagate into any underlying response — so callers may re-merge - idempotently. This is the one low-level merge used by both pagination + overridden when ``url`` is given, and its body is emptied (ADR 0006). + ``base`` and ``headers_from`` are never mutated, and the new + ``httpx.Headers`` means downstream mutations don't back-propagate into any + underlying response — so callers may re-merge idempotently. This is the + one low-level merge used by both pagination (:func:`~dataretrieval.transport.pagination.paginate`) and the chunked / fan-out aggregation (:func:`_combine_chunk_responses`).""" merged = copy.copy(base) + _drop_body(merged) merged.headers = httpx.Headers(headers_from.headers) merged.elapsed = elapsed if url is not None: diff --git a/dataretrieval/ogc/shaping.py b/dataretrieval/ogc/shaping.py index 7486655e..ce7b1408 100644 --- a/dataretrieval/ogc/shaping.py +++ b/dataretrieval/ogc/shaping.py @@ -11,11 +11,13 @@ from __future__ import annotations import logging +import math import re from typing import Any import httpx import pandas as pd +from pandas.api.types import infer_dtype from dataretrieval._response_metadata import BaseMetadata from dataretrieval.ogc.policy import DEFAULT_DIALECT, OgcDialect @@ -35,6 +37,10 @@ # (EPSG:4269). _CRS = "EPSG:4326" +# What ``infer_dtype`` reports for an object column that cannot hide a dict. +# Named this way round so an unfamiliar label falls through to the scan. +_FLAT_DTYPES = frozenset({"string", "empty"}) + # Whether geopandas is present is a static, environment-level fact, so warn # once here at import time rather than per query/chunk. if not GEOPANDAS: @@ -98,23 +104,108 @@ def _geo_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame: ) +def _properties_frame(features: list[dict[str, Any]]) -> pd.DataFrame: + """Build plain-frame properties, flattening nested dicts with underscores.""" + properties = [feature.get("properties") or {} for feature in features] + frame = pd.DataFrame(properties) + if any(_holds_nested_value(column) for _, column in frame.items()): + return pd.json_normalize(properties, sep="_") + return frame + + +def _holds_nested_value(column: pd.Series) -> bool: + """Whether a built column carries a raw dict that needs flattening.""" + if column.dtype != object: + return False + values = column.to_numpy() + if infer_dtype(values, skipna=True) in _FLAT_DTYPES: + return False + return any(isinstance(value, dict) for value in values) + + def _plain_feature_frame( features: list[dict[str, Any]], *, include_geometry: bool ) -> pd.DataFrame: """Build a plain DataFrame from GeoJSON features.""" - properties = [feature.get("properties") or {} for feature in features] - df = pd.json_normalize(properties, sep="_") + df = _properties_frame(features) df["id"] = [feature.get("id") for feature in features] if include_geometry: _attach_coordinates(df, features) return df +def _point_geometries(features: list[dict[str, Any]]) -> Any: + """Build the geometry array for an all-2D-point page in one vectorized + :func:`geopandas.points_from_xy` call. + + Returns ``None`` when any feature carries a non-point or malformed + geometry, so the caller falls back to :func:`_geo_feature_frame`. A + feature with no geometry stays ``None``, matching that fallback. Two + flat x/y lists rather than coordinate pairs: the paired form is slower. + """ + xs: list[Any] = [] + ys: list[Any] = [] + missing: list[int] = [] + for index, feature in enumerate(features): + geometry = feature.get("geometry") or {} + if not geometry: + missing.append(index) + xs.append(math.nan) + ys.append(math.nan) + continue + xy: Any = geometry.get("coordinates") + if geometry.get("type") != "Point" or not _is_pair(xy): + return None + xs.append(xy[0]) + ys.append(xy[1]) + if len(missing) == len(features): + return gpd.array.from_shapely([None] * len(features)) + try: + points = gpd.points_from_xy(xs, ys) + except (TypeError, ValueError): + return None + points[missing] = None + return points + + +def _is_pair(value: Any) -> bool: + """Whether ``value`` is a two-element coordinate sequence without nulls.""" + return ( + isinstance(value, (list, tuple)) + and len(value) == 2 + and value[0] is not None + and value[1] is not None + ) + + +def _point_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame | None: + """Fast-path GeoDataFrame for an all-2D-point page, or ``None`` to fall + back to :func:`_geo_feature_frame`. + + Keeps nested properties as raw dicts, matching ``from_features``. + Declines pages with a ``geometry`` property to preserve the fallback's + collision handling. + """ + points = _point_geometries(features) + if points is None: + return None + frame = pd.DataFrame([feature.get("properties") or {} for feature in features]) + if "geometry" in frame.columns: + return None + return gpd.GeoDataFrame(frame, geometry=points, crs=_CRS, copy=False) + + def _spatial_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame: """Build a GeoDataFrame from GeoJSON features with ``id`` first.""" - df = _geo_feature_frame(features) + df = _point_feature_frame(features) + if df is None: + df = _geo_feature_frame(features) df["id"] = [f.get("id") for f in features] - return df[["id"] + [col for col in df.columns if col != "id"]] + # Pin both names: the fast path appends ``geometry`` last and + # ``from_features`` emits it first, so only naming them keeps the two + # paths' column order identical across a chunked concat. + ordered = ["id", "geometry"] + return df[ordered + [col for col in df.columns if col not in ordered]] def _get_resp_data( diff --git a/dataretrieval/transport/pagination.py b/dataretrieval/transport/pagination.py index f835e7e2..0618bf95 100644 --- a/dataretrieval/transport/pagination.py +++ b/dataretrieval/transport/pagination.py @@ -18,6 +18,7 @@ from dataretrieval import progress as _progress from dataretrieval.combining import ( _QUOTA_HEADER, + _drop_body, _merge_response, _safe_elapsed, ) @@ -79,6 +80,22 @@ def paginated_failure_message( ) +def _finish_page( + page: httpx.Response, frame: pd.DataFrame, *, release_body: bool +) -> None: + """Report a parsed page and release its body when transport owns it.""" + note_progress() + reporter = _progress.current() + if reporter is not None: + reporter.set_rate_remaining( + page.headers.get(_QUOTA_HEADER), + limit=page.headers.get("x-ratelimit-limit"), + ) + reporter.add_page(rows=len(frame)) + if release_body: + _drop_body(page) + + async def paginate( initial_req: httpx.Request, *, @@ -96,17 +113,6 @@ async def paginate( metadata aggregation. """ logger.debug("Requesting: %s", initial_req.url) - reporter = _progress.current() - - def report_page(page: httpx.Response, frame: pd.DataFrame) -> None: - note_progress() # a walk still receiving pages is not stalled - if reporter is not None: - reporter.set_rate_remaining( - page.headers.get(_QUOTA_HEADER), - limit=page.headers.get("x-ratelimit-limit"), - ) - reporter.add_page(rows=len(frame)) - async with _client_for(client) as session: response = await session.send(initial_req) raise_for_status(response) @@ -124,7 +130,7 @@ def report_page(page: httpx.Response, frame: pd.DataFrame) -> None: frames = [frame] nrows = len(frame) seen: set[Any] = set() - report_page(response, frame) + _finish_page(response, frame, release_body=client is None) while ( cursor is not None @@ -139,7 +145,7 @@ def report_page(page: httpx.Response, frame: pd.DataFrame) -> None: frames.append(frame) nrows += len(frame) total_elapsed += _safe_elapsed(response) - report_page(response, frame) + _finish_page(response, frame, release_body=client is None) except Exception as exc: logger.warning( "Request failed at cursor %r. Data download interrupted.", cursor diff --git a/dataretrieval/utils.py b/dataretrieval/utils.py index 68e3cf2e..7d4bcbc5 100644 --- a/dataretrieval/utils.py +++ b/dataretrieval/utils.py @@ -16,6 +16,7 @@ import pandas as pd +import dataretrieval._deprecation as _deprecation import dataretrieval._querying as _querying import dataretrieval.transport.http as _transport_http from dataretrieval._ambient import Ambient # noqa: F401 # compatibility re-export @@ -58,8 +59,16 @@ def format_datetime( df: ``pandas.DataFrame`` The data frame with a formatted 'datetime' column. + Deprecated: the qw services this shaped responses for are retired and + nothing in the package calls it. Combine the columns with + :func:`pandas.to_datetime` directly. See + :data:`dataretrieval._deprecation.REMOVALS` for the removal horizon. """ - # create a datetime index from the columns in qwdata response + _deprecation.warn_deprecated( + "`utils.format_datetime`", + replacement="a direct `pandas.to_datetime` over the combined columns", + removal=_deprecation.REMOVALS["utils.format_datetime"], + ) df[tz_field] = df[tz_field].map(tz) df["datetime"] = pd.to_datetime( diff --git a/docs/source/architecture/decisions/0006-service-neutral-transport.rst b/docs/source/architecture/decisions/0006-service-neutral-transport.rst index dec857a6..9d357d29 100644 --- a/docs/source/architecture/decisions/0006-service-neutral-transport.rst +++ b/docs/source/architecture/decisions/0006-service-neutral-transport.rst @@ -53,6 +53,22 @@ through transport: - ``dataretrieval.combining`` -- pandas frame and response assembly. Transport returns results *through* it. +An aggregated response describes the call, not a page: +its status, headers, URL, and elapsed are meaningful, and its body is empty. +A page body belongs to one request, not the query's combined data; +retaining page bodies also increases memory use during a fan-out. +The aggregate is always a copy with its body emptied, +regardless of who owns the HTTP client. + +Page-body lifetime depends on client ownership. +When no client is explicitly supplied (``client is None``), +transport releases each page body after parsing and progress reporting. +This includes clients obtained from the ambient fan-out context, +which are internally owned. +An adapter needing bytes must consume them in its page parser. +When a caller explicitly supplies a client, +transport leaves its per-page response bodies unchanged. + Transport depends only on stable package leaves and third-party infrastructure. It must not import OGC modules or service adapters. Service adapters inject request construction, response parsing, cursor extraction, and API-specific @@ -132,6 +148,8 @@ Consequences advice to obtain one is not printed for a service that cannot use it. - The transport package is internal infrastructure, not a new public API contract. +- Reading ``.content`` or ``.text`` off an aggregated response returns empty. + Callers wanting the payload use the returned frame. - Keeping presentation and frame assembly out means transport is roughly 570 lines across five modules, each recognizably HTTP execution policy. Retry is the one complex module, because two independent bounds are what make retry @@ -150,14 +168,15 @@ which failures are re-sent, cancellation, no-partial fan-out behavior, and credential host scoping. The exemptions above are covered by the liveness and retry tests over excluded waits and the first-attempt case. Next-page link validation is covered by the shared link-policy tests over foreign hosts and -embedded userinfo. +embedded userinfo. ``test_merge_response_empties_the_body_but_keeps_the_rest`` +pins the empty-body contract and the metadata that must survive it. Notes ----- -The waiting-time, next-page-link, and credentials-leaf clauses were added -after the original decision, consolidating under ADR 0000 the rules the code -was stating in prose -- the budget exemptions were argued in five places +The waiting-time, next-page-link, credentials-leaf, and empty-body clauses were +added after the original decision, consolidating under ADR 0000 the rules the +code was stating in prose -- the budget exemptions were argued in five places across ``transport/retry.py``, ``transport/liveness.py``, and ``transport/fanout.py``. @@ -165,3 +184,6 @@ One sentence of the original Decision was also corrected rather than added to: it scoped automatic retry to "gateway 5xx", which was never true of a fanned-out call -- those re-send any 5xx, and only the single-shot adapters are limited to the gateway statuses. The decision is unchanged; the sentence now describes it. + +The page-body lifetime clause was corrected to distinguish internally owned +clients from explicitly caller-supplied clients. diff --git a/tests/transport_test.py b/tests/transport_test.py index 72b09014..1c840223 100644 --- a/tests/transport_test.py +++ b/tests/transport_test.py @@ -14,6 +14,7 @@ import dataretrieval.exceptions as exceptions import dataretrieval.transport.liveness as liveness +import dataretrieval.transport.pagination as pagination import dataretrieval.transport.retry as retry from dataretrieval._querying import _raise_for_status from dataretrieval.exceptions import ( @@ -67,6 +68,76 @@ async def follow(cursor: str, session: httpx.AsyncClient) -> httpx.Response: assert response.headers["x-ratelimit-remaining"] == "8" +@pytest.mark.parametrize("client_source", ["injected", "fanout", "new"]) +def test_paginate_releases_only_owned_page_bodies(client_source) -> None: + pages: list[httpx.Response] = [] + payloads = { + "/1": {"value": 1, "next": "https://example.test/2"}, + "/2": {"value": 2, "next": None}, + } + + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=payloads[request.url.path]) + + async def retain(response: httpx.Response) -> None: + if pages: + assert bool(pages[-1].content) is (client_source == "injected") + assert bool(pages[-1].text) is (client_source == "injected") + await response.aread() + assert response.text + pages.append(response) + + def parse(response: httpx.Response) -> tuple[pd.DataFrame, str | None]: + body = response.json() + return pd.DataFrame({"value": [body["value"]]}), body["next"] + + async def follow(cursor: str, session: httpx.AsyncClient) -> httpx.Response: + return await session.get(cursor) + + async def run() -> tuple[pd.DataFrame, httpx.Response]: + session = httpx.AsyncClient( + transport=httpx.MockTransport(respond), + event_hooks={"response": [retain]}, + ) + try: + with ( + mock.patch.object( + pagination, + "active_client", + return_value=session if client_source == "fanout" else None, + ), + mock.patch.object( + pagination, "open_async_client", return_value=session + ), + ): + result = await paginate( + httpx.Request("GET", "https://example.test/1"), + parse_response=parse, + follow_up=follow, + raise_for_status=_raise_for_status, + client=session if client_source == "injected" else None, + ) + assert session.is_closed is (client_source == "new") + return result + finally: + await session.aclose() + + frame, aggregate = asyncio.run(run()) + + assert frame["value"].tolist() == [1, 2] + assert len(pages) == 2 + assert aggregate.content == b"" + assert aggregate.text == "" + assert aggregate.url == pages[0].url + for page in pages: + if client_source == "injected": + assert page.json() == payloads[page.url.path] + assert page.text + else: + assert page.content == b"" + assert page.text == "" + + def test_paginate_stops_on_repeated_cursor_and_respects_row_cap() -> None: first = _response(url="https://example.test/page/1") second = _response(url="https://example.test/page/2") diff --git a/tests/utils_test.py b/tests/utils_test.py index c2fcb6c2..beb9103a 100644 --- a/tests/utils_test.py +++ b/tests/utils_test.py @@ -8,6 +8,7 @@ import pytest from dataretrieval import _querying, _wqx, exceptions, nwis, utils +from dataretrieval._deprecation import REMOVALS class Test_Ambient: @@ -608,7 +609,13 @@ def test_retrying_get_maps_invalid_url(monkeypatch): class TestFormatDatetime: """``format_datetime`` joins the three columns NWIS RDB splits a timestamp across, and is the only place the package parses a local time - with a named zone.""" + with a named zone. It is deprecated -- the qw services it shaped + responses for are retired -- so every call here also warns.""" + + def test_is_deprecated_with_the_published_horizon(self): + df = pd.DataFrame({"d": ["2020-01-01"], "t": ["12:00"], "z": ["EST"]}) + with pytest.warns(DeprecationWarning, match=REMOVALS["utils.format_datetime"]): + utils.format_datetime(df, "d", "t", "z") def test_joins_date_time_and_zone_into_utc(self): df = pd.DataFrame( diff --git a/tests/waterdata_utils_test.py b/tests/waterdata_utils_test.py index f0d6eaf2..1924e457 100644 --- a/tests/waterdata_utils_test.py +++ b/tests/waterdata_utils_test.py @@ -401,6 +401,34 @@ def test_next_req_url_stops_when_no_features(): assert _next_req_url(resp, body=body) is None +def test_merge_response_empties_the_body_but_keeps_the_rest(): + """``_drop_body`` writes httpx's private ``_content`` slot, so a rename + upstream would silently stop freeing anything -- ``.content`` would keep + returning real bytes and only the heap would regress. Pin both halves: + the aggregate carries no body, and status/headers/URL still read.""" + from dataretrieval.combining import _merge_response + + page = httpx.Response(200, headers={"x-page": "1"}, content=b'{"features": []}') + page._request = httpx.Request("GET", "https://example.com/items?page=1") + assert page.text # an adapter that parses via .text caches the decoded body + + merged = _merge_response( + page, + headers_from=page, + elapsed=datetime.timedelta(seconds=2), + url="https://example.com/items", + ) + + assert merged.content == b"" + # ``.text`` is cached in its own slot and rides along on the shallow copy, + # so clearing only ``_content`` would leave the two accessors disagreeing. + assert merged.text == "" + assert page.content == b'{"features": []}' # the base is never mutated + assert merged.status_code == 200 + assert merged.headers["x-page"] == "1" + assert str(merged.url) == "https://example.com/items" + + def test_walk_pages_does_not_mutate_initial_response(): """The aggregated response returned from ``_walk_pages`` is built via ``_merge_response``, which returns a new copy. @@ -1393,6 +1421,224 @@ def test_real_queryables_still_pass_through(name): assert _flatten_queryables({"queryables": {name: "v"}}) == {name: "v"} +# --------------------------------------------------------------------------- +# Feature-frame fast paths (vectorized points, flat properties) +# --------------------------------------------------------------------------- + +_POINT_FEATURES = [ + { + "id": "f-1", + "properties": {"id": "wire-1", "value": "1", "site": "USGS-A"}, + "geometry": {"type": "Point", "coordinates": [-77.1, 38.9]}, + }, + { + "id": "f-2", + "properties": {"id": "wire-2", "value": "2", "site": "USGS-B"}, + "geometry": {"type": "Point", "coordinates": [-80.0, 40.0]}, + }, + { + # No geometry at all (NGWMN observation shape). + "id": "f-3", + "properties": {"id": "wire-3", "value": "3", "site": "USGS-C"}, + }, +] + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +def test_spatial_fast_path_matches_from_features(): + """The vectorized point build is a pure speedup: identical frame, + column order, CRS, and missing-geometry handling as the + ``from_features`` fallback — including the feature-level ``id`` + overwriting a properties ``id`` column.""" + fast = _shaping_module._spatial_feature_frame(_POINT_FEATURES) + fallback = _shaping_module._geo_feature_frame(_POINT_FEATURES) + fallback["id"] = [feature["id"] for feature in _POINT_FEATURES] + fallback = fallback[["id", "geometry", "value", "site"]] + + pd.testing.assert_frame_equal(fast, fallback) + assert fast.crs == "EPSG:4326" + assert list(fast["id"]) == ["f-1", "f-2", "f-3"] + assert fast.geometry.isna().tolist() == [False, False, True] + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +def test_spatial_non_point_geometry_uses_from_features(): + """A non-point geometry anywhere disables the vectorized build; the + result still carries the real geometry via ``from_features``.""" + features = _POINT_FEATURES[:1] + [ + { + "id": "f-poly", + "properties": {"value": "4"}, + "geometry": { + "type": "Polygon", + "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]], + }, + } + ] + with mock.patch.object(pd, "DataFrame", side_effect=AssertionError): + assert _shaping_module._point_feature_frame(features) is None + + df = _shaping_module._spatial_feature_frame(features) + fallback = _shaping_module._geo_feature_frame(features) + fallback["id"] = [feature["id"] for feature in features] + pd.testing.assert_frame_equal(df, fallback[df.columns]) + assert df.geometry.iloc[1].geom_type == "Polygon" + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +@pytest.mark.parametrize( + "coordinates", + [ + pytest.param([1.0, 2.0, 3.0], id="3d"), + pytest.param([1.0], id="single"), + pytest.param({"x": 1.0, "y": 2.0}, id="mapping"), + # A pair of non-scalars passes the shape check and is refused by the + # array build instead, which is the only param reaching that branch. + pytest.param([[1.0, 2.0], [3.0, 4.0]], id="ragged"), + ], +) +def test_point_geometries_rejects_malformed_coordinates(coordinates): + """Coordinates that are not a usable 2-element pair disable the fast path + rather than building a wrong geometry.""" + features = [ + { + "id": "f", + "properties": {}, + "geometry": {"type": "Point", "coordinates": coordinates}, + } + ] + assert _shaping_module._point_geometries(features) is None + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +@pytest.mark.parametrize( + "properties", + [ + [{"nested": {"x": {"y": 1}}}, {"nested": {"x": "z"}}], + [{"nested": None}, {"nested": "scalar"}, {}, {"nested": {"x": "y"}}], + [{"nested": [1, 2]}, {"other": 3}, {"nested": {"x": "y"}}], + ], + ids=["nested", "mixed-sparse", "list-and-dict"], +) +def test_spatial_nested_properties_match_the_fallback(properties): + features = [ + { + "id": str(index), + "properties": values, + "geometry": {"type": "Point", "coordinates": [index, 2]}, + } + for index, values in enumerate(properties) + ] + fallback = _shaping_module._geo_feature_frame(features) + with ( + mock.patch.object( + _shaping_module, "_holds_nested_value", side_effect=AssertionError + ), + mock.patch.object(pd, "json_normalize", side_effect=AssertionError), + ): + fast = _shaping_module._point_feature_frame(features) + all_points = _shaping_module._spatial_feature_frame(features) + assert fast is not None + pd.testing.assert_frame_equal(fast[fallback.columns], fallback) + + mixed = features + [ + { + "id": "line", + "properties": properties[-1], + "geometry": {"type": "LineString", "coordinates": [[0, 0], [1, 1]]}, + } + ] + with_a_line = _shaping_module._spatial_feature_frame(mixed) + assert list(all_points.columns) == list(with_a_line.columns) + assert "nested" in all_points.columns and "nested_x" not in all_points.columns + assert list(pd.concat([all_points, with_a_line]).columns) == list( + all_points.columns + ) + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +@pytest.mark.parametrize("coordinates", [[None, None], [None, 2], [1, None]]) +def test_spatial_null_coordinates_match_from_features(coordinates): + features = [ + { + "properties": {"value": 1}, + "geometry": {"type": "Point", "coordinates": coordinates}, + } + ] + assert _shaping_module._point_feature_frame(features) is None + if coordinates == [None, None]: + fallback = _shaping_module._geo_feature_frame(features) + result = _shaping_module._spatial_feature_frame(features) + pd.testing.assert_frame_equal(result[fallback.columns], fallback) + assert result.geometry.iloc[0].wkt == "POINT EMPTY" + else: + with pytest.raises(TypeError) as expected: + _shaping_module._geo_feature_frame(features) + with pytest.raises(type(expected.value)) as actual: + _shaping_module._spatial_feature_frame(features) + assert str(actual.value) == str(expected.value) + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +@pytest.mark.parametrize("properties", [{}, None, {"nested": {"x": 1}}]) +def test_spatial_all_missing_geometry_skips_point_construction(properties): + features = [ + {"properties": properties}, + {"properties": properties, "geometry": None}, + {"properties": properties, "geometry": {}}, + ] + fallback = _shaping_module._geo_feature_frame(features) + with mock.patch.object( + _shaping_module.gpd, "points_from_xy", side_effect=AssertionError + ): + result = _shaping_module._point_feature_frame(features) + assert result is not None + pd.testing.assert_frame_equal(result[fallback.columns], fallback) + assert result.geometry.isna().all() + assert result.crs == "EPSG:4326" + + +@pytest.mark.skipif(not _shaping_module.GEOPANDAS, reason="requires geopandas") +def test_spatial_geometry_property_collision_preserves_fallback(): + features = [ + { + "properties": {"geometry": None, "value": 1}, + "geometry": {"type": "Point", "coordinates": [1, 2]}, + } + ] + assert _shaping_module._point_feature_frame(features) is None + fallback = _shaping_module._geo_feature_frame(features) + result = _shaping_module._spatial_feature_frame(features) + pd.testing.assert_frame_equal(result[fallback.columns], fallback) + assert result.geometry.iloc[0] is None + + +def test_properties_frame_flat_matches_normalize(): + """Flat properties take the plain-DataFrame path and match + ``json_normalize`` exactly.""" + properties = [ + {"a": "1", "b": None}, + {"a": "2", "b": "x"}, + ] + fast = _shaping_module._properties_frame([{"properties": p} for p in properties]) + pd.testing.assert_frame_equal(fast, pd.json_normalize(properties, sep="_")) + + +def test_properties_frame_nested_still_normalizes(): + """One nested value anywhere routes the whole page through + ``json_normalize`` so no row keeps a raw dict.""" + properties = [ + {"a": "1", "nested": None}, + {"a": "2", "nested": "scalar"}, + {"a": "3"}, + {"a": "4", "nested": {"x": "y", "deeper": {"z": 1}}}, + ] + df = _shaping_module._properties_frame([{"properties": p} for p in properties]) + pd.testing.assert_frame_equal(df, pd.json_normalize(properties, sep="_")) + assert "nested_x" in df.columns + assert not any(isinstance(v, dict) for v in df.to_numpy().ravel()) + + class TestWireIdSwitch: """The API keys every collection on ``id``; callers spell it after the collection (``monitoring_location_id``). The switch happens here, and