Implement pagination for API calls

Author: kylstevenson

src/main/java/io/github/jpmorganchase/fusion/Fusion.java

@@ -2,6 +2,10 @@
 
 import static io.github.jpmorganchase.fusion.filter.DatasetFilter.filterDatasets;
 
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
 import io.github.jpmorganchase.fusion.api.APIManager;
 import io.github.jpmorganchase.fusion.api.FusionAPIManager;
 import io.github.jpmorganchase.fusion.api.exception.APICallException;
@@ -11,6 +15,7 @@
 import io.github.jpmorganchase.fusion.builders.APIConfiguredBuilders;
 import io.github.jpmorganchase.fusion.builders.Builders;
 import io.github.jpmorganchase.fusion.http.Client;
+import io.github.jpmorganchase.fusion.http.HttpResponse;
 import io.github.jpmorganchase.fusion.http.JdkClient;
 import io.github.jpmorganchase.fusion.model.*;
 import io.github.jpmorganchase.fusion.oauth.credential.BearerTokenCredentials;
@@ -195,6 +200,72 @@ private Map<String, Map<String, Object>> callForMap(String url) {
         return responseParser.parseResourcesUntyped(json);
     }
 
+    /**
+     * Makes paginated API calls and aggregates all results transparently.
+     * This method handles the pagination logic internally, making multiple API calls
+     * as needed and combining the results into a single response string.
+     *
+     * @param url the API endpoint URL
+     * @param pageSize the number of items per page (null to use API default)
+     * @return aggregated JSON response containing all pages of data
+     */
+    private String callAPIWithPagination(String url, Integer pageSize) {
+        Map<String, String> headers = new HashMap<>();
+        headers.put("x-jpmc-paginate", "true");
+        if (pageSize != null) {
+            headers.put("x-jpmc-page-size", String.valueOf(pageSize));
+        }
+
+        Gson gson = new Gson();
+        JsonArray aggregatedResources = new JsonArray();
+        String nextToken = null;
+
+        do {
+            if (nextToken != null) {
+                headers.put("x-next-token", nextToken);
+            }
+
+            HttpResponse<String> response = this.api.callAPIWithResponse(url, headers);
+            String pageJson = response.getBody();
+
+            JsonObject pageObject = JsonParser.parseString(pageJson).getAsJsonObject();
+            if (pageObject.has("resources") && pageObject.get("resources").isJsonArray()) {
+                JsonArray pageResources = pageObject.getAsJsonArray("resources");
+                pageResources.forEach(aggregatedResources::add);
+            }
+
+            nextToken = getHeaderValue(response.getHeaders(), "x-next-token");
+
+        } while (nextToken != null && !nextToken.isEmpty());
+
+        JsonObject result = new JsonObject();
+        result.add("resources", aggregatedResources);
+        return gson.toJson(result);
+    }
+
+    /**
+     * Gets a header value from the response headers map (case-insensitive).
+     *
+     * @param headers the response headers map
+     * @param headerName the header name to look for
+     * @return the header value, or null if not found
+     */
+    private String getHeaderValue(Map<String, java.util.List<String>> headers, String headerName) {
+        if (headers == null || headerName == null) {
+            return null;
+        }
+
+        for (Map.Entry<String, java.util.List<String>> entry : headers.entrySet()) {
+            if (entry.getKey() != null && entry.getKey().equalsIgnoreCase(headerName)) {
+                java.util.List<String> values = entry.getValue();
+                if (values != null && !values.isEmpty()) {
+                    return values.get(0);
+                }
+            }
+        }
+        return null;
+    }
+
     /**
      * Get a list of the catalogs available to the API account.
      *
@@ -223,7 +294,7 @@ public Map<String, Map<String, Object>> catalogResources(String catalogName) {
     /**
      * Get a filtered list of the data products in the specified catalog
      * <p>
-     * Note that as of current version this search capability is not yet implemented
+     * Note that as of the current version, this search capability is not yet implemented
      *
      * @param catalogName identifier of the catalog to be queried
      * @param contains    a search keyword.
@@ -235,7 +306,7 @@ public Map<String, Map<String, Object>> catalogResources(String catalogName) {
     public Map<String, DataProduct> listProducts(String catalogName, String contains, boolean idContains) {
         // TODO: unimplemented logic implied by the method parameters
         String url = String.format("%1scatalogs/%2s/products", this.rootURL, catalogName);
-        String json = this.api.callAPI(url);
+        String json = callAPIWithPagination(url, null);
         return responseParser.parseDataProductResponse(json);
     }
 
@@ -266,7 +337,7 @@ public Map<String, DataProduct> listProducts() {
     /**
      * Get a filtered list of the datasets in the specified catalog
      * <p>
-     * Note that as of current version this search capability is not yet implemented
+     * Note that as of the current version, this search capability is not yet implemented
      *
      * @param catalogName identifier of the catalog to be queried
      * @param contains    a search keyword.
@@ -277,7 +348,7 @@ public Map<String, DataProduct> listProducts() {
      */
     public Map<String, Dataset> listDatasets(String catalogName, String contains, boolean idContains) {
         String url = String.format("%1scatalogs/%2s/datasets", this.rootURL, catalogName);
-        String json = this.api.callAPI(url);
+        String json = callAPIWithPagination(url, null);
         return filterDatasets(responseParser.parseDatasetResponse(json, catalogName), contains, idContains);
     }
 
