Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 105 additions & 9 deletions src/main/java/io/github/jpmorganchase/fusion/Fusion.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -36,14 +41,20 @@
import java.util.Map;
import java.util.Objects;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;

/**
* Class representing the Fusion API, providing methods that correspond to available API endpoints
*/
@Slf4j
@SuppressWarnings({"LombokSetterMayBeUsed", "LombokGetterMayBeUsed"})
public class Fusion {

private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

@SuppressWarnings({"FieldCanBeLocal", "FieldMayBeFinal"})
private static int defaultPageSize = -1;

private final APIManager api;
private String defaultCatalog;
private final String defaultPath;
Expand Down Expand Up @@ -195,6 +206,91 @@ 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
* @return aggregated JSON response containing all pages of data
*/
private String callAPIWithPagination(String url) {
log.debug("Starting paginated request to URL: {}", url);

Map<String, String> headers = new HashMap<>();
headers.put("x-jpmc-paginate", "true");
if (defaultPageSize > 0) {
log.debug("Using page size: {}", defaultPageSize);
headers.put("x-jpmc-page-size", String.valueOf(defaultPageSize));
}

Gson gson = new Gson();
JsonArray aggregatedResources = new JsonArray();
String nextToken = null;
int pageCount = 0;

do {
pageCount++;
if (nextToken != null) {
headers.put("x-jpmc-next-token", nextToken);
log.debug("Fetching page {} with next token", pageCount);
} else {
log.debug("Fetching page {}", pageCount);
}

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");
int pageResourceCount = pageResources.size();
pageResources.forEach(aggregatedResources::add);
log.debug("Retrieved {} resources from page {}", pageResourceCount, pageCount);
}

nextToken = getHeaderValue(response.getHeaders(), "x-jpmc-next-token");

if (nextToken != null && !nextToken.isEmpty()) {
log.debug("Next token received, more pages available");
}

} while (nextToken != null && !nextToken.isEmpty());

log.debug(
"Pagination complete. Total pages fetched: {}, Total resources: {}",
pageCount,
aggregatedResources.size());

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
*/
@SuppressWarnings("SameParameterValue")
private String getHeaderValue(Map<String, List<String>> headers, String headerName) {
if (headers == null || headerName == null) {
return null;
}

for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
if (entry.getKey() != null && entry.getKey().equalsIgnoreCase(headerName)) {
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.
*
Expand All @@ -203,7 +299,8 @@ private Map<String, Map<String, Object>> callForMap(String url) {
* @throws OAuthException if a token could not be retrieved for authentication
*/
public Map<String, Catalog> listCatalogs() {
String json = this.api.callAPI(rootURL.concat("catalogs"));
String url = rootURL.concat("catalogs");
String json = callAPIWithPagination(url);
return responseParser.parseCatalogResponse(json);
}

Expand All @@ -223,7 +320,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.
Expand All @@ -235,7 +332,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);
return responseParser.parseDataProductResponse(json);
}

Expand Down Expand Up @@ -266,7 +363,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.
Expand All @@ -277,7 +374,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);
return filterDatasets(responseParser.parseDatasetResponse(json, catalogName), contains, idContains);
}

Expand Down Expand Up @@ -400,7 +497,7 @@ public Map<String, Map<String, Object>> datasetResources(String dataset) {
*/
public Map<String, DatasetSeries> listDatasetMembers(String catalogName, String dataset) {
String url = String.format("%1scatalogs/%2s/datasets/%3s/datasetseries", this.rootURL, catalogName, dataset);
String json = this.api.callAPI(url);
String json = callAPIWithPagination(url);
return responseParser.parseDatasetSeriesResponse(json);
}

Expand Down Expand Up @@ -460,7 +557,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);
return responseParser.parseAttributeResponse(json, catalogName, dataset);
}

Expand Down Expand Up @@ -502,11 +599,10 @@ public Map<String, Map<String, Object>> attributeResources(String catalogName, S
* @throws OAuthException if a token could not be retrieved for authentication
*/
public Map<String, Distribution> listDistributions(String catalogName, String dataset, String seriesMember) {

String url = String.format(
"%1scatalogs/%2s/datasets/%3s/datasetseries/%4s/distributions",
this.rootURL, catalogName, dataset, seriesMember);
String json = this.api.callAPI(url);
String json = callAPIWithPagination(url);
return responseParser.parseDistributionResponse(json);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand All @@ -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;

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,38 @@ 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());

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<>();
Expand Down
Loading