From 2e3ae840eaaa7d6e996cdcfff9db3fc3c04b5087 Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Mon, 29 Jun 2026 16:41:36 +0000 Subject: [PATCH 01/10] Update README --- README.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/README.md b/README.md index 0e7ecd4..41e7213 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,15 @@ Read the API docs at `/docs` The publicly accessible endpoints are actually at the path `/api`, i.e. the standard `GET` request goes to `/api/get`, but this is hidden from the user by the reverse proxy, Mulesoft. +### APIs +This API interfaces with the following APIs: +* PostgREST + * This API connects to `databridge-public` and reads PostgreSQL functions unique to each table there + * These functions are created via an [Airflow task](https://github.com/CityOfPhiladelphia/airflow-iac-dags/edit/main/lib/databridge_tasks/upload_to_db_public_simple.py) +* ArcGIS Online +* CARTO V3 + + ### Debugging There is an internal endpoint available for testing. This will only be accessible on the City network, i.e. not by the reverse proxy. From 8273b52d4ce972b30a24a9f9d9982bab2b0184f5 Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Wed, 1 Jul 2026 17:53:32 +0000 Subject: [PATCH 02/10] Working PR - running into 504 error --- app/apis/databridge.py | 15 +++++++++------ app/utils/utils.py | 2 +- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/app/apis/databridge.py b/app/apis/databridge.py index 8aeb1df..92225c8 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -18,7 +18,7 @@ class Databridge(AbstractWorker): def __init__(self): self.name = "Databridge-Public PostgREST API" - self.base_url = "https://postgrest-public-dev.citygeo.phila.city" + self.base_url = "https://postgrest-public-dev.citygeo.phila.city/rpc" async def get_count( self, @@ -84,6 +84,9 @@ async def get( if limit: params["limit"] = limit + if schema.geom_column: + params['out_sr'] = out_sr if out_sr else self.DEFAULT_SRID + async with session.get( url, params=params, headers=headers, timeout=timeout ) as response: @@ -97,12 +100,11 @@ async def normalize_rv( ) -> ReturnJson: links = Links(self=str(request.url)) meta = Meta(service=self.name, service_url=str(response.url)) - # Now how we gonna transform this to geojson? Should we make database views for that instead? - data = await response.json() if response.ok: + data = await response.json() + geom_column = table_schema.geom_column geojsons = [] - for record in data["rows"]: - geom_column = table_schema.geom_column + for record in data: if geom_column: geojson = GeoJsonFeature( id=record.pop("objectid"), @@ -123,10 +125,11 @@ async def normalize_rv( rv = ReturnJson(data=gjfc, links=links, meta=meta) return rv else: + text = await response.text() error = Error( code=response.status, title=f"{self.name} Error", - detail=data["error"], + detail=text, ) rv = ReturnJson(errors=[error], links=links, meta=meta) return rv diff --git a/app/utils/utils.py b/app/utils/utils.py index 418d253..64e9f0f 100644 --- a/app/utils/utils.py +++ b/app/utils/utils.py @@ -265,7 +265,7 @@ def __init__(self): self.map_str_to_api: dict[str, AbstractWorker] = { "ago": Ago(), "carto": Carto(), - # "databridge": Databridge(), + "databridge": Databridge(), } # This is the initial order searched if no API is specified, and is the query param the user must submit self.map_api_to_params: dict[AbstractWorker, list[str]] = {} self.api_priority_queue: list[AbstractWorker] = [] From e6fc3d6a15774d9baa0083562c242ca641a841ed Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Wed, 1 Jul 2026 19:25:11 +0000 Subject: [PATCH 03/10] Ensure the geometry is picked up --- app/apis/databridge.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/apis/databridge.py b/app/apis/databridge.py index 92225c8..4fecbec 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -76,9 +76,9 @@ async def get( else: field_list = [field.strip() for field in fields.split(",")] check_fields_valid(field_list, schema.valid_fields, table) - if schema.geom_column: - fields = f"{schema.geom_column}, " + fields fields = "objectid, " + fields + if schema.geom_column: + fields = f"{schema.geom_column}, " + fields params = {"select": fields} headers = {"prefer": "count=exact"} From d8133a0b3dd4ea1879916e543b781c64bc44c314 Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Wed, 1 Jul 2026 20:42:28 +0000 Subject: [PATCH 04/10] Remove counting from normal query; add comments --- app/apis/databridge.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/app/apis/databridge.py b/app/apis/databridge.py index 4fecbec..a8248ad 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -30,6 +30,9 @@ async def get_count( **kwargs, ) -> ReturnJson: url = f'{self.base_url}/{table}' + # This will be slow on big tables (15s for 5M rows) + # count=estimated not supported for RPC https://github.com/PostgREST/postgrest/issues/3652; + # count=planned returns '*' which is useless headers = {"prefer": "count=exact"} async with session.head(url, headers=headers, timeout=timeout) as response: return await self.normalize_rv_count(request, response) @@ -80,7 +83,6 @@ async def get( if schema.geom_column: fields = f"{schema.geom_column}, " + fields params = {"select": fields} - headers = {"prefer": "count=exact"} if limit: params["limit"] = limit @@ -88,7 +90,7 @@ async def get( params['out_sr'] = out_sr if out_sr else self.DEFAULT_SRID async with session.get( - url, params=params, headers=headers, timeout=timeout + url, params=params, timeout=timeout ) as response: return await self.normalize_rv(request, response, schema) From 0948756b9fb01867716c7357373bbe85bf4328f0 Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Wed, 1 Jul 2026 20:45:21 +0000 Subject: [PATCH 05/10] Add order by objectid to databridge --- app/apis/databridge.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/apis/databridge.py b/app/apis/databridge.py index a8248ad..a267a78 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -82,7 +82,7 @@ async def get( fields = "objectid, " + fields if schema.geom_column: fields = f"{schema.geom_column}, " + fields - params = {"select": fields} + params = {"select": fields, "order": "objectid"} if limit: params["limit"] = limit From 940b988c99126039169db516e51ca7bf3fbd4e0e Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Wed, 1 Jul 2026 21:12:03 +0000 Subject: [PATCH 06/10] Perform count against table rather than SQL RPC function --- app/apis/databridge.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/app/apis/databridge.py b/app/apis/databridge.py index a267a78..ef3de07 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -18,7 +18,8 @@ class Databridge(AbstractWorker): def __init__(self): self.name = "Databridge-Public PostgREST API" - self.base_url = "https://postgrest-public-dev.citygeo.phila.city/rpc" + self.table_url = "https://postgrest-public-dev.citygeo.phila.city" # Used only for counts + self.rpc_url = "https://postgrest-public-dev.citygeo.phila.city/rpc" # PostgreSQL Function used because of ST_Trasform and GeoJSON preparation async def get_count( self, @@ -29,10 +30,7 @@ async def get_count( request: Request, **kwargs, ) -> ReturnJson: - url = f'{self.base_url}/{table}' - # This will be slow on big tables (15s for 5M rows) - # count=estimated not supported for RPC https://github.com/PostgREST/postgrest/issues/3652; - # count=planned returns '*' which is useless + url = f'{self.table_url}/{table}' headers = {"prefer": "count=exact"} async with session.head(url, headers=headers, timeout=timeout) as response: return await self.normalize_rv_count(request, response) @@ -71,7 +69,7 @@ async def get( schema: TableSchema, **kwargs, ) -> ReturnJson: - url = f'{self.base_url}/{table}' + url = f'{self.rpc_url}/{table}' params = {} if not fields: From 80f859b4cb6b8d88ae0e789692c6103f6e961c7c Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Wed, 22 Jul 2026 20:55:36 +0000 Subject: [PATCH 07/10] Complete code for databridge --- app/apis/abstract.py | 2 +- app/apis/ago.py | 70 +++++++++----- app/apis/carto.py | 6 +- app/apis/databridge.py | 205 +++++++++++++++++++++++++++++------------ app/public.py | 86 ++++++----------- app/test_main.py | 62 ++++++++++--- 6 files changed, 278 insertions(+), 153 deletions(-) diff --git a/app/apis/abstract.py b/app/apis/abstract.py index 1e234e8..f1825d9 100644 --- a/app/apis/abstract.py +++ b/app/apis/abstract.py @@ -124,7 +124,7 @@ def check_fields_valid(field_list: list[str], valid_fields: list[str], table: st HTTPException: If a non-existent field was requested """ for field in field_list: - if field not in valid_fields: + if field.lower().strip() not in valid_fields: raise HTTPException( status_code=400, headers={"title": "Bad Request"}, diff --git a/app/apis/ago.py b/app/apis/ago.py index 2955210..0f0eb4c 100644 --- a/app/apis/ago.py +++ b/app/apis/ago.py @@ -29,6 +29,7 @@ async def get_count( timeout: float, session: aiohttp.ClientSession, request: Request, + return_json: ReturnJson, **kwargs, ) -> ReturnJson: url = f"{self.organization_url}{table}{self.query_url}" @@ -42,21 +43,22 @@ async def get_count( if kwargs["token"]: params["token"] = kwargs["token"].removeprefix("Bearer ") async with session.get(url, params=params, timeout=timeout) as response: - return await self.normalize_rv_count(request, response) + return await self.normalize_rv_count(request, response, return_json) async def normalize_rv_count( - self, request: Request, response: aiohttp.ClientResponse + self, + request: Request, + response: aiohttp.ClientResponse, + return_json: ReturnJson, ) -> ReturnJson: - links = Links(self=str(request.url)) service_url = self.mask_service_url(request, response) - meta = Meta(service=self.name, service_url=service_url) + return_json.meta.service_url = service_url # AGO REST API doesn't respect HTTP status codes if response.ok: data = await response.json() if "error" not in data: - meta.records_total = data["properties"]["count"] - rv = ReturnJson(links=links, meta=meta) - return rv + return_json.meta.records_total = data["properties"]["count"] + return return_json else: title = f"{self.name} Error" if data["error"]["message"]: @@ -66,8 +68,8 @@ async def normalize_rv_count( title=title, detail=data["error"]["details"][0], ) - rv = ReturnJson(errors=[error], links=links, meta=meta) - return rv + return_json.errors = [error] + return return_json else: error_detail = await response.text() error = Error( @@ -75,8 +77,8 @@ async def normalize_rv_count( title=f"{self.name} Error", detail=error_detail, ) - rv = ReturnJson(errors=[error], links=links, meta=meta) - return rv + return_json.errors = [error] + return return_json # Do not remove any unused parameters as they are crucial to the documentation async def get( @@ -93,13 +95,30 @@ async def get( schema: TableSchema, **kwargs, ) -> ReturnJson: + links = Links(self=str(request.url)) + meta = Meta(service=self.name) + return_json = ReturnJson(links=links, meta=meta) + + if kwargs['sql']: + error = Error( + code=400, + title='Bad Request', + detail=f"'sql' query parameter is invalid for {self.name} service", + ) + return_json.errors = [error] + return return_json + if count_only: + return await self.get_count( + table, where, timeout, session, request, return_json, **kwargs + ) + url = f"{self.organization_url}{table}{self.query_url}" if not where: where = "1=1" if not fields: fields = ", ".join([field for field in schema.valid_fields]) else: - field_list = [field.strip() for field in fields.split(",")] + field_list = [field for field in fields.split(",")] check_fields_valid(field_list, schema.valid_fields, table) fields = "objectid, " + fields params = { @@ -114,14 +133,17 @@ async def get( if kwargs["token"]: params["token"] = kwargs["token"].removeprefix("Bearer ") async with session.get(url, params=params, timeout=timeout) as response: - return await self.normalize_rv(request, response, schema) + return await self.normalize_rv(request, response, schema, return_json) async def normalize_rv( - self, request: Request, response: aiohttp.ClientResponse, schema: TableSchema + self, + request: Request, + response: aiohttp.ClientResponse, + schema: TableSchema, + return_json: ReturnJson, ) -> ReturnJson: - links = Links(self=str(request.url)) service_url = self.mask_service_url(request, response) - meta = Meta(service=self.name, service_url=service_url) + return_json.meta.service_url = service_url # AGO REST API doesn't respect HTTP status codes if response.ok: data = await response.json() @@ -129,15 +151,15 @@ async def normalize_rv( records = data["features"] self.harmonize_timestamp_fields(records, schema) gjfc = GeoJsonFeatureCollection(**data) - meta.record_count = len(gjfc.features) + return_json.meta.record_count = len(gjfc.features) try: data["properties"]["exceededTransferLimit"] next_url = self.create_next_url(gjfc.features, request) - links.next = next_url + return_json.links.next = next_url except KeyError: pass - rv = ReturnJson(data=gjfc, links=links, meta=meta) - return rv + return_json.data = gjfc + return return_json else: title = f"{self.name} Error" if data["error"]["message"]: @@ -147,8 +169,8 @@ async def normalize_rv( title=title, detail=data["error"]["details"][0], ) - rv = ReturnJson(errors=[error], links=links, meta=meta) - return rv + return_json.errors = [error] + return return_json else: error_detail = await response.text() error = Error( @@ -156,8 +178,8 @@ async def normalize_rv( title=f"{self.name} Error", detail=error_detail, ) - rv = ReturnJson(errors=[error], links=links, meta=meta) - return rv + return_json.errors = [error] + return return_json def harmonize_timestamp_fields(self, records: list[dict], schema: TableSchema): """Coerce to a consistent representation of timestamp fields. AGO returns diff --git a/app/apis/carto.py b/app/apis/carto.py index ad88826..2ea95ee 100644 --- a/app/apis/carto.py +++ b/app/apis/carto.py @@ -42,7 +42,6 @@ async def get_count( max_age: int, session: aiohttp.ClientSession, request: Request, - **kwargs, ) -> ReturnJson: # These queries on their own are unsafe, but we are relying on the safety # checks of the back-end APIs @@ -99,6 +98,9 @@ async def get( ) -> ReturnJson: # These queries on their own are unsafe, but we are relying on the safety # checks of the back-end APIs + if count_only: + return await self.get_count(table, where, timeout, max_age, session, request) + if not sql: subq_select = psql.SQL("SELECT objectid AS geojson_id, ") if schema.geom_column: @@ -111,7 +113,7 @@ async def get( else: subq_select += psql.SQL("NULL AS geojson_shape, ") if fields: - field_list = [field.strip() for field in fields.split(",")] + field_list = [field for field in fields.split(",")] check_fields_valid(field_list, schema.valid_fields, table) fields_composed = psql.SQL(", ").join( [psql.Identifier(field) for field in field_list] diff --git a/app/apis/databridge.py b/app/apis/databridge.py index ef3de07..cb41bc1 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -1,8 +1,6 @@ import aiohttp -import os import datetime as dt from fastapi import Request -from psycopg import sql as psql # Redefine to allow "sql" as a query parameter from .abstract import AbstractWorker, check_fields_valid from ..utils.models import ( ReturnJson, @@ -20,6 +18,8 @@ def __init__(self): self.name = "Databridge-Public PostgREST API" self.table_url = "https://postgrest-public-dev.citygeo.phila.city" # Used only for counts self.rpc_url = "https://postgrest-public-dev.citygeo.phila.city/rpc" # PostgreSQL Function used because of ST_Trasform and GeoJSON preparation + self.sql_to_postgrest_url = "https://dev-sql-to-postgrest-api.citygeo.phila.city/convert" + self.max_records = 1000 async def get_count( self, @@ -28,30 +28,37 @@ async def get_count( timeout: float, session: aiohttp.ClientSession, request: Request, - **kwargs, + return_json: ReturnJson ) -> ReturnJson: - url = f'{self.table_url}/{table}' + generated_sql = self.generate_sql( + table, fields=None, where=where, limit=None, schema=None, count_only=True + ) + translator_rv = await self.get_postgrest_url( + generated_sql, session, timeout, return_json + ) + if isinstance(translator_rv, ReturnJson): + return translator_rv + elif isinstance(translator_rv, str): + postgrest_url = translator_rv + + url = f"{self.table_url}{postgrest_url}" headers = {"prefer": "count=exact"} async with session.head(url, headers=headers, timeout=timeout) as response: - return await self.normalize_rv_count(request, response) + return await self.normalize_rv_count(response, return_json) async def normalize_rv_count( - self, request: Request, response: aiohttp.ClientResponse + self, + response: aiohttp.ClientResponse, + return_json: ReturnJson, ) -> ReturnJson: - links = Links(self=str(request.url)) - meta = Meta(service=self.name, service_url=str(response.url)) + return_json.meta.service_url = str(response.url) if response.ok: - meta.records_total = response.headers.get("Content-Range").split("/")[1] - rv = ReturnJson(links=links, meta=meta) - return rv + return_json.meta.records_total = response.headers.get("Content-Range").split("/")[1] + return return_json else: - error = Error( - code=response.status, - title=f"{self.name} Error", - detail=response.reason, - ) - rv = ReturnJson(errors=[error], links=links, meta=meta) - return rv + error = Error(code=response.status) + return_json.errors = [error] + return return_json # Do not remove any unused parameters as they are crucial to the documentation async def get( @@ -69,70 +76,82 @@ async def get( schema: TableSchema, **kwargs, ) -> ReturnJson: - url = f'{self.rpc_url}/{table}' - params = {} + links = Links(self=str(request.url)) + meta = Meta(service=self.name) + return_json = ReturnJson(links=links, meta=meta) - if not fields: - fields = ", ".join([field for field in schema.valid_fields]) - else: - field_list = [field.strip() for field in fields.split(",")] - check_fields_valid(field_list, schema.valid_fields, table) - fields = "objectid, " + fields - if schema.geom_column: - fields = f"{schema.geom_column}, " + fields - params = {"select": fields, "order": "objectid"} - - if limit: - params["limit"] = limit - if schema.geom_column: - params['out_sr'] = out_sr if out_sr else self.DEFAULT_SRID + if count_only: + return await self.get_count( + table, where, timeout, session, request, return_json + ) + + params = {} + if not sql: + if limit is None: + limit = self.max_records + else: + limit = min(limit, self.max_records) + generated_sql = self.generate_sql(table, fields, where, limit, schema, count_only=False) + if schema.geom_column: + params['out_sr'] = out_sr if out_sr else self.DEFAULT_SRID + else: + generated_sql = sql + translator_rv = await self.get_postgrest_url( + generated_sql, session, timeout, return_json + ) + if isinstance(translator_rv, ReturnJson): + return translator_rv + elif isinstance(translator_rv, str): + postgrest_url = translator_rv - async with session.get( - url, params=params, timeout=timeout - ) as response: - return await self.normalize_rv(request, response, schema) + url = f'{self.rpc_url}{postgrest_url}' + async with session.get(url, params=params, timeout=timeout) as response: + return await self.normalize_rv( + request, response, schema, sql, limit, return_json + ) async def normalize_rv( self, request: Request, response: aiohttp.ClientResponse, - table_schema: TableSchema | None, + schema: TableSchema | None, + sql: str | None, + limit: int, + return_json: ReturnJson ) -> ReturnJson: - links = Links(self=str(request.url)) - meta = Meta(service=self.name, service_url=str(response.url)) + return_json.meta.service_url = str(response.url) + data = await response.json() if response.ok: - data = await response.json() - geom_column = table_schema.geom_column geojsons = [] for record in data: - if geom_column: + if schema and schema.geom_column: geojson = GeoJsonFeature( - id=record.pop("objectid"), + id=record["objectid"], properties=record, - geometry=record.pop(geom_column), + geometry=record.pop(schema.geom_column), ) else: geojson = GeoJsonFeature( - id=record.pop("objectid"), + id=record["objectid"], properties=record, ) geojsons.append(geojson) gjfc = GeoJsonFeatureCollection(features=geojsons) - meta.record_count = len(gjfc.features) - next_url = self.create_next_url(gjfc.features, request) - links.next = next_url - rv = ReturnJson(data=gjfc, links=links, meta=meta) - return rv + return_json.meta.record_count = len(gjfc.features) + if not sql and return_json.meta.record_count == limit: + next_url = self.create_next_url(gjfc.features, request) + return_json.links.next = next_url + return_json.data = gjfc + return return_json else: - text = await response.text() error = Error( code=response.status, - title=f"{self.name} Error", - detail=text, + title=f"{self.name} Error {data['code']}", + detail=f'{data['details']} {data['message']}', ) - rv = ReturnJson(errors=[error], links=links, meta=meta) - return rv + return_json.errors = [error] + return return_json def harmonize_timestamp_fields(self, records: list[dict], schema: TableSchema): """Return a consistent representation of timestamp fields. AGO returns @@ -149,3 +168,75 @@ def harmonize_timestamp_fields(self, records: list[dict], schema: TableSchema): record["properties"][field] = dt.datetime.fromtimestamp( record["properties"][field] / 1000 ) + + async def get_postgrest_url( + self, + generated_sql: str | None, + session: aiohttp.ClientSession, + timeout: float, + return_json: ReturnJson + ) -> str | ReturnJson: + """Call the SQL-to-PostgREST translator API for the given SQL. If any + error is encountered, create a JSON API response and return that instead + + Args: + generated_sql (str | None): SQL + request (Request): The request to this API + session (aiohttp.ClientSession): Session to make requests with + timeout (float): Timeout in seconds + + Returns: + str | ReturnJson: If str, the URL for accessing PostgREST. If a ReturnJson, + then an error was encountered + """ + async with session.get( + self.sql_to_postgrest_url, params={'sql': generated_sql}, timeout=timeout + ) as response: + data = await response.json() + if response.ok: + return data['path'] + else: + error = Error( + code=response.status, + title=f"{self.name} Error", + detail=data["error"], + ) + return_json.errors = [error] + return return_json + + def generate_sql( + self, + table: str, + fields: str | None, + where: str | None, + limit: int | None, + schema: TableSchema, + count_only: bool, + ) -> str: + """Generate the correct SQL syntax to send to the SQL-to-PostgREST translator + from user submitted query parameters. + Ensure that the SQL contains the objectid and geometry columns. This + function only accommodates SELECT, FROM, WHERE, LIMIT, and ORDER BY operators. + """ + stmt = "SELECT " + if count_only: + stmt += "* " + else: + if not fields: + fields = ", ".join([field for field in schema.valid_fields]) + else: + field_list = [field for field in fields.split(",")] + check_fields_valid(field_list, schema.valid_fields, table) + fields = "objectid, " + fields + if schema.geom_column: + fields = f"{schema.geom_column}, " + fields + stmt += f"{fields} " + stmt += f"FROM {table} " + if where: + stmt += f"WHERE {where} " + if not count_only: + stmt += "ORDER BY objectid " + stmt += f"LIMIT {limit} " + + return stmt + \ No newline at end of file diff --git a/app/public.py b/app/public.py index 7087602..3303f14 100644 --- a/app/public.py +++ b/app/public.py @@ -123,9 +123,14 @@ async def get_data( token = request.headers["authorization"] else: token = None - if table: + if not table and not sql: + raise HTTPException( + status_code=400, + detail="'table' or 'sql' parameters are required", + headers={"title": "Bad Request"}, + ) + if table and not sql: table = table.lower() - if not sql: schema = schema_cache.retrieve_table_schema(table) else: schema = None @@ -144,71 +149,38 @@ async def get_data( "schema": schema, "token": token, } - if sql: - if service and service.lower() != "carto": - raise HTTPException( - status_code=400, - detail="SQL parameter can only be used with `service=carto`", - headers={"title": "Bad Request"}, - ) - try: - rv = await api_manager.map_str_to_api["carto"].get(**params) - except TimeoutError: - raise HTTPException( - status_code=408, - detail="Request could not be completed. Request less data, preferably 2,000 rows or fewer, or alternatively try again.", - headers={"title": "Request Timeout"}, - ) - return generate_final_response(rv) if not service: links = Links(self=str(request.url)) rv_combined = ReturnJson(links=links, errors=[]) for api in api_manager.map_api_to_params: - if table: - try: - rv = await api.get(**params) - except TimeoutError: - api_manager.deprioritize(api) - error = Error( - code=408, - title=f"{api.name} Timeout Error", - detail="Request could not be completed. Request less data, preferably 2,000 rows or fewer, or alternatively try again.", + try: + rv = await api.get(**params) + except TimeoutError: + api_manager.deprioritize(api) + error = Error( + code=408, + title=f"{api.name} Timeout Error", + detail="Request could not be completed. Request less data, preferably 2,000 rows or fewer, or alternatively try again.", + ) + rv_combined.errors.append(error) + else: + if not rv.errors: + rv.meta.service_available_query_parameters = ( + api_manager.map_api_to_params[api] ) - rv_combined.errors.append(error) + return rv else: - if not rv.errors: - rv.meta.service_available_query_parameters = ( - api_manager.map_api_to_params[api] - ) - return rv - else: - rv_combined.errors.append(rv.errors[0]) - else: - raise HTTPException( - status_code=400, - detail="'table' or 'sql' parameters are required", - headers={"title": "Bad Request"}, - ) + rv_combined.errors.append(rv.errors[0]) return generate_final_response(rv_combined) else: api = api_manager.map_str_to_api[service.lower()] - if table: - try: - if count_only: - rv = await api.get_count(**params) - else: - rv = await api.get(**params) - except TimeoutError: - raise HTTPException( - status_code=408, - detail="Request could not be completed. Request less data, preferably 2,000 rows or fewer, or alternatively try again", - headers={"title": "Request Timeout"}, - ) - else: + try: + rv = await api.get(**params) + except TimeoutError: raise HTTPException( - status_code=400, - detail="'table' or 'sql' parameters are required", - headers={"title": "Bad Request"}, + status_code=408, + detail="Request could not be completed. Request less data, preferably 2,000 rows or fewer, or alternatively try again", + headers={"title": "Request Timeout"}, ) rv.meta.service_available_query_parameters = api_manager.map_api_to_params[api] return generate_final_response(rv) diff --git a/app/test_main.py b/app/test_main.py index cfae0da..1080658 100644 --- a/app/test_main.py +++ b/app/test_main.py @@ -84,7 +84,7 @@ def test_valid_private(client: TestClient, token: str, service: str, count_only: assert rv["meta"]["record_count"] > 0, f"Service {service} found zero features in table {table}" -@pytest.mark.parametrize("service", ["carto"]) +@pytest.mark.parametrize("service", ["carto", "databridge"]) def test_valid_private_no_interfere(client: TestClient, service: str, token: str): """Test that a private token doesn't interfere with other APIs""" table = GOOD_TABLES[0] @@ -258,9 +258,10 @@ def test_valid_srid(client: TestClient, service: str): assert rv2["data"]["features"], f'Service {service} found zero features in table {table}' -@pytest.mark.parametrize("service", ["carto"]) +@pytest.mark.parametrize("service", ["ago", "carto", "databridge"]) def test_valid_sql(client: TestClient, service: str): - """Test that the `sql` parameter works, only on Carto""" + """Test that the `sql` parameter works for Carto & Databridge and that it + fails correctly for AGO""" table = GOOD_TABLES[0] params = { "table": "ANSTHES", # Should have no effect @@ -271,18 +272,26 @@ def test_valid_sql(client: TestClient, service: str): "max_age": 0, } response = client.get(f"{public_prefix}/get", params=params) - assert response.status_code == 200 - rv = response.json() - assert rv["meta"]["record_count"] == 5 - assert rv["data"]["features"], f'Service {service} found zero features in table {table}' + if service == 'ago': + assert response.status_code >= 400 and response.status_code < 500 + return None + else: + assert response.status_code == 200 + rv = response.json() + assert rv["meta"]["record_count"] == 5 + assert rv["data"]["features"], f'Service {service} found zero features in table {table}' + assert "next" not in rv["links"].keys() @pytest.mark.parametrize("service", ["carto"]) -def test_valid_sql_too_large(client: TestClient, service: str): +def test_valid_sql_too_large_for_carto(client: TestClient, service: str): """Test that the `sql` parameter works will error if the response from Carto - is too large to handle but smaller than a timeout""" + is too large to handle but smaller than a timeout. This test + should only run on Carto which has no inherent limits to response data size. + PostgREST server has a configured limit of 1000 records""" params = { "sql": f"SELECT * FROM {GOOD_TABLES[0]} LIMIT 50000", + "service": service, "max_age": 0, } response = client.get(f"{public_prefix}/get", params=params) @@ -456,11 +465,40 @@ def test_invalid_sql(client: TestClient, service: str): assert "errors" in data -@pytest.mark.parametrize("service", ["None"]) -def test_invalid_sql_large_payload(client: TestClient, service: None): - """Test that the API fails if too large of a dataset is requsted""" +@pytest.mark.parametrize("service", api_manager.map_str_to_api.keys()) +def test_invalid_sql_ddl_insert(client: TestClient, service: str): + """Test that the APIs fail if DDL (INSERT/UPDATE) statements are passed""" + params = { + "sql": "INSERT INTO PPD_COMPLAINTS (objectid) VALUES (0)", + "service": service, + } + response = client.get(f"{public_prefix}/get", params=params) + assert response.status_code >= 400 and response.status_code <= 500 + data = response.json() + assert "errors" in data + + +@pytest.mark.parametrize("service", api_manager.map_str_to_api.keys()) +def test_invalid_sql_ddl_update(client: TestClient, service: str): + """Test that the APIs fail if DDL (INSERT/UPDATE) statements are passed""" + params = { + "sql": "UPDATE PPD_COMPLAINTS SET objectid = 0", + "service": service, + } + response = client.get(f"{public_prefix}/get", params=params) + assert response.status_code >= 400 and response.status_code <= 500 + data = response.json() + assert "errors" in data + + +@pytest.mark.parametrize("service", ["Carto"]) +def test_invalid_sql_too_large_for_carto(client: TestClient, service: None): + """Test that the API fails if too large of a dataset is requsted. This test + should only run on Carto which has no inherent limits to response data size. + PostgREST server has a configured limit of 1000 records""" params = { "sql": f"SELECT * FROM {GOOD_TABLES[0]}", + "service": service, "max_age": 0, } response = client.get(f"{public_prefix}/get", params=params) From 721c6148ebfd5754146afaee32530f45861810e9 Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Thu, 23 Jul 2026 16:27:57 +0000 Subject: [PATCH 08/10] Add two tests; ensure APIs return exact same fields; cleanup --- app/apis/abstract.py | 19 ++++++++++++++ app/apis/ago.py | 22 +++++++++------- app/apis/databridge.py | 58 ++++++++++++++++++++---------------------- app/test_main.py | 39 +++++++++++++++++++++++----- 4 files changed, 91 insertions(+), 47 deletions(-) diff --git a/app/apis/abstract.py b/app/apis/abstract.py index f1825d9..fd95b76 100644 --- a/app/apis/abstract.py +++ b/app/apis/abstract.py @@ -130,3 +130,22 @@ def check_fields_valid(field_list: list[str], valid_fields: list[str], table: st headers={"title": "Bad Request"}, detail=f"Invalid field requested from table '{table}': '{field}'", ) + + +def remove_extra_fields(records: list[dict], field_list: list[dict]): + """Remove any fields not requested by the user that remain in the downstream + API response data. This is particularly the case for the objectid field which + must be requested from AGO even if the user does not specify it so that each + record has its identifier. Modifies the list of records in place. + + Args: + records (list[dict]): Data returned from downstream APIs + field_list (list[dict]): List of fields requested by user + """ + for record in records: + old_properties = record['properties'] + new_properties = {} + for field, value in old_properties.items(): + if field in field_list: + new_properties[field] = value + record['properties'] = new_properties diff --git a/app/apis/ago.py b/app/apis/ago.py index 0f0eb4c..504559b 100644 --- a/app/apis/ago.py +++ b/app/apis/ago.py @@ -11,7 +11,7 @@ ReturnJson, TableSchema, ) -from .abstract import AbstractWorker, check_fields_valid +from .abstract import AbstractWorker, check_fields_valid, remove_extra_fields class Ago(AbstractWorker): @@ -113,17 +113,18 @@ async def get( ) url = f"{self.organization_url}{table}{self.query_url}" - if not where: - where = "1=1" - if not fields: - fields = ", ".join([field for field in schema.valid_fields]) - else: + if fields: field_list = [field for field in fields.split(",")] check_fields_valid(field_list, schema.valid_fields, table) - fields = "objectid, " + fields + fields_needed = "objectid, " + fields + else: + field_list = None + fields_needed = ", ".join([field for field in schema.valid_fields]) + if not where: + where = "1=1" params = { "where": where, - "outFields": fields, + "outFields": fields_needed, "outSR": out_sr, "orderByFields": "objectid", "f": "geojson", @@ -133,7 +134,7 @@ async def get( if kwargs["token"]: params["token"] = kwargs["token"].removeprefix("Bearer ") async with session.get(url, params=params, timeout=timeout) as response: - return await self.normalize_rv(request, response, schema, return_json) + return await self.normalize_rv(request, response, schema, return_json, field_list) async def normalize_rv( self, @@ -141,6 +142,7 @@ async def normalize_rv( response: aiohttp.ClientResponse, schema: TableSchema, return_json: ReturnJson, + field_list: list[str] | None ) -> ReturnJson: service_url = self.mask_service_url(request, response) return_json.meta.service_url = service_url @@ -150,6 +152,8 @@ async def normalize_rv( if "error" not in data: records = data["features"] self.harmonize_timestamp_fields(records, schema) + if field_list: + remove_extra_fields(records, field_list) gjfc = GeoJsonFeatureCollection(**data) return_json.meta.record_count = len(gjfc.features) try: diff --git a/app/apis/databridge.py b/app/apis/databridge.py index cb41bc1..a53863c 100644 --- a/app/apis/databridge.py +++ b/app/apis/databridge.py @@ -27,7 +27,6 @@ async def get_count( where: str | None, timeout: float, session: aiohttp.ClientSession, - request: Request, return_json: ReturnJson ) -> ReturnJson: generated_sql = self.generate_sql( @@ -81,9 +80,7 @@ async def get( return_json = ReturnJson(links=links, meta=meta) if count_only: - return await self.get_count( - table, where, timeout, session, request, return_json - ) + return await self.get_count(table, where, timeout, session, return_json) params = {} if not sql: @@ -103,11 +100,15 @@ async def get( return translator_rv elif isinstance(translator_rv, str): postgrest_url = translator_rv + if fields: + field_list = [field for field in fields.split(",")] + else: + field_list = None url = f'{self.rpc_url}{postgrest_url}' async with session.get(url, params=params, timeout=timeout) as response: return await self.normalize_rv( - request, response, schema, sql, limit, return_json + request, response, schema, sql, limit, return_json, field_list ) async def normalize_rv( @@ -117,24 +118,31 @@ async def normalize_rv( schema: TableSchema | None, sql: str | None, limit: int, - return_json: ReturnJson + return_json: ReturnJson, + field_list: list[str] | None, ) -> ReturnJson: return_json.meta.service_url = str(response.url) data = await response.json() if response.ok: geojsons = [] - for record in data: + for record in data: + objectid = record["objectid"] if schema and schema.geom_column: - geojson = GeoJsonFeature( - id=record["objectid"], - properties=record, - geometry=record.pop(schema.geom_column), - ) - else: - geojson = GeoJsonFeature( - id=record["objectid"], - properties=record, - ) + geometry = record.pop(schema.geom_column) + else: + geometry = None + if field_list: # Remove any fields not requested by the user, especially objectid. + new_record = {} + for field, value in record.items(): + if field in field_list: + new_record[field] = value + else: + new_record = record + geojson = GeoJsonFeature( + id=objectid, + properties=new_record, + geometry=geometry, + ) geojsons.append(geojson) gjfc = GeoJsonFeatureCollection(features=geojsons) @@ -154,20 +162,8 @@ async def normalize_rv( return return_json def harmonize_timestamp_fields(self, records: list[dict], schema: TableSchema): - """Return a consistent representation of timestamp fields. AGO returns - timestamp fields as milliseconds since the epoch - - Args: - records (list[dict]): Data records - schema (TableSchema): TableSchema - """ - for record in records: - for field in record["properties"]: - if field in schema.timestamp_fields: - if record["properties"][field]: - record["properties"][field] = dt.datetime.fromtimestamp( - record["properties"][field] / 1000 - ) + """PostgREST returns ISO-8601 automatically""" + pass async def get_postgrest_url( self, diff --git a/app/test_main.py b/app/test_main.py index 1080658..bd93608 100644 --- a/app/test_main.py +++ b/app/test_main.py @@ -1,4 +1,5 @@ from collections.abc import Generator +import datetime as dt import pytest from fastapi.testclient import TestClient @@ -117,17 +118,12 @@ def test_valid_fields(client: TestClient, service: str): assert rv["data"]["features"], f'Service {service} found zero features in table {table}' -@pytest.mark.skip("""Skipping this test because if user does not request the "objectid" -field and this API doesn't include it, then AGO will not provide feature IDs. -I'm making the design decision to include an extra field in the user response -rather than not providing the "id" column. Either way, AGO (and thus this API) -violates the JSON:API spec.""") @pytest.mark.parametrize("service", api_manager.map_str_to_api.keys()) def test_valid_fields2(client: TestClient, service: str): params = { "table": GOOD_TABLES[0], "limit": 2, - "fields": "addr_std", + "fields": "document_id,document_type,display_date", "service": service, "max_age": 0, } @@ -158,7 +154,7 @@ def test_valid_where(client: TestClient, service: str): @pytest.mark.parametrize("service", api_manager.map_str_to_api.keys()) -def test_valid_where_parethesization(client: TestClient, service: str): +def test_valid_where_parenthesization(client: TestClient, service: str): """Test that the `where` clause given by the next url and joined with an SQL AND doesn't decouple any existing WHERE clause, i.e. because SQL `AND` binds more tightly than `OR`""" @@ -322,6 +318,35 @@ def test_valid_timeout(client: TestClient, service: str): assert response.status_code == 408 +def test_valid_harmonized_timestamps(client: TestClient): + """Test that the data returned from the downstream APIs has the same ISO-8601 + formatted timestamp values""" + timestamp_dict = { + "display_date": None, + "receipt_date": None, + "recording_date": None, + "document_date": None, + } + for service in api_manager.map_str_to_api.keys(): + params = { + "table": GOOD_TABLES[0], + "fields": ",".join(timestamp_dict.keys()), + "service": service, + "max_age": 0, + "limit": 1 + } + response = client.get(f"{public_prefix}/get", params=params) + assert response.status_code == 200 + data = response.json() + for field, value in data['data']['features'][0]['properties'].items(): + if field in timestamp_dict.keys() and value: + assert dt.datetime.fromisoformat(value) + if timestamp_dict[field]: + assert timestamp_dict[field] == value + else: + timestamp_dict[field] = value + + @pytest.mark.skip("""Skipping this test because these tables have differences both in timestamp fields and in geometry fields that are unrelated to this API. """) From 9c26e3417e2ba7cd88ad2ee0a9deb5dfed52c77e Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Thu, 23 Jul 2026 16:42:36 +0000 Subject: [PATCH 09/10] Add schemas tests --- app/test_main.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/app/test_main.py b/app/test_main.py index bd93608..16f7770 100644 --- a/app/test_main.py +++ b/app/test_main.py @@ -45,6 +45,23 @@ def token() -> str: # Valid Parameter Tests # ################################################################################ +@pytest.mark.parametrize("table", GOOD_TABLES) +def test_individual_schemas(client: TestClient, table: str): + """Test that the `/schemas` pathway works as intended and the necessary tables are found""" + response = client.get("/schemas", params={"table": table}) + assert response.status_code == 200 + data = response.json() + assert table in data['schema'] + + +def test_all_schemas(client: TestClient): + """Test that the `/schemas` pathway works as intended""" + response = client.get("/schemas") + assert response.status_code == 200 + data = response.json() + for table in GOOD_TABLES: + assert table in data['schemas'] + @pytest.mark.parametrize("table", GOOD_TABLES) @pytest.mark.parametrize("service", api_manager.map_str_to_api.keys()) From 107617fb468d05a679f38d8dba90d1e35376ab06 Mon Sep 17 00:00:00 2001 From: James Midkiff Date: Thu, 23 Jul 2026 16:44:32 +0000 Subject: [PATCH 10/10] Update API docs & change API priority order --- app/utils/utils.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/app/utils/utils.py b/app/utils/utils.py index 64e9f0f..07a487f 100644 --- a/app/utils/utils.py +++ b/app/utils/utils.py @@ -178,7 +178,7 @@ def retrieve_table_schema(self, table: str) -> dict: description = """ -This wrapper API retrieves data from ArcGIS Online (AGO) and Carto SQL API V3 (Carto) following +This wrapper API retrieves data from Databridge-Public database, ArcGIS Online (AGO), and Carto SQL API V3 (Carto) following the below specifications: 1. [JSON:API](https://jsonapi.org/) for API response, 1. [GeoJSON](https://datatracker.ietf.org/doc/html/rfc7946) for returned data, and @@ -186,11 +186,14 @@ def retrieve_table_schema(self, table: str) -> dict: Note there may be small differences in data values for the same table between the -AGO and Carto APIs specifically in geometry and timestamp fields due to those APIs +APIs specifically in geometry and timestamp fields due to those APIs internal configurations **Source code: https://github.com/CityOfPhiladelphia/databridge_api** +### Databridge-Public +The Databridge-Public contains public tables only; it is configured with a PostgREST server. + ### Carto SQL API V3 Carto solely contains public tables, but they can only be accessed via a private API token. This token will not be accessible @@ -263,10 +266,10 @@ class Api_Manager: def __init__(self): """Note this method must be updated if any new APIs are added""" self.map_str_to_api: dict[str, AbstractWorker] = { + "databridge": Databridge(), "ago": Ago(), "carto": Carto(), - "databridge": Databridge(), - } # This is the initial order searched if no API is specified, and is the query param the user must submit + } # This is the initial priority order searched if no API is specified, and is the query param the user must submit self.map_api_to_params: dict[AbstractWorker, list[str]] = {} self.api_priority_queue: list[AbstractWorker] = [] self.populate_initial_values()