From 2ed12b602a080d98e2025fc0e4edefdc341412ba Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 6 Jul 2026 14:23:14 -0600 Subject: [PATCH 1/5] fix: stop per-request OOM on /geospatial export endpoint The /geospatial endpoint (geojson and shapefile formats) loaded the entire thing/location result set into memory on a single request, spiking memory enough for App Engine to terminate the process mid-request. - get_thing_features: stream the query result via yield_per instead of buffering the whole table with .all(). unique() still dedups eager-loaded rows. Callers iterate exactly once. - geojson format: stream the FeatureCollection feature-by-feature through a StreamingResponse instead of building a full list of feature dicts. - shapefile format: write to a tempfile.mkdtemp() dir instead of the read-only App Engine app directory (/tmp is RAM-backed), and clean up the temp dir via BackgroundTask after the response is sent. Co-Authored-By: Claude Opus 4.8 --- api/geospatial.py | 49 ++++++++++++++++++++++++----------- services/geospatial_helper.py | 10 ++++--- 2 files changed, 41 insertions(+), 18 deletions(-) diff --git a/api/geospatial.py b/api/geospatial.py index 082979f8a..5bf1348f6 100644 --- a/api/geospatial.py +++ b/api/geospatial.py @@ -14,13 +14,17 @@ # limitations under the License. # =============================================================================== import json +import os +import shutil +import tempfile from typing import Annotated, List from fastapi import APIRouter, Query, HTTPException from fastapi.responses import FileResponse from geoalchemy2.shape import to_shape from shapely.io import to_geojson -from starlette.responses import JSONResponse +from starlette.background import BackgroundTask +from starlette.responses import StreamingResponse from core.dependencies import session_dependency, viewer_dependency from db import Group @@ -54,8 +58,7 @@ def get_geospatial( """ if format_ == "geojson": - content = get_feature_collection(session, thing_type, group) - return JSONResponse(content=content, media_type="application/geo+json") + return get_feature_collection(session, thing_type, group) else: return get_location_shapefile(session, thing_type, group) @@ -94,9 +97,12 @@ def get_feature_collection( group: Annotated[ str | int, Query(title="group", description="group", alias="group") ] = None, -) -> FeatureCollectionResponse: +) -> StreamingResponse: """ - Endpoint to retrieve a GeoJSON FeatureCollection. + Retrieve a GeoJSON FeatureCollection. + + Streamed feature-by-feature so the entire result set is never buffered in + memory at once. """ things = get_thing_features(session, thing_type, group) @@ -115,12 +121,15 @@ def make_feature_dict(thing, geometry, elevation, *other): "geometry": geometry, } - features = [make_feature_dict(*item) for item in things] + def generate(): + yield '{"type": "FeatureCollection", "features": [' + first = True + for item in things: + yield ("" if first else ",") + json.dumps(make_feature_dict(*item)) + first = False + yield "]}" - return { - "type": "FeatureCollection", - "features": features, - } + return StreamingResponse(generate(), media_type="application/geo+json") def get_location_shapefile( @@ -133,16 +142,26 @@ def get_location_shapefile( """ things = get_thing_features(session, thing_type, group) - create_shapefile(things, "things.shp") - # Return the shapefile as a zip (optional: zip the .shp, .shx, .dbf files) + # Write into a temp dir: the App Engine app directory is read-only, and /tmp + # is RAM-backed, so build here and clean up after the response is sent. + tmpdir = tempfile.mkdtemp() + shp_path = os.path.join(tmpdir, "things.shp") + zip_path = os.path.join(tmpdir, "things.zip") + + create_shapefile(things, shp_path) + import zipfile - with zipfile.ZipFile("things.zip", "w") as zf: + with zipfile.ZipFile(zip_path, "w") as zf: for ext in ["shp", "shx", "dbf"]: - zf.write(f"things.{ext}") + zf.write(os.path.join(tmpdir, f"things.{ext}"), arcname=f"things.{ext}") + return FileResponse( - "things.zip", media_type="application/zip", filename="things.zip" + zip_path, + media_type="application/zip", + filename="things.zip", + background=BackgroundTask(shutil.rmtree, tmpdir, ignore_errors=True), ) diff --git a/services/geospatial_helper.py b/services/geospatial_helper.py index fc1118aa8..7a32db07f 100644 --- a/services/geospatial_helper.py +++ b/services/geospatial_helper.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== +from typing import Iterable + import shapefile from geoalchemy2.functions import ST_GeomFromText, ST_Within, ST_AsGeoJSON from geoalchemy2.shape import to_shape @@ -31,7 +33,7 @@ def get_thing_features( session, thing_type: list | str | None, group: str | int | None -) -> list: +) -> Iterable: # sql = ( # select(Thing, ST_AsGeoJSON(Location.point).label("geojson")) # .join(LocationThingAssociation, Thing.id == LocationThingAssociation.thing_id) @@ -87,8 +89,10 @@ def get_thing_features( else: sql = sql.where(Group.id == group) - # unique needs to be invoked to prevent duplicates from eager loading - return session.execute(sql).unique().all() + # unique() dedups rows from eager loading; yield_per streams the result in + # chunks so the whole table is never buffered in memory at once. Callers + # iterate the result exactly once. + return session.execute(sql.execution_options(yield_per=1000)).unique() def create_shapefile(things: list, filename: str = "things.shp") -> None: From 51ad0096ac7d8b9612175d115a3f96b26275758b Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 6 Jul 2026 14:26:36 -0600 Subject: [PATCH 2/5] chore: sync uv.lock with pyproject version bump pyproject version was bumped to 1.1.0 without re-locking, so CI's `uv sync --locked` failed. Regenerate the lockfile; no dependency changes. Co-Authored-By: Claude Opus 4.8 --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index e1ac33e08..90b0049b3 100644 --- a/uv.lock +++ b/uv.lock @@ -1489,7 +1489,7 @@ wheels = [ [[package]] name = "ocotilloapi" -version = "1.0.0" +version = "1.1.0" source = { editable = "." } dependencies = [ { name = "aiofiles" }, From fa727c42898e5d8791f248ff4dc0b5f43c7b3da2 Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 6 Jul 2026 14:32:20 -0600 Subject: [PATCH 3/5] fix: drop yield_per in get_thing_features (incompatible with unique) SQLAlchemy raises "Can't use the ORM yield_per feature in conjunction with unique()", and unique() is required to dedup eager-loaded rows. Revert to unique().all(); the memory win still comes from streaming the JSON response and writing the shapefile to a temp dir instead of building a second full copy / writing to the read-only app dir. Co-Authored-By: Claude Opus 4.8 --- services/geospatial_helper.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/services/geospatial_helper.py b/services/geospatial_helper.py index 7a32db07f..e5238c98c 100644 --- a/services/geospatial_helper.py +++ b/services/geospatial_helper.py @@ -13,8 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== -from typing import Iterable - import shapefile from geoalchemy2.functions import ST_GeomFromText, ST_Within, ST_AsGeoJSON from geoalchemy2.shape import to_shape @@ -33,7 +31,7 @@ def get_thing_features( session, thing_type: list | str | None, group: str | int | None -) -> Iterable: +) -> list: # sql = ( # select(Thing, ST_AsGeoJSON(Location.point).label("geojson")) # .join(LocationThingAssociation, Thing.id == LocationThingAssociation.thing_id) @@ -89,10 +87,10 @@ def get_thing_features( else: sql = sql.where(Group.id == group) - # unique() dedups rows from eager loading; yield_per streams the result in - # chunks so the whole table is never buffered in memory at once. Callers - # iterate the result exactly once. - return session.execute(sql.execution_options(yield_per=1000)).unique() + # unique needs to be invoked to prevent duplicates from eager loading + # (yield_per is not compatible with unique(), so the rows are materialized + # here; the callers avoid a second full copy by streaming their output). + return session.execute(sql).unique().all() def create_shapefile(things: list, filename: str = "things.shp") -> None: From f63d9ce2d28ba190b63628f40c61edee09ac1a5e Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 6 Jul 2026 14:39:42 -0600 Subject: [PATCH 4/5] fix: correct shapefile DBF schema and clean up temp dir on failure - create_shapefile defined 2 DBF fields (id, name) but wrote 3 values (id, name, elevation), which raises a field/record mismatch at runtime. This was latent because writes to the read-only app dir failed first; now that the shapefile is written to a temp dir it is exercised. Add the elevation field and fix the id field type (was "L"/logical -> "N"). - Wrap shapefile generation in try/except so the temp dir is removed if generation fails (BackgroundTask only runs on a successful response). Co-Authored-By: Claude Opus 4.8 --- api/geospatial.py | 26 ++++++++++++++++---------- services/geospatial_helper.py | 5 ++++- 2 files changed, 20 insertions(+), 11 deletions(-) diff --git a/api/geospatial.py b/api/geospatial.py index 5bf1348f6..fff814f50 100644 --- a/api/geospatial.py +++ b/api/geospatial.py @@ -146,16 +146,22 @@ def get_location_shapefile( # Write into a temp dir: the App Engine app directory is read-only, and /tmp # is RAM-backed, so build here and clean up after the response is sent. tmpdir = tempfile.mkdtemp() - shp_path = os.path.join(tmpdir, "things.shp") - zip_path = os.path.join(tmpdir, "things.zip") - - create_shapefile(things, shp_path) - - import zipfile - - with zipfile.ZipFile(zip_path, "w") as zf: - for ext in ["shp", "shx", "dbf"]: - zf.write(os.path.join(tmpdir, f"things.{ext}"), arcname=f"things.{ext}") + try: + shp_path = os.path.join(tmpdir, "things.shp") + zip_path = os.path.join(tmpdir, "things.zip") + + create_shapefile(things, shp_path) + + import zipfile + + with zipfile.ZipFile(zip_path, "w") as zf: + for ext in ["shp", "shx", "dbf"]: + zf.write(os.path.join(tmpdir, f"things.{ext}"), arcname=f"things.{ext}") + except Exception: + # BackgroundTask only runs on a successful response, so clean up here to + # avoid leaking temp dirs when generation fails. + shutil.rmtree(tmpdir, ignore_errors=True) + raise return FileResponse( zip_path, diff --git a/services/geospatial_helper.py b/services/geospatial_helper.py index e5238c98c..a6fb099d5 100644 --- a/services/geospatial_helper.py +++ b/services/geospatial_helper.py @@ -96,8 +96,11 @@ def get_thing_features( def create_shapefile(things: list, filename: str = "things.shp") -> None: # Create a point shapefile with shapefile.Writer(filename, shapeType=shapefile.POINT) as shp: - shp.field("id", "L") + # Field schema must match the values written in shp.record() below: + # id (numeric), name (char), elevation (numeric). + shp.field("id", "N") shp.field("name", "C") + shp.field("elevation", "N", decimal=3) for thing, point, elevation in things: # Assume loc.point is WKT or a Shapely geometry or GeoJSON From 877fff833755034e9c0f1b67ed3d6c74a92d4084 Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 6 Jul 2026 14:46:25 -0600 Subject: [PATCH 5/5] fix: stream get_thing_features with yield_per instead of buffering .all() still materialized the entire result set, so both exports loaded everything into memory. Thing has no eager-loaded collections (all relationships are lazy), so unique() was unnecessary -- and unique() is incompatible with yield_per anyway. Make get_thing_features a generator that streams via yield_per and dedups defensively by id with a bounded int set. Because the geojson StreamingResponse body is produced after the request session is closed, the generator now opens a dedicated session via session_ctx() scoped to the stream. The shapefile path consumes the generator inside the endpoint (request session still open). Co-Authored-By: Claude Opus 4.8 --- api/geospatial.py | 21 +++++++++++---------- services/geospatial_helper.py | 21 ++++++++++++++++----- 2 files changed, 27 insertions(+), 15 deletions(-) diff --git a/api/geospatial.py b/api/geospatial.py index fff814f50..2d649d334 100644 --- a/api/geospatial.py +++ b/api/geospatial.py @@ -28,6 +28,7 @@ from core.dependencies import session_dependency, viewer_dependency from db import Group +from db.engine import session_ctx from schemas.thing import FeatureCollectionResponse from services.geospatial_helper import create_shapefile, get_thing_features from services.query_helper import simple_get_by_id @@ -58,7 +59,7 @@ def get_geospatial( """ if format_ == "geojson": - return get_feature_collection(session, thing_type, group) + return get_feature_collection(thing_type, group) else: return get_location_shapefile(session, thing_type, group) @@ -92,7 +93,6 @@ def get_project_area( def get_feature_collection( - session: session_dependency, thing_type: List[str] | None = None, group: Annotated[ str | int, Query(title="group", description="group", alias="group") @@ -105,8 +105,6 @@ def get_feature_collection( memory at once. """ - things = get_thing_features(session, thing_type, group) - def make_feature_dict(thing, geometry, elevation, *other): geometry = json.loads(geometry) geometry["coordinates"].append(elevation) @@ -122,12 +120,15 @@ def make_feature_dict(thing, geometry, elevation, *other): } def generate(): - yield '{"type": "FeatureCollection", "features": [' - first = True - for item in things: - yield ("" if first else ",") + json.dumps(make_feature_dict(*item)) - first = False - yield "]}" + # The request-scoped session is closed before this response body + # streams, so open a dedicated session scoped to the stream. + with session_ctx() as stream_session: + yield '{"type": "FeatureCollection", "features": [' + first = True + for item in get_thing_features(stream_session, thing_type, group): + yield ("" if first else ",") + json.dumps(make_feature_dict(*item)) + first = False + yield "]}" return StreamingResponse(generate(), media_type="application/geo+json") diff --git a/services/geospatial_helper.py b/services/geospatial_helper.py index a6fb099d5..fb8777178 100644 --- a/services/geospatial_helper.py +++ b/services/geospatial_helper.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== +from typing import Iterator + import shapefile from geoalchemy2.functions import ST_GeomFromText, ST_Within, ST_AsGeoJSON from geoalchemy2.shape import to_shape @@ -31,7 +33,7 @@ def get_thing_features( session, thing_type: list | str | None, group: str | int | None -) -> list: +) -> Iterator: # sql = ( # select(Thing, ST_AsGeoJSON(Location.point).label("geojson")) # .join(LocationThingAssociation, Thing.id == LocationThingAssociation.thing_id) @@ -87,10 +89,19 @@ def get_thing_features( else: sql = sql.where(Group.id == group) - # unique needs to be invoked to prevent duplicates from eager loading - # (yield_per is not compatible with unique(), so the rows are materialized - # here; the callers avoid a second full copy by streaming their output). - return session.execute(sql).unique().all() + # Stream the result with yield_per so the whole table is never buffered in + # memory at once. Thing has no eager-loaded collections (all relationships + # are lazy), so unique() is unnecessary -- and unique() is incompatible with + # yield_per anyway. Dedup defensively by id with a bounded set of ints in + # case the joins ever produce duplicate rows. + seen = set() + result = session.execute(sql.execution_options(yield_per=1000)) + for row in result: + thing_id = row[0].id + if thing_id in seen: + continue + seen.add(thing_id) + yield row def create_shapefile(things: list, filename: str = "things.shp") -> None: