Skip to content

Cosmos DB: Migration to V3 SDK #717

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ealsur merged 1 commit into from
Jul 1, 2021
Merged

Cosmos DB: Migration to V3 SDK #717

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion build/common.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
<!-- Extensions can have independent versions and only increment when released -->
<Version>3.0.0$(VersionSuffix)</Version>
<ExtensionsVersion>4.0.3$(VersionSuffix)</ExtensionsVersion> <!-- WebJobs.Extensions -->
<CosmosDBVersion>3.0.10$(VersionSuffix)</CosmosDBVersion>
<CosmosDBVersion>4.0.0$(VersionSuffix)</CosmosDBVersion>
<HttpVersion>3.1.1$(VersionSuffix)</HttpVersion>
<MobileAppsVersion>3.0.0$(VersionSuffix)</MobileAppsVersion>
<SendGridVersion>3.0.2$(VersionSuffix)</SendGridVersion>
Expand Down
21 changes: 12 additions & 9 deletions src/ExtensionsSample/Samples/CosmosDBSamples.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using ExtensionsSample.Models;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Cosmos;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Host;
using Newtonsoft.Json.Linq;
Expand Down Expand Up @@ -73,19 +73,22 @@ public static void QueryDocument(
}

// DocumentClient input binding
//   The binding supplies a DocumentClient directly.
//   The binding supplies a CosmosClient directly.
[Disable]
public static void DocumentClient(
public static async Task CosmosClient(
[TimerTrigger("00:01", RunOnStartup = true)] TimerInfo timer,
[CosmosDB] DocumentClient client,
[CosmosDB] CosmosClient client,
TraceWriter log)
{
var collectionUri = UriFactory.CreateDocumentCollectionUri("ItemDb", "ItemCollection");
var documents = client.CreateDocumentQuery(collectionUri);
var iterator = client.GetContainer("ItemDb", "ItemCollection").GetItemQueryIterator<dynamic>("SELECT * FROM c");

foreach (Document d in documents)
while (iterator.HasMoreResults)
{
log.Info(d.Id);
var documents = await iterator.ReadNextAsync();
foreach (dynamic d in documents)
{
log.Info(d.id);
}
}
}
}
Expand Down
8 changes: 7 additions & 1 deletion src/ExtensionsSample/Samples/CosmosDBTriggerSamples.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@

using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Azure.Documents;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Host;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

namespace ExtensionsSample
Expand Down Expand Up @@ -55,4 +55,10 @@ public static async Task ListenAndCopy(
}
}
}

public class Document
{
[JsonProperty("id")]
public string Id { get; set; }
}
}
29 changes: 10 additions & 19 deletions src/WebJobs.Extensions.CosmosDB/Bindings/CosmosDBAsyncCollector.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,7 @@
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Cosmos;
using Newtonsoft.Json.Linq;

namespace Microsoft.Azure.WebJobs.Extensions.CosmosDB
Expand All @@ -22,19 +21,20 @@ public CosmosDBAsyncCollector(CosmosDBContext docDBContext)

