Add Bulk Api 2.0 and update composite requests

.github/workflows/ci.yml

@@ -50,7 +50,7 @@ jobs:
         run: dotnet test src/NetCoreForce.Client.Tests --configuration $config --no-restore --no-build --verbosity normal
 
       - name: Upload nuget artifact
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: nuget-package
           path: ${{github.workspace}}/packages/NetCoreForce.*.nupkg

src/NetCoreForce.Client/Enumerations/CompositeMethod.cs

@@ -4,6 +4,8 @@ public enum CompositeMethod
     {
         Read,
         Write,
-        Delete
+        Delete,
+        Create,
+        Update
     }
 }

src/NetCoreForce.Client/Enumerations/IngestJobResultType.cs

@@ -0,0 +1,9 @@
+namespace NetCoreForce.Client.Enumerations
+{
+    public enum IngestJobResultType
+    {
+        Successful,
+        Failed,
+        Unprocessed
+    }
+}

src/NetCoreForce.Client/Enumerations/JobType.cs

@@ -0,0 +1,8 @@
+namespace NetCoreForce.Client.Enumerations
+{
+    public enum JobType
+    {
+        Query,
+        Ingest
+    }
+}

src/NetCoreForce.Client/Extensions/UriExtensions.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Web;
+
+namespace NetCoreForce.Client.Extensions
+{
+    public static class UriExtensions
+    {
+        public static Uri AddQueryParameter(this Uri uri, string key, string value)
+        {
+            var uriBuilder = new UriBuilder(uri);
+            var query = HttpUtility.ParseQueryString(uriBuilder.Query);
+            query[key] = value;
+            uriBuilder.Query = query.ToString();
+            return uriBuilder.Uri;
+        }
+    }
+}

src/NetCoreForce.Client/ForceClient.cs

@@ -1,4 +1,5 @@
 using NetCoreForce.Client.Enumerations;
+using NetCoreForce.Client.Extensions;
 using NetCoreForce.Client.Models;
 using System;
 using System.Collections.Generic;
@@ -7,6 +8,7 @@
 using System.Linq;
 using System.Net.Http;
 using System.Runtime.CompilerServices;
+using System.Text;
 using System.Text.RegularExpressions;
 using System.Threading;
 using System.Threading.Tasks;