@@ -421,7 +492,7 @@ public Map<String, Map<String, Object>> datasetMemberResources(String dataset, S
      */
     public Map<String, Attribute> listAttributes(String catalogName, String dataset) {
         String url = String.format("%1scatalogs/%2s/datasets/%3s/attributes", this.rootURL, catalogName, dataset);
-        String json = this.api.callAPI(url);
+        String json = callAPIWithPagination(url, null);
         return responseParser.parseAttributeResponse(json, catalogName, dataset);
     }
 

src/main/java/io/github/jpmorganchase/fusion/api/APIManager.java

@@ -3,10 +3,12 @@
 import io.github.jpmorganchase.fusion.api.exception.APICallException;
 import io.github.jpmorganchase.fusion.api.operations.APIDownloadOperations;
 import io.github.jpmorganchase.fusion.api.operations.APIUploadOperations;
+import io.github.jpmorganchase.fusion.http.HttpResponse;
 import java.io.UnsupportedEncodingException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLEncoder;
+import java.util.Map;
 
 public interface APIManager extends APIDownloadOperations, APIUploadOperations {
 
@@ -19,6 +21,16 @@ public interface APIManager extends APIDownloadOperations, APIUploadOperations {
      */
     String callAPI(String apiPath) throws APICallException;
 
+    /**
+     * Sends a GET request to the specified API endpoint with custom headers and returns the full HTTP response.
+     *
+     * @param apiPath the API endpoint path to which the GET request will be sent
+     * @param headers additional HTTP headers to include in the request
+     * @return the full {@code HttpResponse} including headers and body
+     * @throws APICallException if the response status indicates an error or the request fails
+     */
+    HttpResponse<String> callAPIWithResponse(String apiPath, Map<String, String> headers) throws APICallException;
+
     String callAPIToPost(String apiPath) throws APICallException;
 
     /**

src/main/java/io/github/jpmorganchase/fusion/api/FusionAPIManager.java

@@ -63,6 +63,39 @@ public String callAPI(String apiPath) throws APICallException {
         return response.getBody();
     }
 
+    /**
+     * Sends a GET request to the specified API endpoint with custom headers and returns the full HTTP response.
+     *
+     * <p>This method constructs the necessary authorization headers using a bearer token from
+     * the {@code tokenProvider}, merges them with the provided custom headers, and sends a GET
+     * request to the specified {@code apiPath} using the {@code httpClient}. It checks the HTTP
+     * response status for errors and returns the full response including headers.
+     *
+     * @param apiPath the API endpoint path to which the GET request will be sent
+     * @param customHeaders additional HTTP headers to include in the request
+     * @return the full {@code HttpResponse} including status, headers, and body
+     * @throws APICallException if the response status indicates an error or the request fails
+     */
+    @Override
+    public HttpResponse<String> callAPIWithResponse(String apiPath, Map<String, String> customHeaders)
+            throws APICallException {
+        Map<String, String> requestHeaders = new HashMap<>();
+        requestHeaders.put("Authorization", "Bearer " + tokenProvider.getSessionBearerToken());
+
+        // Merge custom headers, allowing them to override defaults except Authorization
+        if (customHeaders != null) {
+            customHeaders.forEach((key, value) -> {
+                if (!"Authorization".equalsIgnoreCase(key)) {
+                    requestHeaders.put(key, value);
+                }
+            });
+        }
+
+        HttpResponse<String> response = httpClient.get(APIManager.encodeUrl(apiPath), requestHeaders);
+        checkResponseStatus(response);
+        return response;
+    }
+
     @Override
     public String callAPIToPost(String apiPath) throws APICallException {
         Map<String, String> requestHeaders = new HashMap<>();