public async Task AddAsync(T item, CancellationToken cancellationToken = default(CancellationToken))
{
bool create = false;
try
{
await UpsertDocument(_docDBContext, item);
}
catch (Exception ex)
{
if (CosmosDBUtility.TryGetDocumentClientException(ex, out DocumentClientException de) &&
if (CosmosDBUtility.TryGetCosmosException(ex, out CosmosException de) &&
de.StatusCode == HttpStatusCode.NotFound)
{
if (_docDBContext.ResolvedAttribute.CreateIfNotExists)
{
create = true;
await CosmosDBUtility.CreateDatabaseAndCollectionIfNotExistAsync(_docDBContext);

await UpsertDocument(_docDBContext, item);
}
else
{
Expand All @@ -48,13 +48,6 @@ public CosmosDBAsyncCollector(CosmosDBContext docDBContext)
throw;
}
}

if (create)
{
await CosmosDBUtility.CreateDatabaseAndCollectionIfNotExistAsync(_docDBContext);

await UpsertDocument(_docDBContext, item);
}
}

public Task FlushAsync(CancellationToken cancellationToken = default(CancellationToken))
Expand All @@ -63,18 +56,16 @@ public CosmosDBAsyncCollector(CosmosDBContext docDBContext)
return Task.FromResult(0);
}

internal static async Task UpsertDocument(CosmosDBContext context, T item)
internal static Task UpsertDocument(CosmosDBContext context, T item)
{
Uri collectionUri = UriFactory.CreateDocumentCollectionUri(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName);

// DocumentClient does not accept strings directly.
object convertedItem = item;
// Support user sending a string
if (item is string)
{
convertedItem = JObject.Parse(item.ToString());
JObject asJObject = JObject.Parse(item.ToString());
Copy link
Copy Markdown
Member

@brettsam brettsam Jun 2, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we use System.Text.Json here? Would that allow you to remove a ref to Newtonsoft?

Copy link
Copy Markdown
Member

@brettsam brettsam Jun 2, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nevermind... I forgot we require Newtonsoft.

Copy link
Copy Markdown
Member Author

@ealsur ealsur Jun 2, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The SDK could take System.Text.Json using a custom serializer if that is the approach we want to take.

return context.Service.GetContainer(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName).UpsertItemAsync(asJObject);
}

await context.Service.UpsertDocumentAsync(collectionUri, convertedItem);
return context.Service.GetContainer(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName).UpsertItemAsync(item);
}
}
}
12 changes: 6 additions & 6 deletions src/WebJobs.Extensions.CosmosDB/Bindings/CosmosDBClientBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,11 @@
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Cosmos;

namespace Microsoft.Azure.WebJobs.Extensions.CosmosDB.Bindings
{
internal class CosmosDBClientBuilder : IConverter<CosmosDBAttribute, DocumentClient>
internal class CosmosDBClientBuilder : IConverter<CosmosDBAttribute, CosmosClient>
{
private readonly CosmosDBExtensionConfigProvider _configProvider;

Expand All @@ -15,17 +15,17 @@ public CosmosDBClientBuilder(CosmosDBExtensionConfigProvider configProvider)
_configProvider = configProvider;
}

public DocumentClient Convert(CosmosDBAttribute attribute)
public CosmosClient Convert(CosmosDBAttribute attribute)
{
if (attribute == null)
{
throw new ArgumentNullException(nameof(attribute));
}

string resolvedConnectionString = _configProvider.ResolveConnectionString(attribute.ConnectionStringSetting);
ICosmosDBService service = _configProvider.GetService(resolvedConnectionString, attribute.PreferredLocations, attribute.UseMultipleWriteLocations, attribute.UseDefaultJsonSerialization);

return service.GetClient();
return _configProvider.GetService(
connectionString: resolvedConnectionString,
preferredLocations: attribute.PreferredLocations);
}
}
}
43 changes: 26 additions & 17 deletions src/WebJobs.Extensions.CosmosDB/Bindings/CosmosDBEnumerableBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,10 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Cosmos;