@@ -688,9 +690,17 @@ public async Task<CompositeRequestResponse> ExecuteCompositeRecords(
 
             List<CompositeSubRequest> subRequests = sObjects.Select(s =>
             {
+                var method = s.Method switch
+                {
+                    CompositeMethod.Write => string.IsNullOrWhiteSpace(s.Id) ? "POST" : "PATCH",
+                    CompositeMethod.Delete => "DELETE",
+                    CompositeMethod.Create => "POST",
+                    CompositeMethod.Update => "PATCH",
+                    _ => "GET",
+                };
                 return new CompositeSubRequest(
                     s.SObject,
-                    s.Method == CompositeMethod.Write ? string.IsNullOrWhiteSpace(s.Id) ? "POST" : "PATCH" : s.Method == CompositeMethod.Delete ? "DELETE" : "GET",
+                    method,
                     s.ReferenceId,
                     s.CompositeType == CompositeType.SObject ? UriFormatter.CompositeSubRequest(ApiVersion, s.Type, s.Id) : UriFormatter.CompositeSObjectCollectionsSubRequest(ApiVersion)
                 );
@@ -828,6 +838,182 @@ private async Task<HttpResponseMessage> BlobRetrieveResponse(string sObjectTypeN
             throw new ForceApiException($"Failed to download blob data, request returned {responseMessage.StatusCode} {responseMessage.ReasonPhrase}");
         }
 
+        #region bulk methods
+
+        // BULK METHODS
+
+        public async Task<QueryJobInfoResult> CreateQueryJobAsync(string queryString, bool queryAll = false)
+        {
+            if (string.IsNullOrEmpty(queryString)) throw new ArgumentNullException(nameof(queryString));
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var jobInfo = new QueryJobInfo
+            {
+                Operation = queryAll ? "queryAll" : "query",
+                Query = queryString
+            };
+
+            var uri = UriFormatter.CreateJob(InstanceUrl, ApiVersion, JobType.Query);
+
+            return await client.HttpPostAsync<QueryJobInfoResult>(jobInfo, uri);
+        }
+
+        public async Task<QueryJobInfoResult> GetQueryJobInfoResultAsync(string jobId)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Query, jobId);
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            return await client.HttpGetAsync<QueryJobInfoResult>(uri);
+        }
+
+        public async Task<QueryJobResult<T>> GetQueryJobResultAsync<T>(string jobId, Func<string, List<T>> converter, string locator = null, int? maxRecords = null)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Query, jobId, "results");
+
+            if (!string.IsNullOrEmpty(locator))
+            {
+                uri = uri.AddQueryParameter("locator", locator);
+            }
+
+            if (maxRecords.HasValue)
+            {
+                uri = uri.AddQueryParameter("maxRecords", maxRecords.ToString());
+            }
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var result = await client.HttpGetAsync<QueryJobResult>(uri);
+            QueryJobResult<T> response;
+
+            response = new QueryJobResult<T>(result)
+            {
+                Items = converter(result.Items)
+            };
+
+            return response;
+        }
+
+        public async Task<IngestJobInfoResult> CreateIngestJobAsync(IngestJobInfo jobInfo)
+        {
+            if (jobInfo == null) throw new ArgumentNullException(nameof(jobInfo));
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var uri = UriFormatter.CreateJob(InstanceUrl, ApiVersion, JobType.Ingest);
+
+            return await client.HttpPostAsync<IngestJobInfoResult>(jobInfo, uri);
+        }
+
+        public async Task<IngestJobInfoResult> IngestJobAbortAsync(string jobId)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var jobInfoUpdate = new IngestJobInfoUpdate
+            {
+                State = "Aborted"
+            };
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Ingest, jobId);
+
+            return await client.HttpPatchAsync<IngestJobInfoResult>(jobInfoUpdate, uri);
+        }
+
+        public async Task<bool> IngestJobDeleteAsync(string jobId)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Ingest, jobId);
+
+            return await client.HttpDeleteAsync<bool>(uri);
+        }
+
+        public async Task<bool> IngestJobUploadAsync(string jobId, string data)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+            if (data == null) throw new ArgumentNullException(nameof(data));
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Ingest, jobId, "batches");
+
+            await client.HttpAsync<object>(uri, HttpMethod.Put, new StringContent(data, Encoding.UTF8, "text/csv"), deserializeResponse: false);
+
+            return true;
+        }
+
+        public async Task<IngestJobInfoResult> IngestJobUploadCompleteAsync(string jobId)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var jobInfoUpdate = new IngestJobInfoUpdate
+            {
+                State = "UploadComplete"
+            };
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Ingest, jobId);
+
+            return await client.HttpPatchAsync<IngestJobInfoResult>(jobInfoUpdate, uri);
+        }
+
+        public async Task<IngestJobInfoResult> GetIngestJobInfoResultAsync(string jobId)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Ingest, jobId);
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            return await client.HttpGetAsync<IngestJobInfoResult>(uri);
+        }
+
+        public async Task<IngestJobResult<T>> GetIngestJobResultAsync<T>(string jobId, IngestJobResultType ingestJobResultType, Func<string, List<T>> converter)
+        {
+            if (string.IsNullOrEmpty(jobId)) throw new ArgumentNullException(nameof(jobId));
+
+            var path = "";
+
+            switch (ingestJobResultType)
+            {
+                case IngestJobResultType.Successful:
+                    path = "successfulResults";
+                    break;
+                case IngestJobResultType.Failed:
+                    path = "failedResults";
+                    break;
+                case IngestJobResultType.Unprocessed:
+                    path = "unprocessedrecords";
+                    break; ;
+            }
+
+            var uri = UriFormatter.Job(InstanceUrl, ApiVersion, JobType.Ingest, jobId, path);
+
+            JsonClient client = new JsonClient(AccessToken, SharedHttpClient);
+
+            var result = await client.HttpGetAsync<string>(uri);
+            IngestJobResult<T> response;
+
+            response = new IngestJobResult<T>
+            {
+                Items = converter(result)
+            };
+
+            return response;
+        }
+
+        #endregion
+
         #region metadata
 
         /// <summary>

src/NetCoreForce.Client/JsonClient.cs

@@ -1,18 +1,14 @@
+using NetCoreForce.Client.Models;
+using Newtonsoft.Json;
 using System;
-using System.IO;
-using System.IO.Compression;
-using System.Linq;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Linq;
 using System.Net;
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Text;
 using System.Threading.Tasks;
-using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
-using NetCoreForce.Client.Serializer;
-using NetCoreForce.Client.Models;
 
 namespace NetCoreForce.Client
 {
@@ -64,7 +60,7 @@ public JsonClient(string accessToken, HttpClient httpClient = null)
         }
 
         /// <summary>
