From 579a888e9183af4cafaab7ad1e71337862bf438e Mon Sep 17 00:00:00 2001 From: nor <9804927+kuromukira@users.noreply.github.com> Date: Thu, 20 Oct 2022 13:29:43 +0800 Subject: [PATCH 1/2] Implement bulk create and bulk update (#15) * Implementation for bulk create * Implement bulk update --- VERSION | 2 +- code/DarkMatter/Program.cs | 15 +++-- code/Universe/Galaxy.cs | 85 ++++++++++++++++++++++------- code/Universe/Interfaces/IGalaxy.cs | 9 ++- code/Universe/UniverseQuery.csproj | 6 +- code/Universe/UniverseQuery.xml | 11 +++- 6 files changed, 93 insertions(+), 35 deletions(-) diff --git a/VERSION b/VERSION index 0c87a0e..bf7112c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.3.2 \ No newline at end of file +1.4.0 \ No newline at end of file diff --git a/code/DarkMatter/Program.cs b/code/DarkMatter/Program.cs index 989f627..5fd4f5e 100644 --- a/code/DarkMatter/Program.cs +++ b/code/DarkMatter/Program.cs @@ -1,4 +1,5 @@ -using Microsoft.Azure.Cosmos; +using System.Text.Json.Serialization; +using Microsoft.Azure.Cosmos; using Universe; using Universe.Interfaces; using Universe.Options; @@ -14,12 +15,14 @@ CosmosDbPrimaryKey, clientOptions: new() { - Serializer = new UniverseSerializer() + Serializer = new UniverseSerializer(), + AllowBulkExecution = true // This will tell the underlying code to allow async bulk operations } ); IGalaxy galaxy = new MyRepo( - db: cosmosClient.GetDatabase(""), + client: cosmosClient, + database: "", container: "", partitionKey: "/" ); @@ -63,6 +66,8 @@ class MyObject : ICosmicEntity public string id { get; set; } public DateTime AddedOn { get; set; } public DateTime? ModifiedOn { get; set; } + + [JsonIgnore] public string PartitionKey => Code; public string Code { get; set; } @@ -77,11 +82,11 @@ class MyObject : ICosmicEntity class MyRepo : Galaxy { #if DEBUG - public MyRepo(Database db, string container, string partitionKey) : base(db, container, partitionKey, true) + public MyRepo(CosmosClient client, string database, string container, string partitionKey) : base(client, database, container, partitionKey, true) { } #else - public MyRepo(Database db, string container, string partitionKey) : base(db, container, partitionKey) + public MyRepo(CosmosClient client, string database, string container, string partitionKey) : base(client, database, container, partitionKey) { } #endif diff --git a/code/Universe/Galaxy.cs b/code/Universe/Galaxy.cs index 85ce993..025eb54 100644 --- a/code/Universe/Galaxy.cs +++ b/code/Universe/Galaxy.cs @@ -1,6 +1,4 @@ -using System.Linq; -using System.Net; -using Universe.Options; +using System.Net; using Universe.Response; namespace Universe; @@ -11,16 +9,19 @@ public abstract class Galaxy : IDisposable, IGalaxy where T : ICosmicEntit private readonly Container Container; private bool DisposedValue; - private bool RecordQuery; + private readonly bool RecordQuery; + private readonly bool AllowBulk; /// - protected Galaxy(Database db, string container, string partitionKey, bool recordQueries = false) + protected Galaxy(CosmosClient client, string database, string container, string partitionKey, bool recordQueries = false) { if (string.IsNullOrWhiteSpace(container) || string.IsNullOrWhiteSpace(partitionKey)) throw new UniverseException("Container name and PartitionKey are required"); RecordQuery = recordQueries; - Container = db.CreateContainerIfNotExistsAsync(container, partitionKey).GetAwaiter().GetResult(); + if (client.ClientOptions is not null) + AllowBulk = client.ClientOptions.AllowBulkExecution; + Container = client.GetDatabase(database).CreateContainerIfNotExistsAsync(container, partitionKey).GetAwaiter().GetResult(); } private static QueryDefinition CreateQuery(IList catalysts, ColumnOptions? columnOptions = null, IList sorting = null, IList groups = null) @@ -108,33 +109,42 @@ private static QueryDefinition CreateQuery(IList catalysts, ColumnOpti return (new(response.RequestCharge, null), model.id); } - async Task IGalaxy.Create(IList catalysts, T model) + async Task IGalaxy.Create(IList models) { - QueryDefinition query = CreateQuery(catalysts: catalysts); - - using FeedIterator queryResponse = Container.GetItemQueryIterator(query); - if (queryResponse.HasMoreResults) + try { - FeedResponse next = await queryResponse.ReadNextAsync(); - if (next.Count > 0) - throw new UniverseException($"{typeof(T).Name} already exists."); - } + if (!AllowBulk) + throw new UniverseException("Bulk create of documents is not configured properly."); - model.id = Guid.NewGuid().ToString(); - model.AddedOn = DateTime.UtcNow; + Gravity gravity = new(0, string.Empty); + List tasks = new(models.Count); - ItemResponse response = await Container.CreateItemAsync(model, new PartitionKey(model.PartitionKey)); - return new(response.RequestCharge, null, RecordQuery ? (query.QueryText, query.GetQueryParameters()) : default); + foreach (T model in models) + { + if (string.IsNullOrWhiteSpace(model.id)) + model.id = Guid.NewGuid().ToString(); + model.AddedOn = DateTime.UtcNow; + + tasks.Add(Container.CreateItemAsync(model, new PartitionKey(model.PartitionKey)) + .ContinueWith(response => gravity = new(gravity.RU + response.Result.RequestCharge, string.Empty))); + } + + await Task.WhenAll(tasks); + return gravity; + } + catch + { + throw; + } } async Task<(Gravity, T)> IGalaxy.Modify(T model) { try { - ItemResponse response = await Container.ReadItemAsync(model.id, new PartitionKey(model.PartitionKey)); model.ModifiedOn = DateTime.UtcNow; - response = await Container.ReplaceItemAsync(model, response.Resource.id, new PartitionKey(response.Resource.PartitionKey)); + ItemResponse response = await Container.ReplaceItemAsync(model, model.id, new PartitionKey(model.PartitionKey)); return (new(response.RequestCharge, null), response.Resource); } catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound) @@ -147,6 +157,39 @@ async Task IGalaxy.Create(IList catalysts, T model) } } + async Task IGalaxy.Modify(IList models) + { + try + { + if (!AllowBulk) + throw new UniverseException("Bulk modify of documents is not configured properly."); + + Gravity gravity = new(0, string.Empty); + List tasks = new(models.Count); + + foreach (T model in models) + { + model.ModifiedOn = DateTime.UtcNow; + + tasks.Add(Container.ReplaceItemAsync(model, model.id, new PartitionKey(model.PartitionKey)) + .ContinueWith(response => + { + if (!response.IsCompletedSuccessfully) + throw new UniverseException(response.Exception.Flatten().InnerException.Message); + + gravity = new(gravity.RU + response.Result.RequestCharge, string.Empty); + })); + } + + await Task.WhenAll(tasks); + return gravity; + } + catch + { + throw; + } + } + async Task IGalaxy.Remove(string id, string partitionKey) { try diff --git a/code/Universe/Interfaces/IGalaxy.cs b/code/Universe/Interfaces/IGalaxy.cs index 2575793..c4f7ba1 100644 --- a/code/Universe/Interfaces/IGalaxy.cs +++ b/code/Universe/Interfaces/IGalaxy.cs @@ -11,15 +11,20 @@ public interface IGalaxy where T : ICosmicEntity Task<(Gravity g, string t)> Create(T model); /// - /// Create a new model in the database + /// Bulk create new models in the database /// - Task Create(IList catalysts, T model); + Task Create(IList models); /// /// Modify a model in the database /// Task<(Gravity g, T T)> Modify(T model); + /// + /// Bulk modify models in the database + /// + Task Modify(IList models); + /// /// Remove one model from the database /// diff --git a/code/Universe/UniverseQuery.csproj b/code/Universe/UniverseQuery.csproj index c863eb2..0122ecf 100644 --- a/code/Universe/UniverseQuery.csproj +++ b/code/Universe/UniverseQuery.csproj @@ -21,10 +21,10 @@ Git Nor Gelera Universe - 1.3.2 + 1.4.0 View release on https://github.com/kuromukira/universe/releases - 1.3.2.0 - 1.3.2.0 + 1.4.0.0 + 1.4.0.0 diff --git a/code/Universe/UniverseQuery.xml b/code/Universe/UniverseQuery.xml index e0f953f..fe4201d 100644 --- a/code/Universe/UniverseQuery.xml +++ b/code/Universe/UniverseQuery.xml @@ -22,7 +22,7 @@ Inherit repositories to implement Universe - + @@ -57,9 +57,9 @@ Create a new model in the database - + - Create a new model in the database + Bulk create new models in the database @@ -67,6 +67,11 @@ Modify a model in the database + + + Bulk modify models in the database + + Remove one model from the database From 7665ba3841166ec60a8277142e68e8e63832ee15 Mon Sep 17 00:00:00 2001 From: Nor Gelera Date: Thu, 20 Oct 2022 13:34:14 +0800 Subject: [PATCH 2/2] Update README --- README.md | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 5fba59a..6e23103 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ public class MyCosmosEntity : ICosmicEntity public DateTime AddedOn { get; set; } public DateTime ModifiedOn { get; set; } + [JsonIgnore] public string PartitionKey => FirstName; } ``` @@ -23,7 +24,15 @@ public class MyCosmosEntity : ICosmicEntity ```csharp public class MyRepository : Galaxy { - public MyRepository(Database db, string container, string partitionKey) : base(db, container, partitionKey) + public MyRepository(CosmosClient client, string database, string container, string partitionKey) : base(client, database, container, partitionKey) + { + } +} + +// If you want to see debug information such as the full Query text executed, use the format below: +public class MyRepository : Galaxy +{ + public MyRepository(CosmosClient client, string database, string container, string partitionKey) : base(client, database, container, partitionKey, true) { } } @@ -37,6 +46,7 @@ _ = services.AddScoped(_ => new CosmosClient( clientOptions: new() { Serializer = new UniverseSerializer() // This is from Universe.Options + AllowBulkExecution = true // This will tell the underlying code to allow async bulk operations } )); ``` @@ -44,7 +54,8 @@ _ = services.AddScoped(_ => new CosmosClient( 4. In your Startup.cs / Main method / Program.cs, configure your CosmosDb repository like so: ```csharp _ = services.AddScoped, MyRepository>(service => new( - db: service.GetRequiredService().GetDatabase("cosmos-database"), + client: service.GetRequiredService(), + database: "database-name", container: "container-name", partitionKey: "/partitionKey" ));