namespace Microsoft.Azure.WebJobs.Extensions.CosmosDB
{
Expand All @@ -24,28 +22,39 @@ public async Task<IEnumerable<T>> ConvertAsync(CosmosDBAttribute attribute, Canc
{
CosmosDBContext context = _configProvider.CreateContext(attribute);

Uri collectionUri = UriFactory.CreateDocumentCollectionUri(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName);

List<T> finalResults = new List<T>();

string continuation = null;
Container container = context.Service.GetContainer(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName);

SqlQuerySpec sqlSpec = new SqlQuerySpec
QueryDefinition queryDefinition = null;
if (!string.IsNullOrEmpty(attribute.SqlQuery))
{
QueryText = context.ResolvedAttribute.SqlQuery,
Parameters = context.ResolvedAttribute.SqlQueryParameters ?? new SqlParameterCollection()
};
queryDefinition = new QueryDefinition(attribute.SqlQuery);
if (attribute.SqlQueryParameters != null)
{
foreach (var parameter in attribute.SqlQueryParameters)
{
queryDefinition.WithParameter(parameter.Item1, parameter.Item2);
}
}
}

do
QueryRequestOptions queryRequestOptions = new QueryRequestOptions();
if (!string.IsNullOrEmpty(attribute.PartitionKey))
{
DocumentQueryResponse<T> response = await context.Service.ExecuteNextAsync<T>(collectionUri, sqlSpec, continuation);

finalResults.AddRange(response.Results);
continuation = response.ResponseContinuation;
queryRequestOptions.PartitionKey = new PartitionKey(attribute.PartitionKey);
}
while (!string.IsNullOrEmpty(continuation));

return finalResults;
using (FeedIterator<T> iterator = container.GetItemQueryIterator<T>(queryDefinition: queryDefinition, requestOptions: queryRequestOptions))
{
while (iterator.HasMoreResults)
{
FeedResponse<T> response = await iterator.ReadNextAsync(cancellationToken);
finalResults.AddRange(response.Resource);
}

return finalResults;
}
}
}
}
62 changes: 24 additions & 38 deletions src/WebJobs.Extensions.CosmosDB/Bindings/CosmosDBItemValueBinder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,7 @@
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Cosmos;
using Microsoft.Azure.WebJobs.Host.Bindings;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
Expand Down Expand Up @@ -41,48 +40,35 @@ public async Task SetValueAsync(object value, CancellationToken cancellationToke

public async Task<object> GetValueAsync()
{
Uri documentUri = UriFactory.CreateDocumentUri(_context.ResolvedAttribute.DatabaseName, _context.ResolvedAttribute.CollectionName, _context.ResolvedAttribute.Id);
RequestOptions options = null;
T document = default(T);

if (!string.IsNullOrEmpty(_context.ResolvedAttribute.PartitionKey))
{
options = new RequestOptions
{
PartitionKey = new PartitionKey(_context.ResolvedAttribute.PartitionKey)
};
}

Document document = null;

try
{
document = await _context.Service.ReadDocumentAsync(documentUri, options);
}
catch (DocumentClientException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
{
// ignore not found; we'll return null below
}

if (document == null)
{
return document;
}

T item = null;
PartitionKey partitionKey = _context.ResolvedAttribute.PartitionKey == null ? PartitionKey.None : new PartitionKey(_context.ResolvedAttribute.PartitionKey);

// Strings need to be handled differently.
if (typeof(T) == typeof(string))
if (typeof(T) != typeof(string))
{
_originalItem = JObject.FromObject(document);
item = _originalItem.ToString(Formatting.None) as T;
try
{
document = await _context.Service.GetContainer(_context.ResolvedAttribute.DatabaseName, _context.ResolvedAttribute.CollectionName)
.ReadItemAsync<T>(_context.ResolvedAttribute.Id, partitionKey);

_originalItem = JObject.FromObject(document);
Copy link
Copy Markdown
Member

@brettsam brettsam Jun 2, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm wondering if we want to continue supporting this "update-in-place" behavior. It's never worked for out-of-proc languages and has resulted in a decent amount of confusion as it seems all of the bindings handle it differently.

@fabiocav -- thoughts?

}
catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
{
// ignore not found; we'll return null below
}
}
else
{
item = (T)(dynamic)document;
_originalItem = JObject.FromObject(item);
JObject jObject = await _context.Service.GetContainer(_context.ResolvedAttribute.DatabaseName, _context.ResolvedAttribute.CollectionName)
.ReadItemAsync<JObject>(_context.ResolvedAttribute.Id, partitionKey);
_originalItem = jObject;

document = _originalItem.ToString(Formatting.None) as T;
}

return item;
return document;
}

public string ToInvokeString()
Expand Down Expand Up @@ -111,7 +97,7 @@ internal static async Task SetValueInternalAsync(JObject originalItem, T newItem
// make sure it's not the Id that has changed
if (!string.Equals(originalId, currentId, StringComparison.Ordinal))
{
throw new InvalidOperationException("Cannot update the 'Id' property.");
throw new InvalidOperationException("Cannot update the 'id' property.");
}
}
else
Expand All @@ -121,8 +107,8 @@ internal static async Task SetValueInternalAsync(JObject originalItem, T newItem
throw new InvalidOperationException(string.Format("The document must have an 'id' property."));
}

Uri documentUri = UriFactory.CreateDocumentUri(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName, originalId);
await context.Service.ReplaceDocumentAsync(documentUri, newItem);
Container container = context.Service.GetContainer(context.ResolvedAttribute.DatabaseName, context.ResolvedAttribute.CollectionName);
await container.ReplaceItemAsync<T>(newItem, originalId);
}
}

Expand Down
40 changes: 0 additions & 40 deletions src/WebJobs.Extensions.CosmosDB/Config/CosmosDBConnectionString.cs
View file
Open in desktop

This file was deleted.

Loading