-        /// 
+        ///
         /// </summary>
         /// <param name="uri"></param>
         /// <param name="customHeaders"></param>
@@ -78,7 +74,7 @@ public async Task<T> HttpGetAsync<T>(Uri uri, Dictionary<string, string> customH
         }
 
         /// <summary>
-        /// 
+        ///
         /// </summary>
         /// <param name="inputObject"></param>
         /// <param name="uri"></param>
@@ -119,7 +115,7 @@ public async Task<T> HttpPostAsync<T>(
         /// <param name="serializeComplete">Serializes ALL object properties to include in the request, even those not appropriate for some update/patch calls.</param>
         /// <param name="includeSObjectId">includes the SObject ID when serializing the request content</param>
         /// <param name="fieldsToNull">A list of properties that should be set to null, but inclusing the null values in the serialized output</param>
-        /// <param name="ignoreNulls">Use with caution. By default null values are not serialized, this will serialize all explicitly nulled or missing properties as null</param>  
+        /// <param name="ignoreNulls">Use with caution. By default null values are not serialized, this will serialize all explicitly nulled or missing properties as null</param>
         /// <typeparam name="T"></typeparam>
         /// <returns></returns>
         public async Task<T> HttpPatchAsync<T>(
@@ -152,7 +148,7 @@ public async Task<T> HttpPatchAsync<T>(
         }
 
         /// <summary>
-        /// 
+        ///
         /// </summary>
         /// <param name="uri"></param>
         /// <param name="customHeaders"></param>
@@ -169,7 +165,7 @@ public async Task<T> HttpDeleteAsync<T>(Uri uri, Dictionary<string, string> cust
             return await GetResponse<T>(request, customHeaders, deserializeResponse);
         }
 
-        private async Task<T> HttpAsync<T>(Uri uri, HttpMethod httpMethod, HttpContent content = null, Dictionary<string, string> customHeaders = null, bool deserializeResponse = true)
+        public async Task<T> HttpAsync<T>(Uri uri, HttpMethod httpMethod, HttpContent content = null, Dictionary<string, string> customHeaders = null, bool deserializeResponse = true)
         {
             HttpRequestMessage request = new HttpRequestMessage();
             request.Headers.Authorization = _authHeaderValue;
@@ -254,6 +250,16 @@ private async Task<T> GetResponse<T>(HttpRequestMessage request, Dictionary<stri
                             throw new ForceApiException("Response content was empty");
                         }
 
+                        if (typeof(T) == typeof(QueryJobResult))
+                        {
+                            return GetBulkResponse<T>(responseMessage, responseContent);
+                        }
+
+                        if (typeof(T) == typeof(string))
+                        {
+                            return (T)(object)responseContent;
+                        }
+
                         return JsonConvert.DeserializeObject<T>(responseContent);
                     }
                     if (responseMessage.StatusCode == HttpStatusCode.MultipleChoices)
@@ -287,9 +293,9 @@ private async Task<T> GetResponse<T>(HttpRequestMessage request, Dictionary<stri
                             }
                         }
                         catch
-                        {   
+                        {
                             // swallow error and continue to parse as generic error response instead
-                        }                       
+                        }
 
                         // Parse generic API error response
                         string msg = string.Format("Unable to complete request, Salesforce API returned {0}.", responseMessage.StatusCode.ToString());
@@ -333,6 +339,21 @@ private async Task<T> GetResponse<T>(HttpRequestMessage request, Dictionary<stri
             throw new ForceApiException(string.Format("Error processing response: returned {0} for {1}", responseMessage.ReasonPhrase, request.RequestUri.ToString()));
         }
 
+        private T GetBulkResponse<T>(HttpResponseMessage responseMessage, string responseContent)
+        {
+            var locator = GetHeaderValues(responseMessage.Headers, "Sforce-Locator").FirstOrDefault();
+            if (locator == "null")
+            {
+                locator = null;
+            }
+            return (T)(object)new QueryJobResult
+            {
+                NumberOfRecords = int.TryParse(GetHeaderValues(responseMessage.Headers, "Sforce-NumberOfRecords").FirstOrDefault(), out var tempNumberOfRecords) ? tempNumberOfRecords : 0,
+                Locator = locator,
+                Items = responseContent,
+            };
+        }
+
         /// <summary>
         /// Get values for a particular reponse header
         /// </summary>