Skip to content

Commit

Permalink
Merge pull request #142 from IndicoDataSolutions/nate/upload_batched_…
Browse files Browse the repository at this point in the history
…workflow_submission

[DEV-11443] Use Batch Upload for workflow submission call
  • Loading branch information
nateshim-indico authored Mar 11, 2024
2 parents 8e7acc0 + d281349 commit 4a93b59
Show file tree
Hide file tree
Showing 8 changed files with 36 additions and 24 deletions.
2 changes: 1 addition & 1 deletion Examples/AddDataSetFiles/AddDataSetFiles.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net461</TargetFramework>
<TargetFramework>net6.0</TargetFramework>
</PropertyGroup>

<ItemGroup>
Expand Down
2 changes: 1 addition & 1 deletion IndicoV2.Abstractions/IndicoV2.Abstractions.csproj
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">

<PropertyGroup>
<TargetFrameworks>netstandard2.0;net461</TargetFrameworks>
<TargetFramework>net6.0</TargetFramework>
<RootNamespace>IndicoV2</RootNamespace>
</PropertyGroup>

Expand Down
2 changes: 1 addition & 1 deletion IndicoV2.Abstractions/Storage/IStorageClient.cs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ public interface IStorageClient
Task<IEnumerable<IFileMetadata>> UploadAsync(IEnumerable<string> filePaths, CancellationToken cancellationToken);

Task<(string Name, string Meta)[]> UploadAsync(IEnumerable<(string Path, Stream Content)> files,
CancellationToken cancellationToken);
CancellationToken cancellationToken, int batchSize = 20);

JArray Serialize(IEnumerable<IFileMetadata> filesMetadata);
}
Expand Down
4 changes: 2 additions & 2 deletions IndicoV2.Abstractions/Submissions/ISubmissionsClient.cs
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ public interface ISubmissionsClient
/// <param name="resultsFileVersion">Optional. Specifies version to use for the results file.</param>
/// <param name="bundle">Optional. Flag to batch all files under a single submission id.</param>
/// <returns><c><see cref="IEnumerable{T}">IEnumerable</see></c> of submissions ids.</returns>
Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<(string Name, Stream Content)> files, CancellationToken cancellationToken = default, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false);
Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<(string Name, Stream Content)> files, CancellationToken cancellationToken = default, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false, int batchSize = 20);

/// <summary>
/// Method creates <c><see cref="ISubmission"/></c>.
Expand All @@ -56,7 +56,7 @@ public interface ISubmissionsClient
/// <param name="resultsFileVersion">Optional. Specifies version to use for the results file.</param>
/// <param name="bundle">Optional. Flag to batch all files under a single submission id.</param>
/// <returns><c><see cref="IEnumerable{T}"/></c> of submissions ids.</returns>
Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<string> paths, CancellationToken cancellationToken = default, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false);
Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<string> paths, CancellationToken cancellationToken = default, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false, int batchSize = 20);

/// <summary>
/// Method lists <c><see cref="ISubmission"/></c>.
Expand Down
2 changes: 1 addition & 1 deletion IndicoV2/IndicoV2.csproj
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">

<PropertyGroup>
<TargetFrameworks>netstandard2.0;net461</TargetFrameworks>
<TargetFramework>net6.0</TargetFramework>
<BuildType>V2</BuildType>
</PropertyGroup>

Expand Down
39 changes: 25 additions & 14 deletions IndicoV2/Storage/StorageClient.cs
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public async Task<Stream> GetAsync(Uri uri, CancellationToken cancellationToken)

public async Task<IEnumerable<IFileMetadata>> UploadAsync(IEnumerable<string> filePaths, CancellationToken cancellationToken)
{
var metadata = await new UploadFile(_indicoClient) {Files = filePaths.ToList()}.Call(cancellationToken);
var metadata = await new UploadFile(_indicoClient) { Files = filePaths.ToList() }.Call(cancellationToken);

return DeserializeMetadata(metadata);
}
Expand All @@ -63,25 +63,36 @@ public JArray Serialize(IEnumerable<IFileMetadata> filesMetadata)

