Skip to content

Commit e2fd111

Browse files
liliankasemliliankasem
committed
Cosmos DB converter for SDK-type support and samples (#1406)
1 parent 1d0646a commit e2fd111

21 files changed

Lines changed: 1175 additions & 5 deletions

‎extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptions.cs‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
// Copyright (c) .NET Foundation. All rights reserved.
2+
// Licensed under the MIT License. See License.txt in the project root for license information.
3+
4+
using System;
5+
using System.Collections.Concurrent;
6+
using Azure.Core;
7+
using Microsoft.Azure.Cosmos;
8+
using Microsoft.Azure.Functions.Worker.Extensions.CosmosDB;
9+
10+
namespace Microsoft.Azure.Functions.Worker
11+
{
12+
internal class CosmosDBBindingOptions
13+
{
14+
public string? ConnectionString { get; set; }
15+
16+
public string? AccountEndpoint { get; set; }
17+
18+
public TokenCredential? Credential { get; set; }
19+
20+
internal string BuildCacheKey(string connectionString, string region) => $"{connectionString}|{region}";
21+
internal ConcurrentDictionary<string, CosmosClient> ClientCache { get; } = new ConcurrentDictionary<string, CosmosClient>();
22+
23+
internal virtual CosmosClient GetClient(string connection, string preferredLocations = "")
24+
{
25+
if (string.IsNullOrEmpty(connection))
26+
{
27+
throw new ArgumentNullException(nameof(connection));
28+
}
29+
30+
string cacheKey = BuildCacheKey(connection, preferredLocations);
31+
32+
CosmosClientOptions cosmosClientOptions = new ()
33+
{
34+
ConnectionMode = ConnectionMode.Gateway
35+
};
36+
37+
if (!string.IsNullOrEmpty(preferredLocations))
38+
{
39+
cosmosClientOptions.ApplicationPreferredRegions = Utilities.ParsePreferredLocations(preferredLocations);
40+
}
41+
42+
return ClientCache.GetOrAdd(cacheKey, (c) => CreateService(cosmosClientOptions));
43+
}
44+
45+
private CosmosClient CreateService(CosmosClientOptions cosmosClientOptions)
46+
{
47+
return string.IsNullOrEmpty(ConnectionString)
48+
? new CosmosClient(AccountEndpoint, Credential, cosmosClientOptions) // AAD auth
49+
: new CosmosClient(ConnectionString, cosmosClientOptions); // Connection string based auth
50+
}
51+
}
52+
}

‎extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptionsSetup.cs‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
// Copyright (c) .NET Foundation. All rights reserved.
2+
// Licensed under the MIT License. See License.txt in the project root for license information.
3+
4+
using System;
5+
using Microsoft.Azure.Functions.Worker.Extensions;
6+
using Microsoft.Azure.Functions.Worker.Extensions.CosmosDB;
7+
using Microsoft.Extensions.Azure;
8+
using Microsoft.Extensions.Configuration;
9+
using Microsoft.Extensions.Options;
10+
11+
namespace Microsoft.Azure.Functions.Worker
12+
{
13+
internal class CosmosDBBindingOptionsSetup : IConfigureNamedOptions<CosmosDBBindingOptions>
14+
{
15+
private readonly IConfiguration _configuration;
16+
private readonly AzureComponentFactory _componentFactory;
17+
18+
public CosmosDBBindingOptionsSetup(IConfiguration configuration, AzureComponentFactory componentFactory)
19+
{
20+
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
21+
_componentFactory = componentFactory ?? throw new ArgumentNullException(nameof(componentFactory));
22+
}
23+
24+
public void Configure(CosmosDBBindingOptions options)
25+
{
26+
Configure(Options.DefaultName, options);
27+
}
28+
29+
public void Configure(string name, CosmosDBBindingOptions options)
30+
{
31+
IConfigurationSection connectionSection = _configuration.GetWebJobsConnectionStringSection(name);
32+
33+
if (!connectionSection.Exists())
34+
{
35+
throw new InvalidOperationException($"Cosmos DB connection configuration '{name}' does not exist. " +
36+
"Make sure that it is a defined App Setting.");
37+
}
38+
39+
if (!string.IsNullOrWhiteSpace(connectionSection.Value))
40+
{
41+
options.ConnectionString = connectionSection.Value;
42+
}
43+
else
44+
{
45+
options.AccountEndpoint = connectionSection[Constants.AccountEndpoint];
46+
if (string.IsNullOrWhiteSpace(options.AccountEndpoint))
47+
{
48+
throw new InvalidOperationException($"Connection should have an '{Constants.AccountEndpoint}' property or be a " +
49+
$"string representing a connection string.");
50+
}
51+
52+
options.Credential = _componentFactory.CreateTokenCredential(connectionSection);
53+
}
54+
}
55+
}
56+
}

‎extensions/Worker.Extensions.CosmosDB/src/Constants.cs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
2+
// Copyright (c) .NET Foundation. All rights reserved.
3+
// Licensed under the MIT License. See License.txt in the project root for license information.
4+
5+
namespace Microsoft.Azure.Functions.Worker.Extensions.CosmosDB
6+
{
7+
internal static class Constants
8+
{
9+
internal const string CosmosExtensionName = "CosmosDB";
10+
internal const string ConfigurationSectionName = "AzureWebJobs";
11+
internal const string ConnectionStringsSectionName = "ConnectionStrings";
12+
internal const string AccountEndpoint = "accountEndpoint";
13+
internal const string JsonContentType = "application/json";
14+
}
15+
}

‎extensions/Worker.Extensions.CosmosDB/src/CosmosDBConverter.cs‎

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
// Copyright (c) .NET Foundation. All rights reserved.
2+
// Licensed under the MIT License. See License.txt in the project root for license information.
3+
4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
7+
using System.Reflection;
8+
using System.Threading.Tasks;
9+
using Microsoft.Azure.Cosmos;
10+
using Microsoft.Azure.Functions.Worker.Core;
11+
using Microsoft.Azure.Functions.Worker.Converters;
12+
using Microsoft.Azure.Functions.Worker.Extensions.CosmosDB;
13+
using Microsoft.Extensions.Options;
14+
using Microsoft.Extensions.Logging;
15+
16+
namespace Microsoft.Azure.Functions.Worker
17+
{
18+
/// <summary>
19+
/// Converter to bind Cosmos DB type parameters.
20+
/// </summary>
21+
internal class CosmosDBConverter : IInputConverter
22+
{
23+
private readonly IOptionsSnapshot<CosmosDBBindingOptions> _cosmosOptions;
24+
private readonly ILogger<CosmosDBConverter> _logger;
25+
26+
public CosmosDBConverter(IOptionsSnapshot<CosmosDBBindingOptions> cosmosOptions, ILogger<CosmosDBConverter> logger)
27+
{
28+
_cosmosOptions = cosmosOptions ?? throw new ArgumentNullException(nameof(cosmosOptions));
29+
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
30+
}
31+
32+
public async ValueTask<ConversionResult> ConvertAsync(ConverterContext context)
33+
{
34+
return context?.Source switch
35+
{
36+
ModelBindingData binding => await ConvertFromBindingDataAsync(context, binding),
37+
_ => ConversionResult.Unhandled(),
38+
};
39+
}
40+
41+
private async ValueTask<ConversionResult> ConvertFromBindingDataAsync(ConverterContext context, ModelBindingData modelBindingData)
42+
{
43+
if (!IsCosmosExtension(modelBindingData))
44+
{
45+
return ConversionResult.Unhandled();
46+
}
47+
48+
try
49+
{
50+
var cosmosAttribute = GetBindingDataContent(modelBindingData);
51+
object result = await ToTargetTypeAsync(context.TargetType, cosmosAttribute);
52+
53+
if (result is not null)
54+
{
55+
return ConversionResult.Success(result);
56+
}
57+
}
58+
catch (Exception ex)
59+
{
60+
return ConversionResult.Failed(ex);
61+
}
62+
63+
return ConversionResult.Unhandled();
64+
}
65+
66+
private bool IsCosmosExtension(ModelBindingData bindingData)
67+
{
68+
if (bindingData?.Source is not Constants.CosmosExtensionName)
69+
{
70+
_logger.LogTrace("Source '{source}' is not supported by {converter}", bindingData?.Source, nameof(CosmosDBConverter));
71+
return false;
72+
}
73+
74+
return true;
75+
}
76+
77+
private CosmosDBInputAttribute GetBindingDataContent(ModelBindingData bindingData)
78+
{
79+
return bindingData?.ContentType switch
80+
{
81+
Constants.JsonContentType => bindingData.Content.ToObjectFromJson<CosmosDBInputAttribute>(),
82+
_ => throw new NotSupportedException($"Unexpected content-type. Currently only '{Constants.JsonContentType}' is supported.")
83+
};
84+
}
85+
86+
private async Task<object> ToTargetTypeAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute) => targetType switch
87+
{
88+
Type _ when targetType == typeof(CosmosClient) => CreateCosmosClient<CosmosClient>(cosmosAttribute),
89+
Type _ when targetType == typeof(Database) => CreateCosmosClient<Database>(cosmosAttribute),
90+
Type _ when targetType == typeof(Container) => CreateCosmosClient<Container>(cosmosAttribute),
91+
_ => await CreateTargetObjectAsync(targetType, cosmosAttribute)
92+
};
93+
94+
private async Task<object> CreateTargetObjectAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute)
95+
{
96+
MethodInfo createPOCOMethod;
97+
98+
if (targetType.GenericTypeArguments.Any())
99+
{
100+
targetType = targetType.GenericTypeArguments.FirstOrDefault();
101+
102+
createPOCOMethod = GetType()
103+
.GetMethod(nameof(CreatePOCOCollectionAsync), BindingFlags.Instance | BindingFlags.NonPublic)
104+
.MakeGenericMethod(targetType);
105+
}
106+
else
107+
{
108+
createPOCOMethod = GetType()
109+
.GetMethod(nameof(CreatePOCOAsync), BindingFlags.Instance | BindingFlags.NonPublic)
110+
.MakeGenericMethod(targetType);
111+
}
112+
113+
114+
var container = CreateCosmosClient<Container>(cosmosAttribute) as Container;
115+
116+
if (container is null)
117+
{
118+
throw new InvalidOperationException($"Unable to create Cosmos container client for '{cosmosAttribute.ContainerName}'.");
119+
}
120+
121+
return await (Task<object>)createPOCOMethod.Invoke(this, new object[] { container, cosmosAttribute });
122+
}
123+
124+
private async Task<object> CreatePOCOAsync<T>(Container container, CosmosDBInputAttribute cosmosAttribute)
125+
{
126+
if (String.IsNullOrEmpty(cosmosAttribute.Id) || String.IsNullOrEmpty(cosmosAttribute.PartitionKey))
127+
{
128+
throw new InvalidOperationException("The 'Id' and 'PartitionKey' properties of a CosmosDB single-item input binding cannot be null or empty.");
129+
}
130+
131+
ItemResponse<T> item = await container.ReadItemAsync<T>(cosmosAttribute.Id, new PartitionKey(cosmosAttribute.PartitionKey));
132+
133+
if (item is null || item?.StatusCode is not System.Net.HttpStatusCode.OK || item.Resource is null)
134+
{
135+
throw new InvalidOperationException($"Unable to retrieve document with ID '{cosmosAttribute.Id}' and PartitionKey '{cosmosAttribute.PartitionKey}'");
136+
}
137+
138+
return item.Resource;
139+
}
140+
141+
private async Task<object> CreatePOCOCollectionAsync<T>(Container container, CosmosDBInputAttribute cosmosAttribute)
142+
{
143+
QueryDefinition queryDefinition = null!;
144+
if (!String.IsNullOrEmpty(cosmosAttribute.SqlQuery))
145+
{
146+
queryDefinition = new QueryDefinition(cosmosAttribute.SqlQuery);
147+
if (cosmosAttribute.SqlQueryParameters?.Count() > 0)
148+
{
149+
foreach (var parameter in cosmosAttribute.SqlQueryParameters)
150+
{
151+
queryDefinition.WithParameter(parameter.Key, parameter.Value.ToString());
152+
}
153+
}
154+
}
155+
156+
PartitionKey partitionKey = String.IsNullOrEmpty(cosmosAttribute.PartitionKey)
157+
? PartitionKey.None
158+
: new PartitionKey(cosmosAttribute.PartitionKey);
159+
160+
// Workaround until bug in Cosmos SDK is fixed
161+
// Currently pending release: https://github.com/Azure/azure-cosmos-dotnet-v3/commit/d6e04a92f8778565eb1d1452738d37c7faf3c47a
162+
QueryRequestOptions queryRequestOptions = new();
163+
if (partitionKey != PartitionKey.None)
164+
{
165+
queryRequestOptions = new() { PartitionKey = partitionKey };
166+
}
167+
168+
using (var iterator = container.GetItemQueryIterator<T>(queryDefinition: queryDefinition, requestOptions: queryRequestOptions))
169+
{
170+
if (iterator is null)
171+
{
172+
throw new InvalidOperationException($"Unable to retrieve documents for container '{container.Id}'.");
173+
}
174+
175+
return await ExtractCosmosDocumentsAsync(iterator);
176+
}
177+
}
178+
179+
private async Task<IList<T>> ExtractCosmosDocumentsAsync<T>(FeedIterator<T> iterator)
180+
{
181+
var documentList = new List<T>();
182+
while (iterator.HasMoreResults)
183+
{
184+
FeedResponse<T> response = await iterator.ReadNextAsync();
185+
documentList.AddRange(response.Resource);
186+
}
187+
return documentList;
188+
}
189+
190+
private T CreateCosmosClient<T>(CosmosDBInputAttribute cosmosAttribute)
191+
{
192+
var cosmosDBOptions = _cosmosOptions.Get(cosmosAttribute?.Connection);
193+
CosmosClient cosmosClient = cosmosDBOptions.GetClient(cosmosAttribute?.Connection!, cosmosAttribute?.PreferredLocations!);
194+
195+
Type targetType = typeof(T);
196+
object cosmosReference = targetType switch
197+
{
198+
Type _ when targetType == typeof(Database) => cosmosClient.GetDatabase(cosmosAttribute?.DatabaseName),
199+
Type _ when targetType == typeof(Container) => cosmosClient.GetContainer(cosmosAttribute?.DatabaseName, cosmosAttribute?.ContainerName),
200+
_ => cosmosClient
201+
};
202+
203+
return (T)cosmosReference;
204+
}
205+
}
206+
}

