Merge branch 'main' into copilot/remove-legacy-prompt-feature

philnach · web-flow · commit 25027736e444 · 2025-10-10T15:43:12.000-07:00
diff --git a/Extensions/AwsS3/Cosmos.DataTransfer.AwsS3Storage/AwsS3DataSink.cs b/Extensions/AwsS3/Cosmos.DataTransfer.AwsS3Storage/AwsS3DataSink.cs
@@ -11,7 +11,7 @@ public async Task WriteToTargetAsync(Func<Stream, Task> writeToStream, IConfigur
             var settings = config.Get<AwsS3SinkSettings>();
             settings.Validate();
 
-            logger.LogInformation("Saving file {File} to AWS S3 Bucket '{BucketName}'", settings.FileName, settings.S3BucketName);
+            logger.LogInformation("Saving file {File} to AWS S3 Bucket '{BucketName}'", settings!.FileName, settings.S3BucketName);
 
             using var s3 = new S3Client(settings.S3AccessKey, settings.S3SecretKey, settings.S3Region);
             await using var stream = new MemoryStream();
diff --git a/Extensions/AwsS3/Cosmos.DataTransfer.AwsS3Storage/AwsS3DataSource.cs b/Extensions/AwsS3/Cosmos.DataTransfer.AwsS3Storage/AwsS3DataSource.cs
@@ -12,7 +12,7 @@ public class AwsS3DataSource : IComposableDataSource
         var settings = config.Get<AwsS3SourceSettings>();
         settings.Validate();
 
-        logger.LogInformation("Reading file {File} from AWS S3 Bucket '{BucketName}'", settings.FileName, settings.S3BucketName);
+        logger.LogInformation("Reading file {File} from AWS S3 Bucket '{BucketName}'", settings!.FileName, settings.S3BucketName);
 
         using var s3 = new S3Client(settings.S3AccessKey, settings.S3SecretKey, settings.S3Region);
         var stream = await s3.ReadFromS3(settings.S3BucketName, settings.FileName, cancellationToken);
diff --git a/Extensions/AzureBlob/Cosmos.DataTransfer.AzureBlobStorage/AzureBlobDataSink.cs b/Extensions/AzureBlob/Cosmos.DataTransfer.AzureBlobStorage/AzureBlobDataSink.cs
@@ -23,7 +23,7 @@ public async Task WriteToTargetAsync(Func<Stream, Task> writeToStream, IConfigur
             settings.Validate();
 
             BlobContainerClient account;
-            if (settings.UseRbacAuth)
+            if (settings!.UseRbacAuth)
             {
                 logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}'", settings.AccountEndpoint, nameof(AzureBlobSourceSettings.UseRbacAuth), nameof(AzureBlobSourceSettings.EnableInteractiveCredentials));
 
diff --git a/Extensions/AzureBlob/Cosmos.DataTransfer.AzureBlobStorage/AzureBlobDataSource.cs b/Extensions/AzureBlob/Cosmos.DataTransfer.AzureBlobStorage/AzureBlobDataSource.cs
@@ -17,7 +17,7 @@ public class AzureBlobDataSource : IComposableDataSource
         settings.Validate();
 
         BlobContainerClient account;
-        if (settings.UseRbacAuth)
+        if (settings!.UseRbacAuth)
         {
             logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}'", settings.AccountEndpoint, nameof(AzureBlobSourceSettings.UseRbacAuth), nameof(AzureBlobSourceSettings.EnableInteractiveCredentials));
 
diff --git a/Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSinkExtension.cs b/Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSinkExtension.cs
@@ -25,7 +25,7 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
 
             TableServiceClient serviceClient;
 
-            if (settings.UseRbacAuth)
+            if (settings!.UseRbacAuth)
             {
                 logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}", settings.AccountEndpoint, nameof(AzureTableAPIDataSinkSettings.UseRbacAuth), nameof(AzureTableAPIDataSinkSettings.EnableInteractiveCredentials));
 
diff --git a/Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSourceExtension.cs b/Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSourceExtension.cs
@@ -23,7 +23,7 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge
 
             TableServiceClient serviceClient;
 
-            if (settings.UseRbacAuth)
+            if (settings!.UseRbacAuth)
             {
                 logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}'", settings.AccountEndpoint, nameof(AzureTableAPIDataSourceSettings.UseRbacAuth), nameof(AzureTableAPIDataSourceSettings.EnableInteractiveCredentials));
 
diff --git a/Extensions/CognitiveSearch/Cosmos.DataTransfer.CognitiveSearchExtension/CognitiveSearchDataSinkExtension.cs b/Extensions/CognitiveSearch/Cosmos.DataTransfer.CognitiveSearchExtension/CognitiveSearchDataSinkExtension.cs
@@ -22,8 +22,8 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
             var settings = config.Get<CognitiveSearchDataSinkSettings>();
             settings.Validate();
 
-            var indexClient = new SearchIndexClient(new Uri(settings.Endpoint!), new AzureKeyCredential(settings.ApiKey!));
-            var searchClient = indexClient.GetSearchClient(settings.Index);
+            var indexClient = new SearchIndexClient(new Uri(settings!.Endpoint!), new AzureKeyCredential(settings.ApiKey!));
+            var searchClient = indexClient.GetSearchClient(settings.Index!);
 
             var convertedObjects = dataItems.Select(di => BuildObject(di)).Where(o => o != null).OfType<ExpandoObject>();
             var batches = convertedObjects.Buffer(settings.BatchSize);
diff --git a/Extensions/CognitiveSearch/Cosmos.DataTransfer.CognitiveSearchExtension/CognitiveSearchDataSourceExtension.cs b/Extensions/CognitiveSearch/Cosmos.DataTransfer.CognitiveSearchExtension/CognitiveSearchDataSourceExtension.cs
@@ -20,8 +20,8 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge
             var settings = config.Get<CognitiveSearchDataSourceSettings>();
             settings.Validate();
 
-            var indexClient = new SearchIndexClient(new Uri(settings.Endpoint!), new AzureKeyCredential(settings.ApiKey!));
-            var searchClient = indexClient.GetSearchClient(settings.Index);
+            var indexClient = new SearchIndexClient(new Uri(settings!.Endpoint!), new AzureKeyCredential(settings.ApiKey!));
+            var searchClient = indexClient.GetSearchClient(settings.Index!);
 
             var response = await searchClient.SearchAsync<JsonElement>("*"
                 , new Azure.Search.Documents.SearchOptions()
diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs
@@ -139,10 +139,10 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
             var settings = config.Get<CosmosSinkSettings>();
             settings.Validate();
 
-            var client = CosmosExtensionServices.CreateClient(settings, DisplayName, dataSource.DisplayName);
+            var client = CosmosExtensionServices.CreateClient(settings!, DisplayName, dataSource.DisplayName);
 
             Container container;
-            if (settings.UseRbacAuth)
+            if (settings!.UseRbacAuth)
             {
                 var cosmosContainer = client.GetContainer(settings.Database, settings.Container);
                 container = settings.InitClientEncryption
diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSourceExtension.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSourceExtension.cs
@@ -18,11 +18,11 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge
             var settings = config.Get<CosmosSourceSettings>();
             settings.Validate();
 
-            var client = CosmosExtensionServices.CreateClient(settings, DisplayName);
+            var client = CosmosExtensionServices.CreateClient(settings!, DisplayName);
 
             Container container;
             
-            if(settings.InitClientEncryption)
+            if(settings!.InitClientEncryption)
             {
                 container = await client.GetContainer(settings.Database, settings.Container).InitializeEncryptionAsync(cancellationToken);
             }
diff --git a/Extensions/Csv/Cosmos.DataTransfer.CsvExtension/CsvFormatReader.cs b/Extensions/Csv/Cosmos.DataTransfer.CsvExtension/CsvFormatReader.cs
@@ -31,7 +31,7 @@ public async IAsyncEnumerable<IDataItem> ParseDataAsync(IComposableDataSource so
             using var textReader = new StreamReader(source, leaveOpen: true);
             using var reader = new CsvReader(textReader, new CsvConfiguration(CultureInfo.InvariantCulture)
             {
-                HasHeaderRecord = settings.HasHeader,
+                HasHeaderRecord = settings!.HasHeader,
                 Delimiter = settings.Delimiter,
             });
 
diff --git a/Extensions/Mongo/Cosmos.DataTransfer.MongoExtension/MongoDataSinkExtension.cs b/Extensions/Mongo/Cosmos.DataTransfer.MongoExtension/MongoDataSinkExtension.cs
@@ -16,10 +16,10 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
         var settings = config.Get<MongoSinkSettings>();
         settings.Validate();
 
-        if (!string.IsNullOrEmpty(settings.ConnectionString) && !string.IsNullOrEmpty(settings.DatabaseName) && !string.IsNullOrEmpty(settings.Collection))
+        if (!string.IsNullOrEmpty(settings!.ConnectionString) && !string.IsNullOrEmpty(settings.DatabaseName) && !string.IsNullOrEmpty(settings.Collection))
         {
-            var context = new Context(settings.ConnectionString, settings.DatabaseName);
-            var repo = context.GetRepository<BsonDocument>(settings.Collection);
+            var context = new Context(settings.ConnectionString!, settings.DatabaseName!);
+            var repo = context.GetRepository<BsonDocument>(settings.Collection!);
 
             var batchSize = settings.BatchSize ?? 1000;
 
diff --git a/Extensions/Mongo/Cosmos.DataTransfer.MongoExtension/MongoDataSourceExtension.cs b/Extensions/Mongo/Cosmos.DataTransfer.MongoExtension/MongoDataSourceExtension.cs
@@ -18,9 +18,9 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge
         var settings = config.Get<MongoSourceSettings>();
         settings.Validate();
 
-        if (!string.IsNullOrEmpty(settings.ConnectionString) && !string.IsNullOrEmpty(settings.DatabaseName))
+        if (!string.IsNullOrEmpty(settings!.ConnectionString) && !string.IsNullOrEmpty(settings.DatabaseName))
         {
-            var context = new Context(settings.ConnectionString, settings.DatabaseName, settings.KeyVaultNamespace, settings.KMSProviders);
+            var context = new Context(settings.ConnectionString!, settings.DatabaseName!, settings.KeyVaultNamespace, settings.KMSProviders);
 
             var collectionNames = !string.IsNullOrEmpty(settings.Collection)
                 ? new List<string> { settings.Collection }
diff --git a/Extensions/Mongo/Cosmos.DataTransfer.MongoVectorExtension/MongoVectorDataSinkExtension.cs b/Extensions/Mongo/Cosmos.DataTransfer.MongoVectorExtension/MongoVectorDataSinkExtension.cs
@@ -19,7 +19,7 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
         var settings = config.Get<MongoVectorSinkSettings>();        
         settings.Validate();
 
-        if (!string.IsNullOrEmpty(settings.ConnectionString) && !string.IsNullOrEmpty(settings.DatabaseName) && !string.IsNullOrEmpty(settings.Collection))
+        if (!string.IsNullOrEmpty(settings!.ConnectionString) && !string.IsNullOrEmpty(settings.DatabaseName) && !string.IsNullOrEmpty(settings.Collection))
         {
             var Isembeddingsetsvalid = false;
             var client = new OpenAIClient("");
diff --git a/Extensions/PostgreSQL/PostgresqlDataSinkExtension.cs b/Extensions/PostgreSQL/PostgresqlDataSinkExtension.cs
@@ -24,7 +24,7 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
             settings.Validate();
             
             var cols = await FindPostgreDataTypes(dataItems, cancellationToken);
-            NpgsqlConnection con = new(settings.ConnectionString);
+            NpgsqlConnection con = new(settings!.ConnectionString);
 
             if (settings.AppendDataToTable == true && !string.IsNullOrEmpty(settings.TableName))
             {
diff --git a/Extensions/PostgreSQL/PostgresqlDataSourceExtension.cs b/Extensions/PostgreSQL/PostgresqlDataSourceExtension.cs
@@ -24,7 +24,7 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge
         var settings = config.Get<PostgreSourceSettings>();
         settings.Validate();
 
-        await using var connection = new NpgsqlConnection(settings.ConnectionString);
+        await using var connection = new NpgsqlConnection(settings!.ConnectionString);
         await connection.OpenAsync(cancellationToken);
         await using var command = new NpgsqlCommand(settings.QueryText, connection);
         await using var reader = await command.ExecuteReaderAsync(cancellationToken);
diff --git a/Extensions/SqlServer/Cosmos.DataTransfer.SqlServerExtension/SqlServerDataSourceExtension.cs b/Extensions/SqlServer/Cosmos.DataTransfer.SqlServerExtension/SqlServerDataSourceExtension.cs
@@ -48,7 +48,7 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(
                 settings.Validate();
 
                 await connection.OpenAsync(cancellationToken);
-                var command = connection.CreateCommand();
+                var command = connection.CreateCommand()!;
                 command.CommandText = queryText;
                 command.Parameters.AddRange(parameters);
 
diff --git a/Interfaces/Cosmos.DataTransfer.Common/FileDataSink.cs b/Interfaces/Cosmos.DataTransfer.Common/FileDataSink.cs
@@ -11,12 +11,10 @@ public async Task WriteToTargetAsync(Func<Stream, Task> writeToStream, IConfigur
     {
         var settings = config.Get<FileSinkSettings>();
         settings.Validate();
-        if (settings.FilePath != null)
-        {
-            using var writer = GetCompressor(settings.Compression, settings.FilePath, settings.Append);
-            await writeToStream(writer);
-            writer.Close();
-        }
+        
+        using var writer = GetCompressor(settings!.Compression, settings.FilePath!, settings.Append);
+        await writeToStream(writer);
+        writer.Close();
     }
 
     public IEnumerable<IDataExtensionSettings> GetSettings()
diff --git a/Interfaces/Cosmos.DataTransfer.Common/FileDataSource.cs b/Interfaces/Cosmos.DataTransfer.Common/FileDataSource.cs
@@ -32,48 +32,45 @@ public class FileDataSource : IComposableDataSource
     {
         var settings = config.Get<FileSourceSettings>();
         settings.Validate();
-        if (settings.FilePath != null)
-        {
 
-            if (File.Exists(settings.FilePath))
-            {
-                yield return ReadFile(settings.FilePath, settings.Compression, logger);
-            }
-            else if (Directory.Exists(settings.FilePath))
-            {
-                string[] patterns = new string[] { "*.json", "*.json.gz", "*.json.gzip", "*.json.br", "*.json.zz" };
-                var files = patterns.SelectMany(pattern =>  
-                    Directory.GetFiles(settings.FilePath, pattern, SearchOption.AllDirectories));
-                logger.LogInformation("Reading {FileCount} files from '{Folder}'", files.Count(), settings.FilePath);
-                foreach (string filePath in files.OrderBy(f => f))
-                {
-                    yield return ReadFile(filePath, settings.Compression, logger);
-                }
-            }
-            else if (Uri.IsWellFormedUriString(settings.FilePath, UriKind.Absolute))
+        if (File.Exists(settings!.FilePath!))
+        {
+            yield return ReadFile(settings.FilePath, settings.Compression, logger);
+        }
+        else if (Directory.Exists(settings.FilePath))
+        {
+            string[] patterns = new string[] { "*.json", "*.json.gz", "*.json.gzip", "*.json.br", "*.json.zz" };
+            var files = patterns.SelectMany(pattern =>  
+                Directory.GetFiles(settings.FilePath, pattern, SearchOption.AllDirectories));
+            logger.LogInformation("Reading {FileCount} files from '{Folder}'", files.Count(), settings.FilePath);
+            foreach (string filePath in files.OrderBy(f => f))
             {
-                logger.LogInformation("Reading from URI '{FilePath}'", settings.FilePath);
-                HttpClientHandler handler = new HttpClientHandler() { AutomaticDecompression = System.Net.DecompressionMethods.All };
-                HttpClient client = new HttpClient(handler);
-                var response = await client.GetAsync(settings.FilePath, cancellationToken);
-                if (!response.IsSuccessStatusCode)
-                {
-                    logger.LogError("Failed to read {FilePath}. Response was: {ResponseCode} {ResponseMessage}", settings.FilePath, response.StatusCode, response.ReasonPhrase);
-                    yield break;
-                }
-
-                var json = await response.Content.ReadAsStreamAsync(cancellationToken);
-
-                yield return json;
+                yield return ReadFile(filePath, settings.Compression, logger);
             }
-            else
+        }
+        else if (Uri.IsWellFormedUriString(settings.FilePath, UriKind.Absolute))
+        {
+            logger.LogInformation("Reading from URI '{FilePath}'", settings.FilePath);
+            HttpClientHandler handler = new HttpClientHandler() { AutomaticDecompression = System.Net.DecompressionMethods.All };
+            HttpClient client = new HttpClient(handler);
+            var response = await client.GetAsync(settings.FilePath, cancellationToken);
+            if (!response.IsSuccessStatusCode)
             {
-                logger.LogWarning("No content was found at configured path '{FilePath}'", settings.FilePath);
+                logger.LogError("Failed to read {FilePath}. Response was: {ResponseCode} {ResponseMessage}", settings.FilePath, response.StatusCode, response.ReasonPhrase);
                 yield break;
             }
 
-            logger.LogInformation("Completed reading '{FilePath}'", settings.FilePath);
+            var json = await response.Content.ReadAsStreamAsync(cancellationToken);
+
+            yield return json;
+        }
+        else
+        {
+            logger.LogWarning("No content was found at configured path '{FilePath}'", settings.FilePath);
+            yield break;
         }
+
+        logger.LogInformation("Completed reading '{FilePath}'", settings.FilePath);
     }
 
     public IEnumerable<IDataExtensionSettings> GetSettings()

Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ public async Task WriteToTargetAsync(Func<Stream, Task> writeToStream, IConfigur`
`23`	`23`	`settings.Validate();`
`24`	`24`
`25`	`25`	`BlobContainerClient account;`
`26`		`- if (settings.UseRbacAuth)`
	`26`	`+ if (settings!.UseRbacAuth)`
`27`	`27`	`{`
`28`	`28`	`logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}'", settings.AccountEndpoint, nameof(AzureBlobSourceSettings.UseRbacAuth), nameof(AzureBlobSourceSettings.EnableInteractiveCredentials));`
`29`	`29`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ public class AzureBlobDataSource : IComposableDataSource`
`17`	`17`	`settings.Validate();`
`18`	`18`
`19`	`19`	`BlobContainerClient account;`
`20`		`- if (settings.UseRbacAuth)`
	`20`	`+ if (settings!.UseRbacAuth)`
`21`	`21`	`{`
`22`	`22`	`logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}'", settings.AccountEndpoint, nameof(AzureBlobSourceSettings.UseRbacAuth), nameof(AzureBlobSourceSettings.EnableInteractiveCredentials));`
`23`	`23`
Original file line number	Diff line number	Diff line change
`@@ -25,7 +25,7 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati`
`25`	`25`
`26`	`26`	`TableServiceClient serviceClient;`
`27`	`27`
`28`		`- if (settings.UseRbacAuth)`
	`28`	`+ if (settings!.UseRbacAuth)`
`29`	`29`	`{`
`30`	`30`	`logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}", settings.AccountEndpoint, nameof(AzureTableAPIDataSinkSettings.UseRbacAuth), nameof(AzureTableAPIDataSinkSettings.EnableInteractiveCredentials));`
`31`	`31`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge`
`23`	`23`
`24`	`24`	`TableServiceClient serviceClient;`
`25`	`25`
`26`		`- if (settings.UseRbacAuth)`
	`26`	`+ if (settings!.UseRbacAuth)`
`27`	`27`	`{`
`28`	`28`	`logger.LogInformation("Connecting to Storage account {AccountEndpoint} using {UseRbacAuth} with {EnableInteractiveCredentials}'", settings.AccountEndpoint, nameof(AzureTableAPIDataSourceSettings.UseRbacAuth), nameof(AzureTableAPIDataSourceSettings.EnableInteractiveCredentials));`
`29`	`29`
Original file line number	Diff line number	Diff line change
`@@ -18,11 +18,11 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge`
`18`	`18`	`var settings = config.Get<CosmosSourceSettings>();`
`19`	`19`	`settings.Validate();`
`20`	`20`
`21`		`- var client = CosmosExtensionServices.CreateClient(settings, DisplayName);`
	`21`	`+ var client = CosmosExtensionServices.CreateClient(settings!, DisplayName);`
`22`	`22`
`23`	`23`	`Container container;`
`24`	`24`
`25`		`- if(settings.InitClientEncryption)`
	`25`	`+ if(settings!.InitClientEncryption)`
`26`	`26`	`{`
`27`	`27`	`container = await client.GetContainer(settings.Database, settings.Container).InitializeEncryptionAsync(cancellationToken);`
`28`	`28`	`}`