TEAMModelOS/TEAMModelOS.SDK/DI/AzureStorage/AzureStorageTableExtensions.cs

using TEAMModelOS.SDK.Models.Table;
using Microsoft.Azure.Cosmos.Table;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Linq;
using TEAMModelOS.SDK;
using System.Reflection;

namespace TEAMModelOS.SDK.DI
{
    public static class AzureStorageTableExtensions
    {
 
        #region CloudTable 批次操作    

        /// <summary>
        /// 批次新增資料至Table儲存區。
        /// </summary>
        /// <param name="entities">欲快取的集合</param>
        /// <returns></returns>
        public static async Task<TableBatchResult> BatchInsertAsync<T>(this CloudTable table, IEnumerable<T> entities) where T : ITableEntity, new()
        {
            TableBatchOperation batchOperation = new TableBatchOperation();
            foreach (var cache in entities)
            {
                batchOperation.Insert(cache);
            }

            return await table.ExecuteBatchAsync(batchOperation);
        }

        #endregion

        #region CloudTable Get

        /// <summary>
        /// (同步)取得 table 中指定 PartitionKey 的所有集合。
        /// </summary>
        /// <returns></returns>
        public static IEnumerable<T> Get<T>(this CloudTable table, string partitionKey) where T : ITableEntity, new()
        {
            TableQuery<T> query = new TableQuery<T>().Where(TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, partitionKey));
            return table.ExecuteQuery(query);
        }

        /// <summary>
        /// (同步)取得指定 PartitionKey與 RowKey 的數據，效能：點查詢，最佳。
        /// </summary>
        /// <typeparam name="T">T</typeparam>       
        /// <param name="partitionKey">PartitionKey</param>   
        /// <param name="rowKey">RowKey</param>  
        /// <returns></returns>
        public static T Get<T>(this CloudTable table, string partitionKey, string rowKey) where T : ITableEntity, new()
        {
            TableOperation retrieveOperation = TableOperation.Retrieve<T>(partitionKey, rowKey);
            var retrieveResult = table.Execute(retrieveOperation);
            if (retrieveResult.Result != null)
            {
                //DynamicTableEntityJsonSeria1lizer
                return (T)retrieveResult.Result;
            }
            else
            {
                return default;
            }
        }

        public static DynamicTableEntity Get(this CloudTable table, string partitionKey, string rowKey)
        {
            TableOperation retrieveOperation = TableOperation.Retrieve(partitionKey, rowKey);
            var retrieveResult = table.Execute(retrieveOperation);
            if (retrieveResult.Result != null)
            {
                return retrieveResult.Result as DynamicTableEntity;
            }
            else
            {
                return null;
            }
        }

        /// <summary>
        /// 取得指定 PartitionKey 的所有集合(分頁)，效能：範圍查詢，次佳。
        /// </summary>
        /// <typeparam name="T">T</typeparam>       
        /// <param name="partitionKey">PartitionKey</param>
        /// <param name="takeCount">指定每次返回數量</param>
        /// <param name="specifyPropertys">指定只須返回的屬性，效能：伺服器端預測，降低延遲和成本</param>
        /// <param name="onProgress">要返回集合的委派</param>
        /// <param name="ct">CancellationToken</param>
        /// <returns></returns>
        public static async Task GetAsync<T>(this CloudTable table, string partitionKey, int? takeCount = null, List<string> specifyPropertys = null, Action<IList<T>> onProgress = null, CancellationToken ct = default(CancellationToken)) where T : ITableEntity, new()
        {
            TableQuery<T> tableQuery = null;
            TableContinuationToken token = null;
            var filter = TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, partitionKey);

            if (specifyPropertys != null)
                tableQuery = new TableQuery<T>().Where(filter).Select(specifyPropertys);
            else
                tableQuery = new TableQuery<T>().Where(filter);

            if (takeCount != null) tableQuery.TakeCount = takeCount;

