Merge pull request #19 from FizzcodeSoftware/dev

Dev

Author: wickedmachinator

EtLast.AdoNet/AdoNetDbReader/AbstractAdoNetDbReader.cs

@@ -36,6 +36,12 @@ public abstract class AbstractAdoNetDbReader : AbstractRowSource
     /// </summary>
     public bool InlineArrayParameters { get; init; } = true;
 
+    /// <summary>
+    /// If initialized with a dictionary, then all the columns returned by the ADO.NET connector based on the given query will be stored in it.
+    /// Key is the column name in the produced row, value is the exact data type of the field.
+    /// </summary>
+    public Dictionary<string, Type> ColumnTypeMap = null;
+
     protected abstract CommandType GetCommandType();
 
     protected AbstractAdoNetDbReader()
@@ -155,6 +161,9 @@ protected override IEnumerable<IRow> Produce()
                         Config = null,
                     };
                 }
+
+                if (ColumnTypeMap != null && columns[i] != null)
+                    ColumnTypeMap[columns[i].NameInRow] = reader.GetFieldType(i);
             }
 
             while (!FlowState.IsTerminating)

EtLast.AdoNet/Jobs/AbstractSqlStatementWithResult.cs

@@ -3,6 +3,7 @@
 [EditorBrowsable(EditorBrowsableState.Never)]
 public abstract class AbstractSqlStatementWithResult<T> : AbstractProcessWithResult<T>
 {
+    [ProcessParameterMustHaveValue]
     public required NamedConnectionString ConnectionString { get; set; }
 
     /// <summary>
@@ -22,8 +23,6 @@ protected AbstractSqlStatementWithResult()
 
     protected override void ValidateImpl()
     {
-        if (ConnectionString == null)
-            throw new ProcessParameterNullException(this, nameof(ConnectionString));
     }
 
     protected sealed override T ExecuteImpl()

EtLast/Exceptions/JsonDeserializerException.cs

@@ -0,0 +1,8 @@
+namespace FizzCode.EtLast;
+
+[ComVisible(true)]
+[Serializable]
+public class JsonDeserializerException(IProcess process, string message, Exception innerException)
+    : EtlException(process, message, innerException)
+{
+}

EtLast/Exceptions/JsonSerializerException.cs

@@ -0,0 +1,8 @@
+namespace FizzCode.EtLast;
+
+[ComVisible(true)]
+[Serializable]
+public class JsonSerializerException(IProcess process, string message, Exception innerException)
+    : EtlException(process, message, innerException)
+{
+}

EtLast/Processes/Json/DeserializeFromJsonFileJob.cs

@@ -0,0 +1,71 @@
+namespace FizzCode.EtLast;
+
+public sealed class DeserializeFromJsonFileJob<T> : AbstractProcessWithResult<IEnumerable<T>>
+{
+    [ProcessParameterMustHaveValue]
+    public required IStreamProvider StreamProvider { get; init; }
+
+    public Encoding Encoding { get; init; } = Encoding.UTF8;
+
+    public JsonSerializerOptions SerializerOptions { get; init; } = new JsonSerializerOptions()
+    {
+        WriteIndented = true,
+    };
+
+    protected override void ValidateImpl()
+    {
+    }
+
+    protected override IEnumerable<T> ExecuteImpl()
+    {
+        var streams = StreamProvider.GetStreams(this);
+        if (streams == null)
+            yield break;
+
+        var streamIndex = 0;
+        foreach (var stream in streams)
+        {
+            T result = default;
+
+            try
+            {
+                using (var reader = new StreamReader(stream.Stream, Encoding))
+                {
+                    var content = reader.ReadToEnd();
+                    result = JsonSerializer.Deserialize<T>(content, SerializerOptions);
+                }
+
+                stream.IoCommand.AffectedDataCount += 1;
+                stream.IoCommand.End();
+            }
+            catch (Exception ex)
+            {
+                var exception = new JsonDeserializerException(this, "error while deserializing a json file", ex);
+                exception.AddOpsMessage(string.Format(CultureInfo.InvariantCulture, "error while deserializing a json file: {0}", stream.Name));
+                exception.Data["StreamName"] = stream.Name;
+                exception.Data["StreamIndex"] = streamIndex;
+
+                stream.IoCommand.Failed(exception);
+                throw exception;
+            }
+            finally
+            {
+                stream.Stream.Flush();
+                stream.Stream.Close();
+                stream.Stream.Dispose();
+            }
+
+            yield return result;
+            streamIndex++;
+        }
+    }
+}
+
+[EditorBrowsable(EditorBrowsableState.Never)]
+public static class DeserializeFromJsonFileJobFluent
+{
+    public static IFlow DeserializeFromJsonFile<TResult>(this IFlow builder, out IEnumerable<TResult> result, Func<DeserializeFromJsonFileJob<TResult>> processCreator)
+    {
+        return builder.ExecuteProcessWithResult(out result, processCreator);
+    }
+}

EtLast/Processes/Json/SerializeToJsonFileJob.cs

@@ -24,6 +24,18 @@ protected override void ExecuteImpl(Stopwatch netTimeStopwatch)
 
             namedSink.Stream.Write(Encoding.GetBytes(content));
             namedSink.IncreaseRowsWritten();
+
+            namedSink.IoCommand.AffectedDataCount += 1;
+            namedSink.IoCommand.End();
+        }
+        catch (Exception ex)
+        {
+            var exception = new JsonSerializerException(this, "error while serializing into a json file", ex);
+            exception.AddOpsMessage(string.Format(CultureInfo.InvariantCulture, "error while serializing into a json file: {0}", namedSink.Name));
+            exception.Data["StreamName"] = namedSink.Name;
+
+            namedSink.IoCommand.Failed(exception);
+            throw exception;
         }
         finally
         {
@@ -32,4 +44,13 @@ protected override void ExecuteImpl(Stopwatch netTimeStopwatch)
             namedSink.Stream.Dispose();
         }
     }
+}
+
+[EditorBrowsable(EditorBrowsableState.Never)]
+public static class SerializeToJsonFileJobFluent
+{
+    public static IFlow SerializeToJsonFile<T>(this IFlow builder, Func<SerializeToJsonFileJob<T>> processCreator)
+    {
+        return builder.ExecuteProcess(processCreator);
+    }
 }

Samples/Sample.Etl.Modules/Modules/SampleModule/Startup.cs

@@ -11,6 +11,4 @@ public void BuildSession(ISessionBuilder session, IArgumentCollection arguments)
             .LogOpsToFile()
             .LogIoToFile();
     }
-
-    public Dictionary<string, Func<IArgumentCollection, IEtlTask>> CustomTasks => [];
 }