‎extensions/Worker.Extensions.CosmosDB/src/CosmosDBInputAttribute.cs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
11
// Copyright (c) .NET Foundation. All rights reserved.
22
// Licensed under the MIT License. See License.txt in the project root for license information.
33

4+
using System.Collections.Generic;
45
using Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
56

67
namespace Microsoft.Azure.Functions.Worker
78
{
9+
[SupportsDeferredBinding]
810
public sealed class CosmosDBInputAttribute : InputBindingAttribute
911
{
1012
/// <summary>
@@ -44,7 +46,7 @@ public CosmosDBInputAttribute(string databaseName, string containerName)
4446

4547
/// <summary>
4648
/// Optional.
47-
/// When specified on an output binding and <see cref="CreateIfNotExists"/> is true, defines the partition key
49+
/// When specified on an output binding and <see cref="CreateIfNotExists"/> is true, defines the partition key
4850
/// path for the created container.
4951
/// When specified on an input binding, specifies the partition key value for the lookup.
5052
/// May include binding parameters.
@@ -67,5 +69,11 @@ public CosmosDBInputAttribute(string databaseName, string containerName)
6769
/// PreferredLocations = "East US,South Central US,North Europe"
6870
/// </example>
6971
public string? PreferredLocations { get; set; }
72+
73+
/// <summary>
74+
/// Optional.
75+
/// Defines the parameters to be used with the SqlQuery
76+
/// </summary>
77+
public IDictionary<string, object>? SqlQueryParameters { get; set; }
7078
}
7179
}

0 commit comments

Comments
 (0)