From baf09d39474c67141a18523b720901235610f61b Mon Sep 17 00:00:00 2001 From: Wild Karl-Heinz Date: Fri, 9 Oct 2026 21:46:50 +0200 Subject: [PATCH] Betrieb: Verbindungsdatei, Summen mit Abgleich vor dem Schreiben, Laufhistorie --- CHANGELOG.md | 6 + README.md | 70 ++++++++- exe/etl | 8 +- lib/datapipeline.rb | 146 ++++++++++++++++++- lib/datapipeline/connections.rb | 59 ++++++++ test/cli/etl_cli_test.rb | 15 ++ test/core/connections_totals_history_test.rb | 97 ++++++++++++ 7 files changed, 390 insertions(+), 11 deletions(-) create mode 100644 lib/datapipeline/connections.rb create mode 100644 test/core/connections_totals_history_test.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c4c525..8a1f8fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,12 @@ Framework `datapipeline` (Gem). Neueste Version oben. - 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 +- Verbindungen in `config/connections.yml` (mit ERB für ENV): `connection(:name, **opts)` und + `container(:typ, connection: :name)` +- `totals 'spalte'` (exakte Summen über Quelle und Ergebnis, `stats.total`) und `reconcile { |stats| … }`: Abgleich + vor dem Schreiben, eine Meldung bricht mit `DataPipeline::ReconcileError` ab +- Laufhistorie: `history 'datei'` bzw. `etl --history=datei`, eine JSON-Zeile je Lauf; `DataPipeline.history` liest + sie; `name` des Laufs (über `etl` der Skriptname), `Stats#to_h` ## 2.0.0 – 2026-10-08 diff --git a/README.md b/README.md index 0f770a1..2de293d 100644 --- a/README.md +++ b/README.md @@ -575,6 +575,7 @@ flowchart LR | [lib/datapipeline/validation.rb](lib/datapipeline/validation.rb) | Regeln für `validate` | | [lib/datapipeline/watermark.rb](lib/datapipeline/watermark.rb) | Wasserzeichen für inkrementelle Läufe | | [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe | +| [lib/datapipeline/connections.rb](lib/datapipeline/connections.rb) | Verbindungsdatei `config/connections.yml` | | [lib/datapipeline/layout.rb](lib/datapipeline/layout.rb) | Satzarten-Layouts (Feldliste je Satzart, siehe 2.9) | | [exe/etl](exe/etl) | Kommando `etl` des Gems mit Exit-Codes (siehe 2.7); [bin/etl](bin/etl) startet es im Repository mit dem Bundle | | [datapipeline.gemspec](datapipeline.gemspec), [CHANGELOG.md](CHANGELOG.md) | Gem-Beschreibung, Versionsverlauf | @@ -616,6 +617,10 @@ flowchart LR | `validate(rules \| layout, warn: false, ignore: []) { \|row\| ... }` | Prüfschritt (siehe 2.7, 2.9); `warn: true` nur Warnung, Zeile bleibt; schaltet Verkettung ein | | `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Datenschritten (siehe 2.7) | | `on_finish { \|stats\| ... }` | nach jedem Lauf, auch nach Abbruch | +| `totals 'dmbtr', …` | Summen dieser Spalten über die Quellzeilen und über die Zeilen nach allen Schritten (siehe 2.7) | +| `reconcile { \|stats\| ... }` | Abgleich nach allen Schritten, vor dem Schreiben; eine Meldung bricht mit `DataPipeline::ReconcileError` ab | +| `name(str)`, `history(file)` | Name des Laufs; eine JSON-Zeile je Lauf in `file` (siehe 2.7) | +| `connection(:name, **opts)` | Storage aus der Verbindungsdatei (siehe 2.7); `container(:typ, connection: :name)` ebenso | | `watermark(name, column:, file:, initial:)` | letzter verarbeiteter Wert für inkrementelle Läufe (siehe 2.7) | | `lock(file)` | Sperre während `run`; läuft schon ein Lauf mit derselben Datei, bricht dieser mit `DataPipeline::LockError` ab (siehe 2.7) | | `dry_run(flag = true)`, `limit(n)` | nichts schreiben bzw. nur die ersten n Quellzeilen | @@ -859,8 +864,11 @@ validate { |row| 'Betrag 0' if row['dmbtr'].to_f.zero? } | `warnings` | Zeilen mit Warnung aus `validate …, warn: true` | | `duration`, `started_at`, `finished_at` | Laufzeit | | `error` | Exception eines abgebrochenen Laufs | +| `name` | Name aus `name`, über `etl` Default der Skriptname ohne `.rb` | +| `totals`, `total(spalte, :read \| :written)` | Summen aus `totals` als `Rational` (exakt) | +| `to_h` | alles als Hash, wie in der Laufhistorie | | `ok?` | kein Abbruch und nichts verworfen | -| `to_s` | `read=9 written=4 rejected=4 0.097s` | +| `to_s` | `read=9 written=4 rejected=4 0.097s`, mit `totals` zusätzlich `dmbtr=1500.25/1500.25` (Quelle/Ergebnis) | Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile (Warnung) und einen Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem Abbruch; ein Fehler im Hook wird nur geloggt. @@ -955,6 +963,61 @@ wm = watermark :rechnungen, column: 'rechnung_id', file: 'data/.state.json', ini read pg, query: 'select * from rechnung where rechnung_id > $1 order by rechnung_id', params: [wm.value] ``` +**Verbindungen** – Zugangsdaten stehen nicht in den Skripten, sondern einmal in `config/connections.yml` (oder der +Datei aus `DataPipeline.connections_file = …` bzw. `ENV['DATAPIPELINE_CONNECTIONS']`). Die Datei wird mit ERB +gelesen, Passwörter kommen so aus der Umgebung und die Datei kann ins Repository. `type` ist der Plugin-Typ, alle +anderen Schlüssel sind dessen Optionen; Optionen beim Aufruf haben Vorrang. + +```yaml +# config/connections.yml +pds: + type: postgres + host: <%= ENV.fetch('PDS_HOST', 'localhost') %> + database: pds + username: etl + password: <%= ENV['PDS_PASSWORD'] %> +``` + +```ruby +read connection(:pds, query: 'select * from rechnung where datum >= $1', params: [params[:from]]) +write container(:postgres, connection: :pds, table: 'export_log') # Typ muss zur Verbindung passen +``` + +Eine unbekannte Verbindung, eine fehlende Datei oder ein abweichender Typ bricht mit `ArgumentError` ab. Die Datei +wird je Prozess einmal gelesen. + +**Summen und Abgleich** – Bei einer Buchhaltungsschnittstelle reicht es nicht, dass der Lauf durchläuft: Die Summen +müssen stimmen. `totals` summiert Spalten über die Quellzeilen (`:read`) und über die Zeilen nach allen Schritten +(`:written`, ohne Kopf- und Schlusssätze), exakt als `Rational`. Leere und nicht numerische Werte zählen nicht mit. +`reconcile` läuft nach allen Schritten und **vor dem Schreiben**; gibt der Block eine Meldung zurück, bricht der Lauf +mit `DataPipeline::ReconcileError` ab und schreibt nichts (Exit-Code `1`, `on_finish` läuft). Im Streaming läuft +`reconcile` nach der letzten Zeile, die Zeilen sind dann schon geschrieben; mit Postgres und `batch_size: :all` werden +sie beim Abbruch zurückgerollt (2.5). + +```ruby +totals 'dmbtr' +reconcile do |stats| + quelle, ziel = stats.total('dmbtr', :read), stats.total('dmbtr') + "Summe dmbtr: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ziel)}" if quelle != ziel +end +reconcile { |stats| 'keine Buchungen' if stats.written.zero? } +``` + +`DataPipeline.decimal(wert)` gibt eine Summe als exakten Dezimaltext aus (`1500.25` statt `6001/4`). + +**Laufhistorie** – `history 'data/runs.ndjson'` (oder `etl … --history=data/runs.ndjson` für alle Pipelines eines +Skripts) hängt nach jedem Lauf eine JSON-Zeile an, auch nach einem Abbruch und im Dry-Run. Damit lässt sich später +beantworten, wann der letzte erfolgreiche Lauf war und wie viele Zeilen er hatte. Ein Fehler beim Schreiben der +Historie wird nur geloggt. + +```json +{"name":"cudo","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","finished_at":"2026-10-09T06:00:04.336+02:00","duration":4.213,"read":1250,"written":1243,"output":1245,"rejected":7,"warnings":0,"dry_run":false,"ok":false,"error":null,"totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}}} +``` + +```ruby +letzter = DataPipeline.history('data/runs.ndjson', name: 'cudo').reverse.find { |run| run['ok'] } +``` + **Sperre gegen parallele Läufe** – Startet ein Cronjob, während der vorige Lauf noch arbeitet, lesen beide dasselbe Wasserzeichen und exportieren doppelt. `lock` verhindert das mit einer exklusiven Sperre (`flock`) auf eine Datei: Ein zweiter Lauf wartet nicht, sondern bricht sofort mit `DataPipeline::LockError` ab (Exit-Code `3`), schreibt nichts @@ -988,6 +1051,7 @@ bin/etl run pipeline.rb --param land=DE # im Skript: params[:land] | `--param KEY=VALUE` | `params[:key]`, mehrfach möglich | | `--dry-run`, `--limit=N` | für alle Pipelines des Skripts | | `--lock=FILE` | Sperre für das ganze Skript; ist sie belegt, endet `etl` sofort mit Exit-Code `3` | +| `--history=FILE` | Laufhistorie für alle Pipelines des Skripts; als `name` gilt der Skriptname | | `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default über `bin/etl`: Info) | | jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` | @@ -996,7 +1060,7 @@ Exit-Code: `0` ok, `1` Abbruch, `2` fertig, aber mit verworfenen Zeilen, `3` ein ### 2.8 Tests -`rake test` (oder `bundle exec rake`) startet die Minitest-Suite in [test/](test/): rund 165 Fälle für Plugins, Kern, +`rake test` (oder `bundle exec rake`) startet die Minitest-Suite in [test/](test/): rund 205 Fälle für Plugins, Kern, Streaming, Betrieb, Kommandozeile und die Beispiele. | Ordner | Inhalt | @@ -1008,7 +1072,7 @@ Streaming, Betrieb, Kommandozeile und die Beispiele. Die `.env` wird in Tests nicht geladen; kein Test erreicht die Produktivdatenbank. Postgres-Tests legen je Test eine Wegwerf-Datenbank auf einem lokalen Server an (`ETL_TEST_PG_HOST`, `_PORT`, `_USER`, `_PASSWORD`; Default -`localhost`, `$USER`). Ohne erreichbaren Server werden sie übersprungen (21 Tests). Ein eigenes Plugin lässt sich mit +`localhost`, `$USER`). Ohne erreichbaren Server werden sie übersprungen (23 Tests). Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als Quelle oder Ziel testen; mit `include StorageContract` und einer Methode `contract_storage(env, :write | :read)` bekommt es die gemeinsamen Tests. diff --git a/exe/etl b/exe/etl index 5cac587..5455218 100755 --- a/exe/etl +++ b/exe/etl @@ -10,6 +10,7 @@ # --limit=N only the first N source rows # --verbose log level debug, --quiet only warnings (default info) # --lock=FILE exclusive lock for the whole script, exit 3 when another run holds it +# --history=FILE one json line per pipeline run (DataPipeline.history) # --name=value any other option: DataPipeline.params[:name] and passed on to the script # # exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run @@ -33,6 +34,7 @@ parser = OptionParser.new do |opts| opts.on('--verbose') { options[:settings][:log_level] = Logger::DEBUG } opts.on('--quiet') { options[:settings][:log_level] = Logger::WARN } opts.on('--lock=FILE', String) { |v| options[:lock] = v } + opts.on('--history=FILE', String) { |v| options[:settings][:history] = v } end # own options for the parser, everything else (also --name=value of the script) goes to the script @@ -40,9 +42,9 @@ own = [] args = [] argv = ARGV.dup while (arg = argv.shift) - if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock)(=|\z)/) + if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history)(=|\z)/) own << arg - own << argv.shift if %w[--param --limit --from --till].include?(arg) + own << argv.shift if %w[--param --limit --from --till --lock --history].include?(arg) else args << arg end @@ -54,7 +56,7 @@ DataPipeline.params.merge!(DataPipeline.parse_params(args)) { |_key, own_value, abort parser.help if script.nil? abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script) -DataPipeline.settings(options[:settings]) +DataPipeline.settings(options[:settings].merge(name: File.basename(script, '.rb'))) begin script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock] # held until the process ends diff --git a/lib/datapipeline.rb b/lib/datapipeline.rb index 59185de..97432d0 100644 --- a/lib/datapipeline.rb +++ b/lib/datapipeline.rb @@ -7,6 +7,7 @@ require_relative 'datapipeline/validation' require_relative 'datapipeline/layout' require_relative 'datapipeline/watermark' require_relative 'datapipeline/lock' +require_relative 'datapipeline/connections' class DataPipeline # plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel), @@ -30,21 +31,39 @@ class DataPipeline # read: source rows, written: body rows, output: rows to the target incl. header/trailer, # rejected: rows dropped by on_error :skip/:collect, errors: first 100 of them, # warnings: rows that failed a validate step with warn: true (kept), error: exception of an aborted run - Stats = Struct.new(:version, :read, :written, :output, :rejected, :errors, :warnings, :started_at, :finished_at, - :dry_run, :error, keyword_init: true) do + # totals: { 'column' => { read: Rational, written: Rational } } of the columns named by totals + Stats = Struct.new(:name, :version, :read, :written, :output, :rejected, :errors, :warnings, :started_at, + :finished_at, :dry_run, :error, :totals, keyword_init: true) do def duration = finished_at && started_at ? (finished_at - started_at).round(3) : nil def ok? = error.nil? && rejected.zero? + # stats.total('dmbtr') -> sum after all steps, stats.total('dmbtr', :read) -> sum of the source rows + def total column, side = :written + totals.to_h.dig(column.to_s, side) + end + def to_s - "read=#{read} written=#{written} rejected=#{rejected}#{warnings.to_i.positive? ? " warnings=#{warnings}" : ''} " \ + "read=#{read} written=#{written} rejected=#{rejected}#{warnings.to_i.positive? ? " warnings=#{warnings}" : ''}" \ + "#{totals.to_h.map { |col, sum| " #{col}=#{DataPipeline.decimal(sum[:read])}/#{DataPipeline.decimal(sum[:written])}" }.join} " \ "#{duration}s#{dry_run ? ' (dry run)' : ''}" \ "#{error ? " ABORTED: #{error.class}: #{error.message}" : ''}" end + + # one line of the run history + def to_h + { name:, version:, started_at: started_at&.iso8601(3), finished_at: finished_at&.iso8601(3), duration:, + read:, written:, output:, rejected:, warnings:, dry_run: dry_run ? true : false, ok: ok?, + error: error && "#{error.class}: #{error.message}", + totals: totals.to_h.transform_values { |sum| sum.transform_values { |v| DataPipeline.decimal(v) } } } + end end # stops reading at the limit, not an error class LimitReached < StandardError; end + # raised when a reconcile check fails + class ReconcileError < StandardError; end + EDGE_BEFORE = %i[prepend header].freeze EDGE_AFTER = %i[append trailer].freeze CHAINING = %i[map filter validate].freeze @@ -73,6 +92,8 @@ class DataPipeline preview: 10, on_error: :abort, lock: nil, + name: nil, + history: nil, log_level: Logger::WARN }.freeze @@ -157,6 +178,29 @@ class DataPipeline def self.reset_deprecations! = SETTINGS_LOCK.synchronize { @deprecations_logged = [] } + # see DataPipeline::Connections + def self.connections_file = Connections.file + def self.connections_file=(path) + Connections.file = path + end + + # entries of a run history file (see history), oldest first; name: only the runs of this pipeline + def self.history file, name: nil + return [] unless File.exist?(file) + + entries = File.foreach(file).filter_map { |line| JSON.parse(line) unless line.strip.empty? } + name ? entries.select { |entry| entry['name'] == name.to_s } : entries + end + + # exact decimal text of a sum: 1234.5 -> '1234.5', 1/3 -> '0.333333' + def self.decimal value + return value.to_s unless value.is_a?(Rational) + return value.to_i.to_s if value.denominator == 1 + + places = (1..10).find { |n| (value * 10**n).denominator == 1 } || 6 + format("%.#{places}f", value) + end + # DataPipeline.new(verbose: true) { ... } overrides the defaults for this pipeline only def initialize(**settings, &block) @settings = DataPipeline.settings.merge(settings) @@ -237,6 +281,43 @@ class DataPipeline @settings[:lock] = file end + # name of the pipeline in the stats and the run history; etl sets the script name as default + def name value + @settings[:name] = value + end + + # history 'data/runs.ndjson' -> one json line per run (also aborted and dry runs), see DataPipeline.history + def history file + @settings[:history] = file + end + + # + # totals 'dmbtr', 'menge' -> sums of these columns over the source rows and over the rows after all steps + # (stats.totals, stats.total('dmbtr')); empty and non numeric values are left out + # + def totals *columns + @totals = columns.flatten.map(&:to_s) + end + + # + # reconcile { |stats| 'Summe weicht ab' if stats.total('dmbtr', :read) != stats.total('dmbtr') } + # + # runs after all steps, before writing (streaming: after the last row); a message (or array of messages) + # aborts the run with ReconcileError, nil means ok + # + def reconcile(&block) + raise ArgumentError, 'reconcile: block missing' unless block + + (@reconcile ||= []) << block + end + + # connection(:pds, query: '...') -> container of the type in the connections file, options of the call win + def connection name, **options, &block + settings = Connections.fetch(name) + type = settings.fetch(:type) { raise ArgumentError, "connection #{name.inspect}: type missing" } + container(type, **settings.except(:type), **options, &block) + end + # see DataPipeline::Watermark def watermark name, column:, file:, initial: nil @watermark = Watermark.new(name, column:, file:, initial:) @@ -244,10 +325,21 @@ class DataPipeline # only a missing plugin is reported as unsupported, errors while loading it # (missing gem) or from the plugin itself (connection, options) are passed through + # + # container :postgres, connection: :pds, table: 'x' -> options from the connections file, the call wins + # def container type, **options, &block klass = DataPipeline.plugin(type) raise "Unsupported storage type or class missing: #{type.to_s.capitalize}Storage" unless klass + if (name = options.delete(:connection)) + settings = Connections.fetch(name) + if settings[:type] && settings[:type].to_s != type.to_s + raise ArgumentError, "connection #{name.inspect} is #{settings[:type]}, not #{type}" + end + + options = settings.except(:type).merge(options) + end klass.new(**options, &block) end @@ -371,8 +463,9 @@ class DataPipeline end def run - @stats = Stats.new(version: @settings[:version], read: 0, written: 0, output: 0, rejected: 0, errors: [], warnings: 0, - started_at: Time.now, dry_run: @settings[:dry_run]) + @stats = Stats.new(name: @settings[:name], version: @settings[:version], read: 0, written: 0, output: 0, rejected: 0, + errors: [], warnings: 0, started_at: Time.now, dry_run: @settings[:dry_run], + totals: @totals&.to_h { |col| [col, { read: 0r, written: 0r }] }) @logger.info "start#{" #{@settings[:version]}" if @settings[:version]}#{@settings[:dry_run] ? ' (dry run)' : ''}" @preview = [] raise ArgumentError, 'read storage missing' unless @source @@ -444,6 +537,7 @@ class DataPipeline self.class.runs << @stats print_preview if @stats.dry_run && @stats.error.nil? @logger.info "finish #{@stats}" + append_history @on_finish.each do |hook| hook.call(@stats) rescue StandardError => e @@ -451,6 +545,41 @@ class DataPipeline end end + # one json line per run, appended under an flock (parallel runs of other pipelines) + def append_history + return unless @settings[:history] + + FileUtils.mkdir_p(File.dirname(@settings[:history])) + File.open(@settings[:history], 'a') do |file| + file.flock(File::LOCK_EX) + file.puts JSON.generate(@stats.to_h) + end + rescue StandardError => e + @logger.error "history: #{e.class}: #{e.message}" + end + + def add_total side, row + @stats.totals&.each do |col, sum| + value = row.key?(col) ? row[col] : row[col.to_sym] + number = to_number(value) + sum[side] += number if number + end + end + + def to_number value + case value + when Float then value.finite? ? value.to_s.to_r : nil + when Numeric then value.to_r + when String then value.strip.match?(Layout::NUMBER) ? value.strip.to_r : nil + end + end + + # messages of the reconcile blocks abort the run + def check_reconcile + messages = Array(@reconcile).flat_map { |block| Array(block.call(@stats)).compact } + raise ReconcileError, messages.join('; ') if messages.any? + end + def source_rows data = @source.read data = data.first(@settings[:limit]) if @settings[:limit] @@ -475,6 +604,7 @@ class DataPipeline data = source_rows @stats.read = data.length data.each { |row| @watermark.track(row) } if @watermark + data.each { |row| add_total(:read, row) } if @totals dump 'Source#data', data # prepend keeps its old timing: called before the body @@ -489,6 +619,9 @@ class DataPipeline body.each { |step| rows.concat apply(step, data) } end dump 'Transformation#data', rows + @stats.written = rows.length + rows.each { |row| add_total(:written, row) } if @totals + check_reconcile ctx = Context.new(read: data.length, written: rows.length) before.select { |t| t[:kind] == :header }.each { |t| results[t] = edge(t, ctx, 0) } @@ -525,8 +658,10 @@ class DataPipeline read += 1 @watermark&.track(row) + add_total(:read, row) if @totals push(row, body, counters, nil) do |out| written += 1 + add_total(:written, out) if @totals emit.call out end end @@ -535,6 +670,7 @@ class DataPipeline end @stats.read = read @stats.written = written + check_reconcile ctx = Context.new(read:, written:) after.each { |t| emit.call edge(t, ctx, chained? ? written : read) } diff --git a/lib/datapipeline/connections.rb b/lib/datapipeline/connections.rb new file mode 100644 index 0000000..cf69e0c --- /dev/null +++ b/lib/datapipeline/connections.rb @@ -0,0 +1,59 @@ +require 'erb' +require 'yaml' + +class DataPipeline + # + # connections in one yaml file instead of credentials in every script + # + # # config/connections.yml (ERB allowed, secrets from ENV) + # pds: + # type: postgres + # host: <%= ENV.fetch('PDS_HOST', 'localhost') %> + # database: pds + # username: etl + # password: <%= ENV['PDS_PASSWORD'] %> + # + # read connection(:pds, query: 'select ...') -> container :postgres with these options + # container :postgres, connection: :pds, table: 'x' -> same, options of the call win + # + # file: DataPipeline.connections_file = '...', else ENV['DATAPIPELINE_CONNECTIONS'], else config/connections.yml + # + module Connections + DEFAULT_FILE = 'config/connections.yml'.freeze + + module_function + + def file + @file || ENV['DATAPIPELINE_CONNECTIONS'] || DEFAULT_FILE + end + + def file= path + @file = path + @all = nil + end + + # options of one connection with symbol keys, type included + def fetch name + all.fetch(name.to_s) do + raise ArgumentError, "connection #{name.inspect} not in #{file} (#{all.keys.join(', ')})" + end + end + + def all + @all = nil if @loaded_from != file + @loaded_from = file + @all ||= begin + raise ArgumentError, "connections file #{file} missing" unless File.exist?(file) + + data = YAML.safe_load(ERB.new(File.read(file)).result, aliases: true) || {} + raise ArgumentError, "#{file}: a mapping of connection names expected" unless data.is_a?(Hash) + + data.to_h { |name, options| [name.to_s, (options || {}).to_h { |key, value| [key.to_sym, value] }] } + end + end + + def reset! + @all = nil + end + end +end diff --git a/test/cli/etl_cli_test.rb b/test/cli/etl_cli_test.rb index 3f407d2..69f24b0 100644 --- a/test/cli/etl_cli_test.rb +++ b/test/cli/etl_cli_test.rb @@ -112,4 +112,19 @@ class EtlCliTest < PipelineTest assert_equal 0, status.exitstatus, err end end + + def test_history_with_script_name + with_tmpdir do |dir| + path = script(dir, 'export_kunden.rb', <<~RUBY) + DataPipeline.new do + read container(:csv, file: 'in.csv') + write container(:csv, file: 'out.csv') + end.run + RUBY + _out, err, status = etl(dir, 'run', path, '--history', 'runs.ndjson') + assert_equal 0, status.exitstatus, err + entry = JSON.parse(File.read("#{dir}/runs.ndjson")) + assert_equal ['export_kunden', 3, true], entry.values_at('name', 'read', 'ok') + end + end end diff --git a/test/core/connections_totals_history_test.rb b/test/core/connections_totals_history_test.rb new file mode 100644 index 0000000..fb22c12 --- /dev/null +++ b/test/core/connections_totals_history_test.rb @@ -0,0 +1,97 @@ +require_relative '../test_helper' + +# connections file, totals with reconcile, run history +class ConnectionsTotalsHistoryTest < PipelineTest + def teardown + DataPipeline.connections_file = nil + ENV.delete('ETL_TEST_SECRET') + super + end + + def write_connections dir + File.write("#{dir}/connections.yml", <<~YAML) + mem: + type: memory + rows: [{ a: 1 }] + secret: + type: csv + file: <%= ENV.fetch('ETL_TEST_SECRET', 'default.csv') %> + YAML + DataPipeline.connections_file = "#{dir}/connections.yml" + end + + def test_connection_and_container_with_connection + with_tmpdir do |dir| + write_connections(dir) + ENV['ETL_TEST_SECRET'] = "#{dir}/from_env.csv" + pipeline = DataPipeline.new { logger PipelineHelper::QUIET } + assert_kind_of MemoryStorage, pipeline.connection(:mem) + csv = pipeline.container(:csv, connection: 'secret') + assert_equal "#{dir}/from_env.csv", csv.instance_variable_get(:@file_name) + csv = pipeline.connection(:secret, file: "#{dir}/override.csv") + assert_equal "#{dir}/override.csv", csv.instance_variable_get(:@file_name) + + assert_includes assert_raises(ArgumentError) { pipeline.container(:json, connection: :secret) }.message, 'is csv, not json' + assert_includes assert_raises(ArgumentError) { pipeline.connection(:nope) }.message, 'mem, secret' + DataPipeline.connections_file = "#{dir}/missing.yml" + assert_includes assert_raises(ArgumentError) { pipeline.connection(:mem) }.message, 'missing.yml missing' + end + end + + ROWS = [{ 'beleg' => '1', 'dmbtr' => '10.10' }, { 'beleg' => '2', 'dmbtr' => 5.05 }, + { 'beleg' => '3', 'dmbtr' => '' }, { 'beleg' => '4', 'dmbtr' => 'x' }].freeze + + def test_totals_batch_and_streaming + [false, true].each do |streamed| + _target, stats = run_pipeline(ROWS, streamed:) do + totals 'dmbtr' + filter { |row| row['beleg'] != '2' } + end + assert_equal Rational(1515, 100), stats.total('dmbtr', :read) + assert_equal Rational(1010, 100), stats.total('dmbtr') + assert_includes stats.to_s, 'dmbtr=15.15/10.1' + end + end + + def test_reconcile_aborts_before_writing + target = nil + error = assert_raises(DataPipeline::ReconcileError) do + target, = run_pipeline(ROWS) do + totals 'dmbtr' + filter { |row| row['beleg'] != '2' } + reconcile { |stats| "Summe: #{stats.total('dmbtr', :read).to_f} statt #{stats.total('dmbtr').to_f}" if stats.total('dmbtr', :read) != stats.total('dmbtr') } + reconcile { |stats| ['zu wenig Zeilen'] if stats.written < 4 } + end + end + assert_equal 'Summe: 15.15 statt 10.1; zu wenig Zeilen', error.message + assert_nil target + + _target, stats = run_pipeline(ROWS) { totals 'dmbtr'; reconcile { |s| nil if s.total('dmbtr', :read) == s.total('dmbtr') } } + assert stats.ok? + end + + def test_history_lines_also_for_aborted_runs + with_tmpdir do |dir| + file = "#{dir}/log/runs.ndjson" + run_pipeline(ROWS) { name 'buchungen'; version '1.0'; history file; totals 'dmbtr' } + assert_raises(RuntimeError) { run_pipeline(ROWS) { name 'buchungen'; history file; map { raise 'kaputt' } } } + run_pipeline(ROWS) { name 'andere'; history file } + + entries = DataPipeline.history(file, name: 'buchungen') + assert_equal 2, entries.size + first, second = entries + assert_equal({ 'name' => 'buchungen', 'version' => '1.0', 'read' => 4, 'written' => 4, 'rejected' => 0, 'ok' => true, + 'error' => nil, 'totals' => { 'dmbtr' => { 'read' => '15.15', 'written' => '15.15' } } }, + first.slice('name', 'version', 'read', 'written', 'rejected', 'ok', 'error', 'totals')) + assert_equal [false, 'RuntimeError: kaputt'], second.values_at('ok', 'error') + assert Time.iso8601(first['started_at']) + assert_equal 3, DataPipeline.history(file).size + assert_equal [], DataPipeline.history("#{dir}/none.ndjson") + end + end + + def test_decimal + assert_equal ['1234.5', '0.333333', '7', '12'], [DataPipeline.decimal(Rational(12_345, 10)), DataPipeline.decimal(Rational(1, 3)), + DataPipeline.decimal(7r), DataPipeline.decimal(12)] + end +end