diff --git a/CHANGELOG.md b/CHANGELOG.md index fd0b188..5c4c525 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,9 @@ Framework `datapipeline` (Gem). Neueste Version oben. kaufmännisch gerundet; `DAT`/`TIM` aus `Date`, `Time` und ISO-Text) - neues Plugin `:fixed` (`DataPipeline::FixedStorage`): Festbreiten-Dateien nach Layout schreiben und lesen - csv: mit `layout:` und `headers: false` werden Dateien mit Satzarten gelesen, Schlüssel aus den Feldnamen +- DSL: Kurzschritte `rename`, `select`, `cast` (`:integer`, `:decimal`, `:date`, `:string`) und `default` +- veraltet, entfällt in 3.0: `transform`, der Parallelmodus und `condition:` als String; je Prozess einmal + `WARN deprecated: …` im Log ## 2.0.0 – 2026-10-08 diff --git a/README.md b/README.md index 3bc8ffc..0f770a1 100644 --- a/README.md +++ b/README.md @@ -601,9 +601,13 @@ flowchart LR | `version(str)`, `verbose(flag = true)`, `streamed(flag = true)`, `chained(flag = true)` | dasselbe als DSL-Aufruf | | `container(:typ, **opts)` | erzeugt das registrierte Plugin, sonst `Storage.new(**opts)` (z. B. `:postgres` → `DataPipeline::PostgresStorage`); fehlt es: „Unsupported storage type“; fehlende Gems, Verbindungs- und Optionsfehler kommen unverändert durch | | `read(storage, opts = {})` | legt die Quelle fest; `opts` gehen an `storage.set` | -| `transform(opts = {}) { \|row, index\| ... }` | Transformation; `processing: :prepend \| :normal \| :append`, `condition:` (siehe 2.3) | +| `transform(opts = {}) { \|row, index\| ... }` | **veraltet**, entfällt in 3.0 (siehe 2.3): Transformation; `processing: :prepend \| :normal \| :append`, `condition:` | | `map(opts = {}) { \|row, ctx\| ... }` | Schritt in der Kette: Hash, Array oder `nil` zurückgeben; schaltet Verkettung ein | | `filter(opts = {}) { \|row, ctx\| ... }` | Schritt in der Kette: Zeile bleibt, wenn der Block wahr ist; schaltet Verkettung ein | +| `rename 'kunnr' => 'kundennr'` | Kurzschritt: Schlüssel umbenennen, Position bleibt | +| `select %w[belnr betrag]` | Kurzschritt: nur diese Schlüssel in dieser Reihenfolge, fehlende als `nil` | +| `cast 'betrag' => :decimal` | Kurzschritt: umwandeln in `:integer`, `:decimal` (Float), `:date`, `:string`; leere Werte werden `nil`, ein Fehler läuft über `on_error` | +| `default 'land' => 'CH'` | Kurzschritt: Wert für `nil` oder leeren Text; ein Lambda bekommt die Zeile | | `header { \|ctx\| ... }` | Zeile(n) vor den Daten; wird nach den Daten berechnet, kennt also `ctx.written` | | `trailer { \|ctx\| ... }` | Zeile(n) nach den Daten | | `query(storage, query: sql, params: [])` | führt sofort eine Abfrage aus (Subquery in einem Schritt) | @@ -623,6 +627,12 @@ flowchart LR ### 2.3 Ausführungsmodell (`run`) +> **Veraltet ab 2.1, entfällt in 3.0:** `transform`, der Parallelmodus (mehrere `transform` ohne `chained`) und +> `condition:` als String (`eval`). Ab 3.0 laufen alle Schritte verkettet. Wer sie nutzt, bekommt je Prozess einmal +> eine Warnung im Log (`WARN deprecated: …`). Umstellung: `transform` → `map` (Block bekommt `ctx` statt `index`, +> `ctx.index`), `processing: :prepend`/`:append` → `header`/`trailer`, `condition: "row['a'] == 1"` → +> `condition: ->(row) { row['a'] == 1 }`. Das bisherige Verhalten beschreibt der Rest dieses Abschnitts. + 1. Batch (Default): `data = source.read` lädt die gesamte Quelle in den Speicher. Streaming: siehe unten. 2. Die Schritte werden in drei Gruppen geteilt, innerhalb jeder Gruppe gilt die Reihenfolge der Registrierung: **vorne** (`:prepend`, `header`), **Daten** (`:normal`, `map`, `filter`), **hinten** (`:append`, `trailer`). @@ -811,6 +821,16 @@ CSV- oder JSON-Pipeline lädt keines dieser Gems. Wer ein Plugin nicht braucht, | `:skip` | Zeile fällt weg, Warnung im Log, Zähler `rejected` | | `:collect, to: storage` | wie `:skip`, zusätzlich die Eingangszeile mit `_step` und `_error` per `write_row` ins Ziel `storage` | +**Kurzschritte** – Häufige kleine `map`-Blöcke gibt es als eigene Schritte. Sie schalten wie `map` die Verkettung ein +und heissen in Fehlermeldungen und der Fehlerdatei nach sich selbst (`cast#3`). + +```ruby +rename 'kunnr' => 'kundennr', 'name1' => 'name' +default 'land' => 'CH', 'erfasst' => ->(row) { row['geaendert'] } +cast 'betrag' => :decimal, 'datum' => :date, 'menge' => :integer +select %w[kundennr name land betrag datum] +``` + **Validierung** – `validate` prüft Spalten nach Regeln, nach einem Layout (2.9) und/oder mit einem Block. Eine verletzte Regel löst `DataPipeline::ValidationError` aus und läuft damit über `on_error`. Leere Werte prüft nur `required`. Regeln je Spalte: `required: true`, `max:`/`min:` (Länge als Text), `format:` (Regex auf den Text), diff --git a/lib/datapipeline.rb b/lib/datapipeline.rb index 86bd848..59185de 100644 --- a/lib/datapipeline.rb +++ b/lib/datapipeline.rb @@ -51,6 +51,18 @@ class DataPipeline ON_ERROR = %i[abort skip collect].freeze MAX_ERRORS = 100 + # removed in 3.0, logged once per process as a warning + DEPRECATIONS = { + transform: 'transform is deprecated and will be removed in 3.0: use map/filter, ' \ + 'header/trailer instead of processing: :prepend/:append', + parallel: 'parallel mode (several transform steps without chained) is deprecated and will be removed in 3.0: ' \ + 'set chained or use map', + condition: 'condition: as string (eval) is deprecated and will be removed in 3.0: use a lambda ->(row) { ... }' + }.freeze + + # value conversions for cast, empty values become nil + CASTS = Validation::TYPES.merge(string: ->(v) { v.to_s }).freeze + DEFAULT_SETTINGS = { version: nil, streamed: false, @@ -67,6 +79,7 @@ class DataPipeline SETTINGS_LOCK = Mutex.new @defaults = DEFAULT_SETTINGS @plugins = {} + @deprecations_logged = [] # # defaults for every following DataPipeline.new, e.g. from bin/etl @@ -132,6 +145,18 @@ class DataPipeline @runs ||= [] end + # true the first time a deprecation is reported in this process + def self.deprecation_once key + SETTINGS_LOCK.synchronize do + next false if @deprecations_logged.include?(key) + + @deprecations_logged << key + true + end + end + + def self.reset_deprecations! = SETTINGS_LOCK.synchronize { @deprecations_logged = [] } + # DataPipeline.new(verbose: true) { ... } overrides the defaults for this pipeline only def initialize(**settings, &block) @settings = DataPipeline.settings.merge(settings) @@ -247,6 +272,46 @@ class DataPipeline add_step :filter, options, &block end + # + # short steps, each one a map step (turns chaining on) + # + # rename 'kunnr' => 'kundennr' -> new key at the same position + # select %w[belnr betrag] -> only these keys in this order, missing ones nil + # cast 'betrag' => :decimal -> :integer, :decimal (Float), :date, :string; empty -> nil + # default 'land' => 'CH' -> for nil or blank values; a lambda gets the row + # + def rename mapping + mapping = mapping.to_h { |from, to| [from.to_s, to.to_s] } + add_step(:map, { name: 'rename' }) { |row| row.to_h { |key, value| [mapping.fetch(key.to_s, key), value] } } + end + + def select *keys + keys = keys.flatten.map(&:to_s) + add_step(:map, { name: 'select' }) { |row| keys.to_h { |key| [key, value_of(row, key)] } } + end + + def cast types + types = types.to_h { |key, type| [key.to_s, type.to_sym] } + unknown = types.values - CASTS.keys + raise ArgumentError, "cast: #{unknown.join(', ')} unknown, use #{CASTS.keys.join(', ')}" if unknown.any? + + add_step(:map, { name: 'cast' }) do |row| + types.each_with_object(row.dup) do |(key, type), out| + value = value_of(row, key) + out[key] = blank?(value) ? nil : cast_value(key, value, type) + end + end + end + + def default values + values = values.to_h { |key, value| [key.to_s, value] } + add_step(:map, { name: 'default' }) do |row| + values.each_with_object(row.dup) do |(key, value), out| + out[key] = value.respond_to?(:call) ? value.call(row) : value if blank?(value_of(row, key)) + end + end + end + # header { |ctx| ... } -> rows in front of the body, called after the body with full counters def header(&block) add_step :header, {}, &block @@ -317,6 +382,7 @@ class DataPipeline before = @transformations.select { |t| EDGE_BEFORE.include?(t[:kind]) } after = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) } body = @transformations - before - after + log_deprecations body @settings[:streamed] ? run_streamed(before, body, after) : run_batch(before, body, after) @@ -338,6 +404,29 @@ class DataPipeline @transformations << { kind:, block:, options: } end + def value_of row, key + row.key?(key) ? row[key] : row[key.to_sym] + end + + def blank? value + value.nil? || (value.is_a?(String) && value.strip.empty?) + end + + def cast_value key, value, type + CASTS.fetch(type).call(value) + rescue ArgumentError, TypeError, Date::Error + raise ArgumentError, "#{key}: #{value.inspect} is not #{type}" + end + + # each deprecated feature once per process, with the logger of the pipeline + def log_deprecations body + used = [] + used << :transform if @transformations.any? { |t| %i[prepend normal append].include?(t[:kind]) } + used << :parallel if !chained? && body.size > 1 + used << :condition if @transformations.any? { |t| t[:options].is_a?(Hash) && t[:options][:condition].is_a?(String) } + used.each { |key| @logger.warn "deprecated: #{DEPRECATIONS[key]}" if DataPipeline.deprecation_once(key) } + end + # without any transformation the source is copied unchanged def chained? @settings[:chained] || @transformations.empty? || @@ -489,7 +578,8 @@ class DataPipeline end def step_name step - "#{step[:kind]}##{@transformations.index(step) + 1}" + name = step[:options].is_a?(Hash) && step[:options][:name] || step[:kind] + "#{name}##{@transformations.index(step) + 1}" end # rows produced by one step for one input row diff --git a/test/core/steps_test.rb b/test/core/steps_test.rb new file mode 100644 index 0000000..8c2285b --- /dev/null +++ b/test/core/steps_test.rb @@ -0,0 +1,76 @@ +require_relative '../test_helper' + +# rename, select, cast, default and the deprecation warnings +class StepsTest < PipelineTest + ROWS = [{ 'kunnr' => '1', 'betrag' => '12.50', 'datum' => '2026-10-09', 'land' => nil }, + { 'kunnr' => '2', 'betrag' => '', 'datum' => ' ', 'land' => 'DE' }].freeze + + def test_rename_keeps_the_position + assert_equal [%w[kundennr betrag datum land]] * 2, output(ROWS) { rename 'kunnr' => 'kundennr' }.map(&:keys) + assert_equal [{ 'b' => 1 }], output([{ a: 1 }]) { rename a: 'b' } + end + + def test_select_order_and_missing_keys + assert_equal [{ 'betrag' => '12.50', 'kunnr' => '1', 'x' => nil }], output(ROWS.first(1)) { select %w[betrag kunnr x] } + assert_equal [{ 'kunnr' => '1' }], output([{ kunnr: '1', b: 2 }]) { select 'kunnr' } + end + + def test_cast_types_and_empty_values + rows = output(ROWS) { cast 'betrag' => :decimal, 'datum' => :date, 'kunnr' => :integer } + assert_equal [1, 12.5, Date.new(2026, 10, 9)], rows[0].values_at('kunnr', 'betrag', 'datum') + assert_equal [2, nil, nil], rows[1].values_at('kunnr', 'betrag', 'datum') + assert_equal [{ 'a' => '5' }], output([{ 'a' => 5 }]) { cast 'a' => :string } + assert_includes assert_raises(ArgumentError) { output(ROWS) { cast 'a' => :money } }.message, 'money unknown' + end + + def test_cast_error_goes_to_on_error + target, stats = run_pipeline([{ 'n' => 'x' }, { 'n' => '3' }]) { on_error :skip; cast 'n' => :integer } + assert_equal [{ 'n' => 3 }], target.output + assert_equal [{ step: 'cast#1', error: 'n: "x" is not integer', row: { 'n' => 'x' } }], stats.errors + end + + def test_default_value_and_lambda + rows = output(ROWS) { default 'land' => 'CH', 'datum' => ->(row) { "neu #{row['kunnr']}" }, 'neu' => 0 } + assert_equal [['CH', '2026-10-09', 0], ['DE', 'neu 2', 0]], rows.map { |r| r.values_at('land', 'datum', 'neu') } + end + + def test_steps_chain_with_map_and_validate + rows = output(ROWS) do + rename 'kunnr' => 'id' + default 'betrag' => '0' + cast 'betrag' => :decimal + validate 'id' => { required: true } + select %w[id betrag] + end + assert_equal [{ 'id' => '1', 'betrag' => 12.5 }, { 'id' => '2', 'betrag' => 0.0 }], rows + end + + def deprecation_log(&block) + log = StringIO.new + DataPipeline.new do + logger Logger.new(log) + read MemoryStorage.new(rows: [{ 'a' => 1 }]) + write MemoryStorage.new + instance_eval(&block) + end.run + log.string.lines.grep(/deprecated/) + end + + def test_deprecations_once_per_process + lines = deprecation_log do + transform(condition: 'true') { |r| r } + transform { |r| r } + end + assert_equal 3, lines.size + assert lines.any? { |l| l.include?('transform is deprecated') } + assert lines.any? { |l| l.include?('parallel mode') } + assert lines.any? { |l| l.include?('condition: as string') } + assert_empty deprecation_log { transform { |r| r } } + end + + def test_no_deprecation_for_map_filter_and_single_transform_in_chained_mode + assert_empty deprecation_log { map { |r| r }; filter(condition: ->(_r) { true }) { true } } + lines = deprecation_log { chained; transform { |r| r }; transform { |r| r } } + assert_equal 1, lines.size # only transform, no parallel mode + end +end diff --git a/test/test_helper.rb b/test/test_helper.rb index 175d92c..eb5c3d2 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -55,6 +55,7 @@ module PipelineHelper DataPipeline.reset_settings! DataPipeline.runs.clear DataPipeline.params.clear + DataPipeline.reset_deprecations! end # runs a pipeline from rows into a MemoryStorage, returns [target, stats]