public async Task<(string Name, string Meta)> UploadAsync(Stream content,
string filePath,
CancellationToken cancellationToken) =>
(await UploadAsync(new[] {(filePath, content)}, cancellationToken)).SingleOrDefault();
CancellationToken cancellationToken, int batchSize = 20) =>
(await UploadAsync(new[] { (filePath, content) }, cancellationToken, batchSize)).SingleOrDefault();

public async Task<(string Name, string Meta)[]> UploadAsync(IEnumerable<(string Path, Stream Content)> files,
CancellationToken cancellationToken)
CancellationToken cancellationToken, int batchSize = 20)
{
var content = await CreateRequest(files, cancellationToken);
var response = await _indicoClient.HttpClient.PostAsync(UploadUri, content, cancellationToken);

using (var reader = new JsonTextReader(new StreamReader(await response.Content.ReadAsStreamAsync())))
if (batchSize <= 0)
{
throw new ArgumentException("Batch size must be greater than 0.");
}
var results = new (string Name, string Meta)[] { };
var batches = files.Chunk(batchSize);
foreach (var batch in batches)
{
var uploadResult = await JArray.LoadAsync(reader, cancellationToken);
var content = await CreateRequest(files, cancellationToken);
var response = await _indicoClient.HttpClient.PostAsync(UploadUri, content, cancellationToken);

return uploadResult
.Select(t => (
Name: t.Value<string>("name"),
Meta: t.ToString()))
.ToArray();
using (var reader = new JsonTextReader(new StreamReader(await response.Content.ReadAsStreamAsync())))
{
var uploadResult = await JArray.LoadAsync(reader, cancellationToken);

var result = uploadResult
.Select(t => (
Name: t.Value<string>("name"),
Meta: t.ToString()))
.ToArray();
results = results.Concat(result).ToArray();
}
}
return results;
}

private async Task<HttpContent> CreateRequest(IEnumerable<(string Path, Stream Content)> files, CancellationToken cancellationToken)
Expand Down
1 change: 1 addition & 0 deletions IndicoV2/Storage/UploadStream.cs
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ public async Task<JArray> Call(CancellationToken cancellationToken = default)

return uploadResult;
}

}
}

8 changes: 4 additions & 4 deletions IndicoV2/Submissions/SubmissionsClient.cs
Original file line number Diff line number Diff line change
Expand Up @@ -51,9 +51,9 @@ public async Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<Stre
return await _strawberryShakeClient.Submissions().Create(workflowId, (IEnumerable<(string Name, string Meta)>)files, cancellationToken, (SubmissionResultVersion?) resultsFileVersion, bundle);
}

public async Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<(string Name, Stream Content)> filesToUpload, CancellationToken cancellationToken = default, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false)
public async Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<(string Name, Stream Content)> filesToUpload, CancellationToken cancellationToken = default, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false, int batchSize = 20)
{
var filesUploaded = await _indicoClient.Storage().UploadAsync(filesToUpload, cancellationToken);
var filesUploaded = await _indicoClient.Storage().UploadAsync(filesToUpload, cancellationToken, batchSize: batchSize);
return await _strawberryShakeClient.Submissions().Create(workflowId, filesUploaded, cancellationToken, (SubmissionResultVersion?)resultsFileVersion, bundle);
}

Expand All @@ -68,7 +68,7 @@ public async Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<Uri>
await _strawberryShakeClient.Submissions().CreateUri(workflowId, uris, cancellationToken);

public async Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<string> paths,
CancellationToken cancellationToken, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false)
CancellationToken cancellationToken, SubmissionResultsFileVersion? resultsFileVersion = null, bool bundle = false, int batchSize = 20)
{
var filesToUpload = new List<(string Name, Stream content)>();
foreach (var path in paths)
Expand All @@ -84,7 +84,7 @@ public async Task<IEnumerable<int>> CreateAsync(int workflowId, IEnumerable<stri
}

}
return await CreateAsync(workflowId, filesToUpload: filesToUpload, cancellationToken, resultsFileVersion, bundle);
return await CreateAsync(workflowId, filesToUpload: filesToUpload, cancellationToken, resultsFileVersion, bundle, batchSize);
}

[Obsolete("This is the Legacy version and will be deprecated. Please use ListAsync instead.")]
Expand Down

0 comments on commit 4a93b59

Please sign in to comment.