Add PostgREST server as a fully available API

README.md

@@ -6,6 +6,15 @@ Read the API docs at `<api_endpoint>/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. 
 

app/apis/abstract.py

@@ -124,9 +124,28 @@ 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"},
                 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

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):
@@ -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,17 +68,17 @@ 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(
                 code=response.status,
                 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,18 +95,36 @@ 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 fields:
+            field_list = [field for field in fields.split(",")]
+            check_fields_valid(field_list, schema.valid_fields, table)
+            fields_needed = "objectid, " + fields
+        else:
+            field_list = None
+            fields_needed = ", ".join([field for field in schema.valid_fields])
         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(",")]
-            check_fields_valid(field_list, schema.valid_fields, table)
-            fields = "objectid, " + fields
         params = {
             "where": where,
-            "outFields": fields,
+            "outFields": fields_needed,
             "outSR": out_sr,
             "orderByFields": "objectid",
             "f": "geojson",
@@ -114,30 +134,36 @@ 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, field_list)
 
     async def normalize_rv(
-        self, request: Request, response: aiohttp.ClientResponse, schema: TableSchema
+        self,
+        request: Request,
+        response: aiohttp.ClientResponse,
+        schema: TableSchema,
+        return_json: ReturnJson,
+        field_list: list[str] | None
     ) -> 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:
                 records = data["features"]
                 self.harmonize_timestamp_fields(records, schema)
+                if field_list: 
+                    remove_extra_fields(records, field_list)
                 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,17 +173,17 @@ 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(
                 code=response.status,
                 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 

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]