            do
            {
                var queryResponse = await table.ExecuteQuerySegmentedAsync(tableQuery, token);
                token = queryResponse.ContinuationToken;
                onProgress?.Invoke(queryResponse.Results);
            } while (token != null && !ct.IsCancellationRequested);
        }

        /// <summary>
        /// 取得指定 PartitionKey 的所有集合，效能：範圍查詢，次佳。
        /// </summary>
        /// <typeparam name="T">T</typeparam>       
        /// <param name="partitionKey">PartitionKey</param>
        /// <param name="takeCount">指定每次返回數量</param>
        /// <param name="specifyPropertys">指定只須返回的屬性，效能：伺服器端預測，降低延遲和成本</param>
        /// <param name="onProgress">要返回集合的委派</param>
        /// <param name="ct">CancellationToken</param>
        /// <returns></returns>
        public static async Task<IEnumerable<T>> GetAsync<T>(this CloudTable table, string partitionKey, int? takeCount = null, List<string> specifyPropertys = null, CancellationToken ct = default(CancellationToken)) where T : ITableEntity, new()
        {
            TableQuery<T> tableQuery = null;
            TableContinuationToken token = null;
            var filter = TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, partitionKey);

            if (specifyPropertys != null)
                tableQuery = new TableQuery<T>().Where(filter).Select(specifyPropertys);
            else
                tableQuery = new TableQuery<T>().Where(filter);

            if (takeCount != null) tableQuery.TakeCount = takeCount;
            var items = new List<T>();

            do
            {
                var queryResponse = await table.ExecuteQuerySegmentedAsync(tableQuery, token);
                token = queryResponse.ContinuationToken;
                items.AddRange(queryResponse.Results);
            } while (token != null && !ct.IsCancellationRequested);
            return items;
        }

        /// <summary>
        /// 傳回指定partitionKey的Count數量
        /// </summary>        
        /// <param name="partitionKey"></param>
        /// <returns></returns>
        public static int GetCount(this CloudTable table, string partitionKey)
        {           

            TableQuery query = new TableQuery().Where(TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, partitionKey)).Select(new List<string> { "PartitionKey" });
            var queryResponse = table.ExecuteQuery(query).ToList();
            return queryResponse.Count();
        }

        #endregion

        #region CloudTable Get All
        /// <summary>
        /// (同步)取得 table 的所有集合，效能：資料表掃描，不佳。
        /// </summary>
        /// <returns></returns>
        public static IEnumerable<T> GetAll<T>(this CloudTable table) where T : ITableEntity, new()
        {
            return table.ExecuteQuery(new TableQuery<T>());
        }

        /// <summary>
        /// 取得 table 的所有集合(分頁)，效能：資料表掃描，不佳。
        /// </summary>
        /// <param name="takeCount">指定每次返回數量</param>
        /// <param name="specifyPropertys">指定只須返回的屬性，效能：伺服器端預測，降低延遲和成本</param>
        /// <param name="onProgress">要返回集合的委派</param>
        /// <returns></returns>        
        public static async Task GetAllAsync<T>(this CloudTable table, int? takeCount = null, List<string> specifyPropertys = null, Action<IList<T>> onProgress = null, CancellationToken ct = default(CancellationToken)) where T : ITableEntity, new()
        {
            TableContinuationToken token = null;
            TableQuery<T> tableQuery = new TableQuery<T>();
            if (specifyPropertys != null)
                tableQuery = new TableQuery<T>().Select(specifyPropertys);
            else
                tableQuery = new TableQuery<T>();
            if (takeCount != null) tableQuery.TakeCount = takeCount;
            var items = new List<T>();

            do
            {
                var queryResponse = await table.ExecuteQuerySegmentedAsync(tableQuery, token);
                token = queryResponse.ContinuationToken;
                items.AddRange(queryResponse.Results);
                onProgress?.Invoke(queryResponse.Results);
            } while (token != null && !ct.IsCancellationRequested);
        }

        /// <summary>
        /// 取得 table 的所有集合，效能：資料表掃描，不佳。
        /// </summary>
        /// <param name="takeCount">指定每次返回數量</param>
        /// <param name="specifyPropertys">指定只須返回的屬性，效能：伺服器端預測，降低延遲和成本</param>       
        /// <returns></returns>        
        public static async Task<IEnumerable<T>> GetAllAsync<T>(this CloudTable table, int? takeCount = null, List<string> specifyPropertys = null, CancellationToken ct = default(CancellationToken)) where T : ITableEntity, new()
        {
            TableContinuationToken token = null;
            TableQuery<T> tableQuery = new TableQuery<T>();
            if (takeCount != null) tableQuery.TakeCount = takeCount;
            if (specifyPropertys != null)
                tableQuery = new TableQuery<T>().Select(specifyPropertys);
            else
                tableQuery = new TableQuery<T>();
            var items = new List<T>();

            do
            {
                var queryResponse = await table.ExecuteQuerySegmentedAsync(tableQuery, token);
                token = queryResponse.ContinuationToken;
                items.AddRange(queryResponse.Results);
            } while (token != null && !ct.IsCancellationRequested);
            return items;
        }
        #endregion

        public static async Task<T> Save<T>(this CloudTable table, TableEntity entity) where T : TableEntity, new()
        {                  
            TableResult result = await table.ExecuteAsync(TableOperation.Insert(entity));
            return result.Result as T;
        }
        public static async Task<T> SaveOrUpdate<T>(this CloudTable table, TableEntity entity) where T : TableEntity, new()
        {
            TableResult result = await table.ExecuteAsync(TableOperation.InsertOrReplace(entity));
            return result.Result as T;
        }

        public static async Task<T> Update<T>(this CloudTable table, TableEntity entity) where T : TableEntity, new()
        {                       
            TableResult result = await table.ExecuteAsync(TableOperation.Replace(entity));
            return result.Result as T;
        }

        public static async Task<T> Delete<T>(this CloudTable table, TableEntity entity) where T : TableEntity, new()
        {
            TableResult result = await table.ExecuteAsync(TableOperation.Delete(entity));
            return result.Result as T;
        }

        public static async Task<List<T>> DeleteAll<T>(this CloudTable table,  List<T> entitys) where T : TableEntity, new()
        {
            if (!entitys.IsNotEmpty())
            {
                return null;
            }
            
            IList<Dictionary<string, List<T>>> listInfo = new List<Dictionary<string, List<T>>>();
            foreach (IGrouping<string, T> group in entitys.GroupBy(c => c.PartitionKey))
            {
                Dictionary<string, List<T>> dictInfo = new()
                {
                    { group.Key, group.ToList() }
                };
                listInfo.Add(dictInfo);
            }

            foreach (Dictionary<string, List<T>> dict in listInfo)
            {
                IList<TableResult> result = null;
                foreach (string key in dict.Keys)
                {
                    List<T> values = dict[key];
                     
                    int pageSize = 100;
                    int pages = (int)Math.Ceiling((double)values.Count / pageSize);
                    for (int i = 0; i < pages; i++)
                    {
                        List<T> lists = values.Skip((i) * pageSize).Take(pageSize).ToList();
                        TableBatchOperation batchOperation = new();
                        for (int j = 0; j < lists.Count; j++)
                        {
                            batchOperation.Delete(lists[j]);
                        }
                        result = await table.ExecuteBatchAsync(batchOperation);
                    }
                }
            }
            return entitys;
        }

        public static async Task<List<T>> SaveOrUpdateAll<T>(this CloudTable table, List<T> entitys) where T : TableEntity, new()
        {
            if (!entitys.IsNotEmpty()) return null;           
            
            IList<Dictionary<string, List<T>>> listInfo = new List<Dictionary<string, List<T>>>();
            foreach (IGrouping<string, T> group in entitys.GroupBy(c => c.PartitionKey))
            {
                Dictionary<string, List<T>> dictInfo = new Dictionary<string, List<T>>
                {
                    { group.Key, group.ToList() }
                };
                listInfo.Add(dictInfo);
            }

            foreach (Dictionary<string, List<T>> dict in listInfo)
            {
                IList<TableResult> result = null;
                foreach (string key in dict.Keys)
                {
                    List<T> values = dict[key];

                    int pageSize = 100;
                    int pages = (int)Math.Ceiling((double)values.Count / pageSize);
                    for (int i = 0; i < pages; i++)
                    {
                        List<T> lists = values.Skip((i) * pageSize).Take(pageSize).ToList();
                        TableBatchOperation batchOperation = new TableBatchOperation();
                        for (int j = 0; j < lists.Count; j++)
                        {
                            batchOperation.InsertOrReplace(lists[j]);
                        }
                        result = await table.ExecuteBatchAsync(batchOperation);
                    }
                }
            }
            return entitys;
        }
        public static async Task<List<T>> UpdateAll<T>(this CloudTable table, List<T> entitys) where T : TableEntity, new()
        {
            if (!entitys.IsNotEmpty()) return null;            
            
            IList<Dictionary<string, List<T>>> listInfo = new List<Dictionary<string, List<T>>>();
            foreach (IGrouping<string, T> group in entitys.GroupBy(c => c.PartitionKey))
            {
                Dictionary<string, List<T>> dictInfo = new Dictionary<string, List<T>>
                {
                    { group.Key, group.ToList() }
                };
                listInfo.Add(dictInfo);
            }

            foreach (Dictionary<string, List<T>> dict in listInfo)
            {
                IList<TableResult> result = null;
                foreach (string key in dict.Keys)
                {
                    List<T> values = dict[key];
                  
                    int pageSize = 100;
                    int pages = (int)Math.Ceiling((double)values.Count / pageSize);
                    for (int i = 0; i < pages; i++)
                    {
                        List<T> lists = values.Skip((i) * pageSize).Take(pageSize).ToList();
                        TableBatchOperation batchOperation = new TableBatchOperation();
                        for (int j = 0; j < lists.Count; j++)
                        {
                            batchOperation.Replace(lists[j]);
                        }
                        result = await table.ExecuteBatchAsync(batchOperation);
                    }
                }
            }
            return entitys;
        }
        public static async Task<List<T>> SaveAll<T>(this CloudTable table, List<T> entitys) where T : TableEntity, new()
        {
            if (!entitys.IsNotEmpty()) return null;            
           
            IList<Dictionary<string, List<T>>> listInfo = new List<Dictionary<string, List<T>>>();
            foreach (IGrouping<string, T> group in entitys.GroupBy(c => c.PartitionKey))
            {
                Dictionary<string, List<T>> dictInfo = new Dictionary<string, List<T>>
                {
                    { group.Key, group.ToList() }
                };
                listInfo.Add(dictInfo);
            }

            foreach (Dictionary<string, List<T>> dict in listInfo)
            {
                IList<TableResult> result = null;
                foreach (string key in dict.Keys)
                {
                    List<T> values = dict[key];
                  
                    int pageSize = 100;
                    int pages = (int)Math.Ceiling((double)values.Count / pageSize);
                    for (int i = 0; i < pages; i++)
                    {
                        List<T> lists = values.Skip((i) * pageSize).Take(pageSize).ToList();
                        TableBatchOperation batchOperation = new TableBatchOperation();
                        for (int j = 0; j < lists.Count; j++)
                        {
                            batchOperation.Insert(lists[j]);
                        }
                        result = await table.ExecuteBatchAsync(batchOperation);
                    }
                }
            }
            return entitys;
        }
        public static   List<T>  FindListByDictSync<T>(this CloudTable table, Dictionary<string, object> dict) where T : TableEntity, new()
        {
            var exQuery = new TableQuery<T>();
            StringBuilder builder = new();
            if (null != dict && dict.Count > 0)
            {
                var keys = dict.Keys;
                int index = 1;
                foreach (string key in keys)
                {
                    if (dict[key] != null && !string.IsNullOrEmpty(dict[key].ToString()))
                    {
                        string typeStr = SwitchType<T>(dict[key], key);
                        if (string.IsNullOrEmpty(typeStr))
                        {
                            continue;
                        }
                        if (index == 1)
                        {
                            //builder.Append(TableQuery.GenerateFilterCondition(key, QueryComparisons.Equal, dict[key].ToString()));
                            builder.Append(typeStr);

                        }
                        else
                        {
                            //builder.Append("  " + TableOperators.And + "  " + TableQuery.GenerateFilterCondition(key, QueryComparisons.Equal, dict[key].ToString()));
                            builder.Append("  " + TableOperators.And + "  " + typeStr);

                        }
                        index++;
                    }
                    else
                    {
                        throw new Exception("The parameter must have value!");
                    }
                }

                exQuery.Where(builder.ToString());
                return   QueryListSync<T>(exQuery, table);
            }
            else
            {
                return null;
            }
        }
        public static async Task<List<T>> FindListByDict<T>(this CloudTable table, Dictionary<string, object> dict) where T : TableEntity, new()
        {            
            var exQuery = new TableQuery<T>();
            StringBuilder builder = new();
            if (null != dict && dict.Count > 0)
            {
                var keys = dict.Keys;
                int index = 1;
                foreach (string key in keys)
                {
                    if (dict[key] != null && !string.IsNullOrEmpty(dict[key].ToString()))
                    {
                        string typeStr = SwitchType<T>(dict[key], key);
                        if (string.IsNullOrEmpty(typeStr))
                        {
                            continue;
                        }
                        if (index == 1)
                        {
                            //builder.Append(TableQuery.GenerateFilterCondition(key, QueryComparisons.Equal, dict[key].ToString()));
                            builder.Append(typeStr);

                        }
                        else
                        {
                            //builder.Append("  " + TableOperators.And + "  " + TableQuery.GenerateFilterCondition(key, QueryComparisons.Equal, dict[key].ToString()));
                            builder.Append("  " + TableOperators.And + "  " + typeStr);

                        }
                        index++;
                    }
                    else
                    {
                        throw new Exception("The parameter must have value!");
                    }
                }

                exQuery.Where(builder.ToString());
                return await QueryList<T>(exQuery, table);
            }
            else
            {
                return null;
            }
        }

        private static List<T> QueryListSync<T>(TableQuery<T> exQuery, CloudTable TableName) where T : TableEntity, new()
        {
            TableContinuationToken continuationToken = null;
            List<T> entitys = new();
            do
            {
                var result =   TableName.ExecuteQuerySegmented(exQuery, continuationToken);
                if (result.Results.Count > 0)
                {
                    entitys.AddRange(result.ToList());
                }
                continuationToken = result.ContinuationToken;
            } while (continuationToken != null);
            return entitys;
        }

        private static async Task<List<T>> QueryList<T>(TableQuery<T> exQuery, CloudTable TableName  ) where T : TableEntity, new()
        {
            TableContinuationToken continuationToken = null;
            List<T> entitys = new();
            do
            {
                var result = await TableName.ExecuteQuerySegmentedAsync(exQuery, continuationToken);
                if (result.Results.Count > 0)
                {
                    entitys.AddRange(result.ToList());
                }
                continuationToken = result.ContinuationToken;
            } while (continuationToken != null);
            return entitys;
        }

        private static string SwitchType<T>(object obj, string key)
        {
            Type objType = typeof(T);
            PropertyInfo property = objType.GetProperty(key);

            //Type s = obj.GetType();
            //TypeCode typeCode = Type.GetTypeCode(s);
            if (property == null)
            {
                //return null;
                throw new Exception(objType.FullName + " PropertyInfo doesn't include this parameter :" + key);
            }
            TypeCode typeCode = Type.GetTypeCode(property.PropertyType);


            switch (typeCode)
            {

                case TypeCode.String: return TableQuery.GenerateFilterCondition(key, QueryComparisons.Equal, obj.ToString());
                case TypeCode.Int32: return TableQuery.GenerateFilterConditionForInt(key, QueryComparisons.Equal, int.Parse(obj.ToString()));
                case TypeCode.Double: return TableQuery.GenerateFilterConditionForDouble(key, QueryComparisons.Equal, (double)obj);
                case TypeCode.Byte: return TableQuery.GenerateFilterConditionForBinary(key, QueryComparisons.Equal, (byte[])obj);
                case TypeCode.Boolean: return TableQuery.GenerateFilterConditionForBool(key, QueryComparisons.Equal, (bool)obj);
                case TypeCode.DateTime: return TableQuery.GenerateFilterConditionForDate(key, QueryComparisons.Equal, (DateTimeOffset)obj);
                case TypeCode.Int64: return TableQuery.GenerateFilterConditionForLong(key, QueryComparisons.Equal, long.Parse(obj.ToString()));
                default: return null;
            }
        }


        /// <summary>
        /// 查询条件传string类型
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="azureStorage"></param>
        /// <param name="strWhere">查询条件| 例:$"RowKey {QueryComparisons.GreaterThanOrEqual} '{123456}' {TableOperators.And} RowKey {QueryComparisons.LessThanOrEqual} '{123456}'" </param>
        /// <returns></returns>
        public static async Task<List<T>> QueryWhereString<T>(this CloudTable table, string strWhere = null) where T : TableEntity, new()
        {            
            var exQuery = new TableQuery<T>();
            exQuery.Where(strWhere);
            return await QueryList<T>(exQuery, table);
        }

        /// <summary>
        /// 删除单个记录
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="azureStorage"></param>
        /// <param name="partitionKey"></param>
        /// <param name="rowKey"></param>
        /// <returns></returns>
        public static async Task<TableResult> DeleteSingle<T>(this CloudTable table, string partitionKey, string rowKey) where T : ITableEntity, new()
        {
            // query all rows            
            var queryOperation = TableOperation.Retrieve<T>(partitionKey, rowKey);

            var tableResult = await table.ExecuteAsync(queryOperation);
            if (tableResult.Result is T item)
            {                
                var dels = await table.ExecuteAsync(TableOperation.Delete(item));
                return dels;
            }
            else return null;
        }

        /// <summary>
        /// 删除多个  带有后期优化该方法
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="azureStorage"></param>
        /// <param name="rowKey"></param>
        /// <returns></returns>
        public static async Task<List<string>> DeleteStringWhere<T>(this CloudTable table, string rowKey) where T : ITableEntity, new()
        {
            // query all rows           

            var exQuery = new TableQuery<T>();
            exQuery.Where(rowKey);
            TableContinuationToken continuationToken = null;
            List<T> entitys = new();
            do
            {
                var result = await table.ExecuteQuerySegmentedAsync(exQuery, continuationToken);                
                if (result.Results.Count > 0)
                {
                    entitys.AddRange(result.ToList());
                }
                continuationToken = result.ContinuationToken;
            } while (continuationToken != null);

            List<string> ster = new List<string>();
            foreach (var item in entitys)
            {                
                var tableResult = await table.ExecuteAsync(TableOperation.Retrieve<T>(item.PartitionKey, item.RowKey));
                if (tableResult.Result is T items)
                {
                    var deleteOperation = TableOperation.Delete(items);
                    await table.ExecuteAsync(deleteOperation);
                    ster.Add(items.RowKey);
                }
            }

            return ster;
        }
    }
}