Blob converter for ParameterBindingData support (#1108)

Co-authored-by: Surgupta <[email protected]>

Author: liliankasem

extensions/Worker.Extensions.Shared/Configuration/ConfigurationExtensions.cs

@@ -0,0 +1,61 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using Microsoft.Extensions.Configuration;
+
+namespace Microsoft.Azure.Functions.Worker.Extensions
+{
+    internal static class ConfigurationExtensions
+    {
+        private const string WebJobsConfigurationSectionName = "AzureWebJobs";
+        private const string ConnectionStringsConfigurationSectionName = "ConnectionStrings";
+
+        /// <summary>
+        /// Gets the configuration section for a given connection name.
+        /// </summary>
+        /// <param name="configuration">The configuration.</param>
+        /// <param name="connectionName">The connection string key.</param>
+        internal static IConfigurationSection GetWebJobsConnectionStringSection(this IConfiguration configuration, string connectionName)
+        {
+            // first try prefixing
+            string prefixedConnectionStringName = GetPrefixedConnectionStringName(connectionName);
+            IConfigurationSection section = configuration.GetConnectionStringOrSetting(prefixedConnectionStringName);
+
+            if (!section.Exists())
+            {
+                // next try a direct un-prefixed lookup
+                section = configuration.GetConnectionStringOrSetting(connectionName);
+            }
+
+            return section;
+        }
+
+        /// <summary>
+        /// Creates a WebJobs specific prefixed string using a given connection name.
+        /// </summary>
+        /// <param name="connectionName">The connection string key.</param>
+        private static string GetPrefixedConnectionStringName(string connectionName)
+        {
+            return WebJobsConfigurationSectionName + connectionName;
+        }
+
+        /// <summary>
+        /// Looks for a connection string by first checking the ConfigurationStrings section, and then the root.
+        /// </summary>
+        /// <param name="configuration">The configuration.</param>
+        /// <param name="connectionName">The connection string key.</param>
+        private static IConfigurationSection GetConnectionStringOrSetting(this IConfiguration configuration, string connectionName)
+        {
+            if (configuration.GetSection(ConnectionStringsConfigurationSectionName).Exists())
+            {
+                IConfigurationSection onConnectionStrings = configuration.GetSection(ConnectionStringsConfigurationSectionName).GetSection(connectionName);
+                if (onConnectionStrings.Exists())
+                {
+                    return onConnectionStrings;
+                }
+            }
+
+            return configuration.GetSection(connectionName);
+        }
+    }
+}

extensions/Worker.Extensions.Storage.Blobs/release_notes.md

@@ -1,4 +1,7 @@
 ### Release notes
 <!-- Please add your release notes in the following format:
 - My change description (#PR/#issue)
--->
+-->
+
+- Add support for SDK-type bindings via deferred binding feature #1108
+- Assign cardinality correctly for Blob collection scenarios #1271

extensions/Worker.Extensions.Storage.Blobs/src/BlobInputAttribute.cs

@@ -1,11 +1,11 @@
 // 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 Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
 
 namespace Microsoft.Azure.Functions.Worker
 {
+    [SupportsDeferredBinding]
     public sealed class BlobInputAttribute : InputBindingAttribute
     {
         private readonly string _blobPath;

extensions/Worker.Extensions.Storage.Blobs/src/BlobStorageConverter.cs

@@ -0,0 +1,236 @@
+// 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.IO;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Storage.Blobs;
+using Azure.Storage.Blobs.Specialized;
+using Microsoft.Azure.Functions.Worker.Converters;
+using Microsoft.Azure.Functions.Worker.Core;
+using Microsoft.Extensions.Options;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.Azure.Functions.Worker
+{
+    /// <summary>
+    /// Converter to bind Blob Storage type parameters.
+    /// </summary>
+    internal class BlobStorageConverter : IInputConverter
+    {
+        private readonly IOptions<WorkerOptions> _workerOptions;
+        private readonly IOptionsSnapshot<BlobStorageBindingOptions> _blobOptions;
+
+        private readonly ILogger<BlobStorageConverter> _logger;
+
+        public BlobStorageConverter(IOptions<WorkerOptions> workerOptions, IOptionsSnapshot<BlobStorageBindingOptions> blobOptions, ILogger<BlobStorageConverter> logger)
+        {
+            _workerOptions = workerOptions ?? throw new ArgumentNullException(nameof(workerOptions));
+            _blobOptions = blobOptions ?? throw new ArgumentNullException(nameof(blobOptions));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        }
+
+        public async ValueTask<ConversionResult> ConvertAsync(ConverterContext context)
+        {
+            return context?.Source switch
+            {
+                ModelBindingData binding => await ConvertFromBindingDataAsync(context, binding),
+                CollectionModelBindingData binding => await ConvertFromCollectionBindingDataAsync(context, binding),
+                _ => ConversionResult.Unhandled(),
+            };
+        }
+
+        private async ValueTask<ConversionResult> ConvertFromBindingDataAsync(ConverterContext context, ModelBindingData modelBindingData)
+        {
+            if (!IsBlobExtension(modelBindingData))
+            {
+                return ConversionResult.Unhandled();
+            }
+
+            try
+            {
+                Dictionary<string, string> content = GetBindingDataContent(modelBindingData);
+                var result = await ConvertModelBindingDataAsync(content, context.TargetType, modelBindingData);
+
+                if (result is not null)
+                {
+                    return ConversionResult.Success(result);
+                }
+            }
+            catch (Exception ex)
+            {
+                return ConversionResult.Failed(ex);
+            }
+
+            return ConversionResult.Unhandled();
+        }
+
+        private async ValueTask<ConversionResult> ConvertFromCollectionBindingDataAsync(ConverterContext context, CollectionModelBindingData collectionModelBindingData)
+        {
+            var blobCollection = new List<object>(collectionModelBindingData.ModelBindingDataArray.Length);
+            Type elementType = context.TargetType.IsArray ? context.TargetType.GetElementType() : context.TargetType.GenericTypeArguments[0];
+
+            try
+            {
+                foreach (ModelBindingData modelBindingData in collectionModelBindingData.ModelBindingDataArray)
+                {
+                    if (!IsBlobExtension(modelBindingData))
+                    {
+                        return ConversionResult.Unhandled();
+                    }
+
+                    Dictionary<string, string> content = GetBindingDataContent(modelBindingData);
+                    var element = await ConvertModelBindingDataAsync(content, elementType, modelBindingData);
+
+                    if (element is not null)
+                    {
+                        blobCollection.Add(element);
+                    }
+                }
+
+                var methodName = context.TargetType.IsArray ? nameof(CloneToArray) : nameof(CloneToList);
+                var result = ToTargetTypeCollection(blobCollection, methodName, elementType);
+
+                return ConversionResult.Success(result);
+            }
+            catch (Exception ex)
+            {
+                return ConversionResult.Failed(ex);
+            }
+        }
+
+        private bool IsBlobExtension(ModelBindingData bindingData)
+        {
+            if (bindingData?.Source is not Constants.BlobExtensionName)
+            {
+                _logger.LogTrace("Source '{source}' is not supported by {converter}", bindingData?.Source, nameof(BlobStorageConverter));
+                return false;
+            }
+
+            return true;
+        }
+
+        private Dictionary<string, string> GetBindingDataContent(ModelBindingData bindingData)
+        {
+            return bindingData?.ContentType switch
+            {
+                Constants.JsonContentType => new Dictionary<string, string>(bindingData?.Content?.ToObjectFromJson<Dictionary<string, string>>(), StringComparer.OrdinalIgnoreCase),
+                _ => throw new NotSupportedException($"Unexpected content-type. Currently only {Constants.JsonContentType} is supported.")
+            };
+        }
+
+        private async Task<object?> ConvertModelBindingDataAsync(IDictionary<string, string> content, Type targetType, ModelBindingData bindingData)
+        {
+            content.TryGetValue(Constants.Connection, out var connectionName);
+            content.TryGetValue(Constants.ContainerName, out var containerName);
+            content.TryGetValue(Constants.BlobName, out var blobName);
+
+            if (string.IsNullOrEmpty(connectionName) || string.IsNullOrEmpty(containerName))
+            {
+                throw new ArgumentNullException("'Connection' and 'ContainerName' cannot be null or empty");
+            }
+
+            return await ToTargetTypeAsync(targetType, connectionName, containerName, blobName);
+        }
+
+        private async Task<object?> ToTargetTypeAsync(Type targetType, string connectionName, string containerName, string blobName) => targetType switch
+        {
+            Type _ when targetType == typeof(String) => await GetBlobStringAsync(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(Stream) => await GetBlobStreamAsync(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(Byte[]) => await GetBlobBinaryDataAsync(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(BlobBaseClient) => CreateBlobClient<BlobBaseClient>(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(BlobClient) => CreateBlobClient<BlobClient>(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(BlockBlobClient) => CreateBlobClient<BlockBlobClient>(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(PageBlobClient) => CreateBlobClient<PageBlobClient>(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(AppendBlobClient) => CreateBlobClient<AppendBlobClient>(connectionName, containerName, blobName),
+            Type _ when targetType == typeof(BlobContainerClient) => CreateBlobContainerClient(connectionName, containerName),
+            _ => await DeserializeToTargetObjectAsync(targetType, connectionName, containerName, blobName)
+        };
+
+        private async Task<object?> DeserializeToTargetObjectAsync(Type targetType, string connectionName, string containerName, string blobName)
+        {
+            var content = await GetBlobStreamAsync(connectionName, containerName, blobName);
+            return _workerOptions?.Value?.Serializer?.Deserialize(content, targetType, CancellationToken.None);
+        }
+
+        private object? ToTargetTypeCollection(IEnumerable<object> blobCollection, string methodName, Type type)
+        {
+            blobCollection = blobCollection.Select(b => Convert.ChangeType(b, type));
+            MethodInfo method = typeof(BlobStorageConverter).GetMethod(methodName, BindingFlags.Static | BindingFlags.NonPublic);
+            MethodInfo genericMethod = method.MakeGenericMethod(type);
+
+            return genericMethod.Invoke(null, new[] { blobCollection.ToList() });
+        }
+
+        private static T[] CloneToArray<T>(IList<object> source)
+        {
+            return source.Cast<T>().ToArray();
+        }
+
+        private static IEnumerable<T> CloneToList<T>(IList<object> source)
+        {
+            return source.Cast<T>();
+        }
+
+        private async Task<string> GetBlobStringAsync(string connectionName, string containerName, string blobName)
+        {
+            var client = CreateBlobClient<BlobClient>(connectionName, containerName, blobName);
+            return await GetBlobContentStringAsync(client);
+        }
+
+        private async Task<string> GetBlobContentStringAsync(BlobClient client)
+        {
+            var download = await client.DownloadContentAsync();
+            return download.Value.Content.ToString();
+        }
+
+        private async Task<Byte[]> GetBlobBinaryDataAsync(string connectionName, string containerName, string blobName)
+        {
+            using MemoryStream stream = new();
+            var client = CreateBlobClient<BlobClient>(connectionName, containerName, blobName);
+            await client.DownloadToAsync(stream);
+            return stream.ToArray();
+        }
+
+        private async Task<Stream> GetBlobStreamAsync(string connectionName, string containerName, string blobName)
+        {
+            var client = CreateBlobClient<BlobClient>(connectionName, containerName, blobName);
+            var download = await client.DownloadStreamingAsync();
+            return download.Value.Content;
+        }
+
+        private BlobContainerClient CreateBlobContainerClient(string connectionName, string containerName)
+        {
+            var blobStorageOptions = _blobOptions.Get(connectionName);
+            BlobServiceClient blobServiceClient = blobStorageOptions.CreateClient();
+            BlobContainerClient container = blobServiceClient.GetBlobContainerClient(containerName);
+            return container;
+        }
+
+        private T CreateBlobClient<T>(string connectionName, string containerName, string blobName) where T : BlobBaseClient
+        {
+            if (string.IsNullOrEmpty(blobName))
+            {
+                throw new ArgumentNullException(nameof(blobName));
+            }
+
+            BlobContainerClient container = CreateBlobContainerClient(connectionName, containerName);
+
+            Type targetType = typeof(T);
+            BlobBaseClient blobClient = targetType switch
+            {
+                Type _ when targetType == typeof(BlobClient) => container.GetBlobClient(blobName),
+                Type _ when targetType == typeof(BlockBlobClient) => container.GetBlockBlobClient(blobName),
+                Type _ when targetType == typeof(PageBlobClient) => container.GetPageBlobClient(blobName),
+                Type _ when targetType == typeof(AppendBlobClient) => container.GetAppendBlobClient(blobName),
+                _ => container.GetBlobBaseClient(blobName)
+            };
+
+            return (T)blobClient;
+        }
+    }
+}

extensions/Worker.Extensions.Storage.Blobs/src/BlobTriggerAttribute.cs

@@ -5,6 +5,7 @@
 
 namespace Microsoft.Azure.Functions.Worker
 {
+    [SupportsDeferredBinding]
     public sealed class BlobTriggerAttribute : TriggerBindingAttribute
     {
         private readonly string _blobPath;

extensions/Worker.Extensions.Storage.Blobs/src/Config/BlobStorageBindingOptions.cs

@@ -0,0 +1,30 @@
+// 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 Azure.Core;
+using Azure.Storage.Blobs;
+
+namespace Microsoft.Azure.Functions.Worker
+{
+    internal class BlobStorageBindingOptions
+    {
+        public string? ConnectionString { get; set; }
+
+        public Uri? ServiceUri { get; set; }
+
+        public TokenCredential? Credential { get; set; }
+
+        public BlobClientOptions? BlobClientOptions { get; set; }
+
+        public BlobServiceClient CreateClient()
+        {
+            if (ServiceUri is not null && Credential is not null)
+            {
+                return new BlobServiceClient(ServiceUri, Credential, BlobClientOptions);
+            }
+
+            return new BlobServiceClient(ConnectionString, BlobClientOptions);
+        }
+    }
+}