9 Commits
41 changed files with 3470 additions and 620 deletions
+55
View File
@@ -2,6 +2,61 @@
Framework `datapipeline` (Gem). Neueste Version oben.
## 3.2.0 – 2026-10-09
- Benachrichtigungen: `notify notifier, on: :failure | :problems | :always` nach einem Lauf, `DataPipeline::Report`
(Ereignis, Text, JSON-Payload `schema: 1`), Basisklasse `DataPipeline::Notifier` für eigene Kanäle; verworfene
Zeilen nur mit `include_rows: true`
- `DataPipeline::Webhook`: POST mit HMAC-SHA256-Signatur, Wiederholung bei 429/5xx (`Retry-After`), Timeout;
`Webhook.valid_signature?` für Empfänger
- `DataPipeline::Email`: Bericht als Text-Mail über SMTP (STARTTLS, TLS, Anmeldung), Gem `net-smtp`
- `DataPipeline::Slack`: `deliver(report)` mit Block Kit, Maskierung von `&`, `<`, `>`, `mention:` bei Abbruch,
Incoming Webhooks (`webhook:`), Timeout und Wiederholung; Fehler als `DataPipeline::Slack::Error`
- Notifier aus der Verbindungsdatei (`notifier(:name)`, `type: webhook | slack | email`); `etl --notify=NAME
[--notify-on=…]` sendet einen Bericht für alle Pipelines eines Skripts
## 3.1.0 – 2026-10-09
- sqlite: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion;
`batch_size: :all` = eine Transaktion für den ganzen Stream, nach einem Abbruch zurückgerollt
- postgres, sqlite: `query` schreibt noch gesammelte Zeilen von `write_row` vorher
- Lizenz BSD-3-Clause (`LICENSE`, gemspec)
## 3.0.0 – 2026-10-09
- **Breaking:** `transform` (auch `processing: :prepend`/`:append`), der Parallelmodus, `chained` und `condition:` als
String (`eval`) entfernt; alle Schritte laufen verkettet. Umstellung: `map`, `header`, `trailer`,
`condition: ->(row) { … }` (README 7.1)
- `condition:` muss ein Lambda sein, sonst `ArgumentError` beim Aufbau der Pipeline
- Hinweise `WARN deprecated` und `test/core/compat_test.rb` entfernt
## 2.1.0 – 2026-10-09
- Sperre gegen parallele Läufe: `lock 'datei'` in der Pipeline, `etl --lock=datei` für das ganze Skript; ein zweiter
Lauf bricht mit `DataPipeline::LockError` ab, Exit-Code `3`
- postgres, sqlite: `upsert: %w[schlüssel]` (`insert … on conflict … do update`); sqlite legt mit `create: true` den
Unique-Index an
- postgres: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion statt je Zeile
einzeln; `batch_size: :all` = eine Transaktion für den ganzen Stream, nach einem Abbruch zurückgerollt
- csv: `encoding:` (z. B. `'ISO-8859-1'`, `'Windows-1252'`), `replace:` für nicht darstellbare Zeichen, `bom: true`;
ein UTF-8-BOM wird beim Lesen übersprungen
- Layout: `format: true` bringt Werte vor Prüfen und Schreiben ins Format ihres Typs (`N` mit Nachkommastellen,
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
- 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`
- README neu gegliedert (Überblick, Beispiele, Pipelines schreiben, Plugins, Layouts, Betrieb, Umstieg,
Entwicklung) mit Inhaltsverzeichnis, Fehlersuche und Umstiegshinweisen; neues Beispiel
`examples/buchungen_export.rb` für einen produktiven Lauf
## 2.0.0 – 2026-10-08
- Gem `datapipeline`: `datapipeline.gemspec`, Executable `etl` (exe/etl), Code unter `lib/datapipeline/`
+2 -1
View File
@@ -7,7 +7,8 @@ gemspec
# optional plugin gems: needed for the plugin tests, loaded only when a plugin is used
group :plugins do
gem 'caxlsx', require: false # excel write
gem 'http', require: false # rest, slack
gem 'http', require: false # rest, slack, webhook
gem 'net-smtp', require: false # email
gem 'pg', require: false # postgres
gem 'roo', require: false # excel read
gem 'sqlite3', require: false # sqlite
+7 -1
View File
@@ -1,7 +1,7 @@
PATH
remote: .
specs:
datapipeline (2.0.0)
datapipeline (3.2.0)
csv (~> 3.3)
logger (~> 1.6)
@@ -26,6 +26,10 @@ GEM
logger (1.7.0)
marcel (1.2.1)
minitest (5.27.0)
net-protocol (0.4.0)
timeout
net-smtp (0.5.2)
net-protocol
nokogiri (1.19.4-aarch64-linux-gnu)
racc (~> 1.4)
nokogiri (1.19.4-aarch64-linux-musl)
@@ -66,6 +70,7 @@ GEM
sqlite3 (2.9.6-x86_64-darwin)
sqlite3 (2.9.6-x86_64-linux-gnu)
sqlite3 (2.9.6-x86_64-linux-musl)
timeout (0.6.1)
PLATFORMS
aarch64-linux
@@ -84,6 +89,7 @@ DEPENDENCIES
datapipeline!
http
minitest (~> 5.25)
net-smtp
pg
rake
roo
+28
View File
@@ -0,0 +1,28 @@
BSD 3-Clause License
Copyright (c) 2026, Karl-Heinz Wild
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
1. Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
3. Neither the name of the copyright holder nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+1167 -328
View File
File diff suppressed because it is too large Load Diff
+3 -3
View File
@@ -12,13 +12,13 @@ Gem::Specification.new do |spec|
Fehlerdateien, Dry-Run, inkrementellen Läufen und dem Kommando etl mit Exit-Codes.
TEXT
spec.homepage = 'https://git.wicom.li/wkh/datapipeline'
spec.license = 'Nonstandard'
spec.license = 'BSD-3-Clause'
spec.required_ruby_version = '>= 3.4'
spec.metadata['changelog_uri'] = "#{spec.homepage}/src/branch/main/CHANGELOG.md"
spec.metadata['rubygems_mfa_required'] = 'true'
spec.files = Dir['lib/**/*.rb', 'exe/*', 'README.md', 'CHANGELOG.md']
spec.files = Dir['lib/**/*.rb', 'exe/*', 'README.md', 'CHANGELOG.md', 'LICENSE']
spec.bindir = 'exe'
spec.executables = ['etl']
spec.require_paths = ['lib']
@@ -28,5 +28,5 @@ Gem::Specification.new do |spec|
spec.add_dependency 'logger', '~> 1.6'
# optional, loaded only when the plugin is used (add them to the Gemfile of the application):
# pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack)
# pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack, webhook), net-smtp (email)
end
+54
View File
@@ -0,0 +1,54 @@
#
# Produktiver Lauf: Buchungen prüfen, Summe und Saldo abgleichen, wiederholbar in die Buchhaltungs-Datenbank
# und als CSV in Windows-1252 für das Zielsystem schreiben; mit Sperre, Laufhistorie und Slack bei Problemen
#
# bin/etl run examples/buchungen_export.rb [--out=tmp] [--dry-run]
#
require 'fileutils'
require 'datapipeline'
DATA = File.expand_path('data', __dir__)
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)
# Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml)
DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__)
DataPipeline.new do
name 'buchungen_export'
lock "#{out}/.buchungen.lock" # läuft schon ein Export, bricht dieser ab (Exit-Code 3)
history "#{out}/runs.ndjson" # eine JSON-Zeile je Lauf, auch nach Abbruch
notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] # Meldung nur bei Problemen
read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' })
# 1. prüfen: ohne on_error bricht ein Fehler den Lauf ab, es entsteht keine halbe Buchung
validate 'belnr' => { required: true, format: /\A\d{10}\z/ },
'konto' => { required: true, format: /\A\d+\z/ },
'betrag' => { required: true, type: :decimal }
# 2. umbauen mit Kurzschritten und einem map
default 'text' => 'ohne Text'
rename 'text' => 'buchungstext'
cast 'buzei' => :integer, 'datum' => :date, 'betrag' => :decimal
map do |row|
row.merge('shkzg' => row['betrag'].negative? ? 'H' : 'S', 'dmbtr' => format('%.2f', row['betrag'].abs))
end
select %w[belnr buzei datum konto shkzg dmbtr betrag buchungstext]
# 3. abgleichen, bevor etwas geschrieben wird
totals 'betrag'
reconcile do |stats|
quelle, ergebnis = stats.total('betrag', :read), stats.total('betrag')
next "Summe geändert: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ergebnis)}" if quelle != ergebnis
"Soll und Haben ungleich, Saldo #{DataPipeline.decimal(ergebnis)}" unless ergebnis.zero?
end
# 4. schreiben: Upsert macht den Lauf wiederholbar; die CSV-Datei im Zeichensatz des Zielsystems
write connection(:buchhaltung)
write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' },
columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext])
on_finish { |stats| puts "#{stats.name}: #{stats}" }
end.run
+8
View File
@@ -0,0 +1,8 @@
# Verbindungen der Beispiele. Die Datei wird mit ERB gelesen: Zugangsdaten und Pfade kommen aus der Umgebung,
# bei Postgres z. B. password: <%= ENV['PDS_PASSWORD'] %>
buchhaltung:
type: sqlite
database: <%= ENV.fetch('BUCHHALTUNG_DB', 'tmp/buchhaltung.sqlite3') %>
table: buchungen
create: true
upsert: [belnr, buzei]
+8
View File
@@ -0,0 +1,8 @@
belnr;buzei;datum;konto;betrag;text
5100000001;1;2026-10-01;1020;1250.00;Zahlung Müller GmbH
5100000001;2;2026-10-01;1100;-1250.00;Zahlung Müller GmbH
5100000002;1;2026-10-02;6500;380.50;Büromaterial
5100000002;2;2026-10-02;1170;29.30;Vorsteuer
5100000002;3;2026-10-02;2000;-409.80;Papeterie Zürcher AG
5100000003;1;2026-10-03;3200;99.95;
5100000003;2;2026-10-03;1020;-99.95;Gutschrift
1 belnr buzei datum konto betrag text
2 5100000001 1 2026-10-01 1020 1250.00 Zahlung Müller GmbH
3 5100000001 2 2026-10-01 1100 -1250.00 Zahlung Müller GmbH
4 5100000002 1 2026-10-02 6500 380.50 Büromaterial
5 5100000002 2 2026-10-02 1170 29.30 Vorsteuer
6 5100000002 3 2026-10-02 2000 -409.80 Papeterie Zürcher AG
7 5100000003 1 2026-10-03 3200 99.95
8 5100000003 2 2026-10-03 1020 -99.95 Gutschrift
+43 -5
View File
@@ -9,9 +9,13 @@
# --dry-run run everything, write nothing, print the first rows
# --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)
# --notify=NAME one report for all runs of the script, notifier NAME from the connections file
# --notify-on=WHEN failure, problems (default) or always
# --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
# exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run
#
require 'optparse'
require 'logger'
@@ -31,6 +35,10 @@ parser = OptionParser.new do |opts|
opts.on('--limit=N', Integer) { |v| options[:settings][:limit] = v }
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 }
opts.on('--notify=NAME', String) { |v| options[:notify] = v }
opts.on('--notify-on=WHEN', String) { |v| options[:notify_on] = v }
end
# own options for the parser, everything else (also --name=value of the script) goes to the script
@@ -38,9 +46,9 @@ own = []
args = []
argv = ARGV.dup
while (arg = argv.shift)
if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet)(=|\z)/)
if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history|notify|notify-on)(=|\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 --notify --notify-on].include?(arg)
else
args << arg
end
@@ -52,7 +60,33 @@ 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])
name = File.basename(script, '.rb')
DataPipeline.settings(options[:settings].merge(name:))
# one report for the whole script; created before the run, so a wrong name fails early
if options[:notify]
begin
notifier = DataPipeline::Notifier.from_connection(options[:notify])
notify_when = DataPipeline::Notifier.condition(options.fetch(:notify_on, :problems))
rescue ArgumentError, LoadError => e
abort "etl: --notify=#{options[:notify]}: #{e.message}"
end
end
summary = lambda do |error = nil|
return unless notifier && !options[:settings][:dry_run]
report = DataPipeline::Report.new(DataPipeline.runs, title: name, params: DataPipeline.params, error:)
notifier.deliver(report) if notify_when.call(report)
rescue StandardError => e
warn "#{name}: notify #{options[:notify]}: #{e.class}: #{e.message}"
end
begin
script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock] # held until the process ends
rescue DataPipeline::LockError => e
warn "#{File.basename(script)}: #{e.message}"
exit 3
end
# the pipeline script sees only its own arguments
ARGV.replace(options[:passthrough] + args)
@@ -61,10 +95,14 @@ begin
load File.expand_path(script)
rescue StandardError => e
warn "#{File.basename(script)}: #{e.class}: #{e.message}"
exit 1
# an aborted pipeline is already in the runs, other errors (script, lock) are reported as such
summary.call(DataPipeline.runs.any? { |stats| stats.error.equal?(e) } ? nil : e) unless e.is_a?(DataPipeline::LockError)
exit(e.is_a?(DataPipeline::LockError) ? 3 : 1)
end
summary.call
runs = DataPipeline.runs
exit 3 if runs.any? { |stats| stats.error.is_a?(DataPipeline::LockError) }
exit 1 if runs.any?(&:error)
exit 2 if runs.any? { |stats| stats.rejected.positive? }
exit 0
+273 -87
View File
@@ -6,17 +6,23 @@ require_relative 'datapipeline/storage/storage_plugin'
require_relative 'datapipeline/validation'
require_relative 'datapipeline/layout'
require_relative 'datapipeline/watermark'
require_relative 'datapipeline/lock'
require_relative 'datapipeline/connections'
require_relative 'datapipeline/notification/notifier'
class DataPipeline
# plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel),
# http (rest, slack)
# http (rest, slack, webhook), net-smtp (email)
autoload :CsvStorage, File.expand_path('datapipeline/storage/csv', __dir__)
autoload :PostgresStorage, File.expand_path('datapipeline/storage/postgres', __dir__)
autoload :JsonStorage, File.expand_path('datapipeline/storage/json', __dir__)
autoload :SqliteStorage, File.expand_path('datapipeline/storage/sqlite', __dir__)
autoload :ExcelStorage, File.expand_path('datapipeline/storage/excel', __dir__)
autoload :RestStorage, File.expand_path('datapipeline/storage/rest', __dir__)
autoload :FixedStorage, File.expand_path('datapipeline/storage/fixed', __dir__)
autoload :Slack, File.expand_path('datapipeline/notification/slack', __dir__)
autoload :Webhook, File.expand_path('datapipeline/notification/webhook', __dir__)
autoload :Email, File.expand_path('datapipeline/notification/email', __dir__)
# passed to header, trailer, map and filter
# read: number of source rows (nil while streaming, until the trailer)
@@ -28,36 +34,56 @@ 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}" : ''}" \
"#{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
EDGE_BEFORE = %i[prepend header].freeze
EDGE_AFTER = %i[append trailer].freeze
CHAINING = %i[map filter validate].freeze
# raised when a reconcile check fails
class ReconcileError < StandardError; end
ON_ERROR = %i[abort skip collect].freeze
MAX_ERRORS = 100
# value conversions for cast, empty values become nil
CASTS = Validation::TYPES.merge(string: ->(v) { v.to_s }).freeze
DEFAULT_SETTINGS = {
version: nil,
streamed: false,
chained: false,
verbose: false,
dry_run: false,
limit: nil,
preview: 10,
on_error: :abort,
lock: nil,
name: nil,
history: nil,
log_level: Logger::WARN
}.freeze
@@ -129,10 +155,33 @@ class DataPipeline
@runs ||= []
end
# 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)
@transformations = []
@steps = []
@targets = []
@on_finish = []
@logger = Logger.new($stderr, level: @settings[:log_level], progname: 'etl')
@@ -160,11 +209,6 @@ class DataPipeline
@settings[:streamed] = flag
end
# each body step gets the output of the previous step instead of the source rows
def chained flag = true
@settings[:chained] = flag
end
# run everything, write nothing (target, rejects, watermark); prints the first rows
def dry_run flag = true
@settings[:dry_run] = flag
@@ -199,11 +243,73 @@ class DataPipeline
@reject_target = to
end
#
# notify notifier, on: :problems -> a report after the run, see DataPipeline::Notifier
# on: :failure (abort), :problems (abort, rejected rows or warnings, default), :always
# title: default the name of the pipeline
# dry_run: true -> also after a dry run (default: not)
# include_rows: true -> rejected rows in the report (default: not, they may hold customer data)
#
# an error while sending is only logged
#
def notify notifier, on: :problems, title: nil, dry_run: false, include_rows: false
raise ArgumentError, 'notify: notifier needs deliver(report)' unless notifier.respond_to?(:deliver)
(@notifiers ||= []) << { notifier:, when: Notifier.condition(on), title:, dry_run:, include_rows: }
end
# notifier(:alerts) -> Webhook, Slack or Email from the connections file (type webhook, slack, email)
def notifier name
Notifier.from_connection(name)
end
# on_finish { |stats| ... } -> after every run, also after an abort (stats.error)
def on_finish(&block)
@on_finish << block
end
# lock 'data/.cudo.lock' -> held during run; a second run at the same time aborts with LockError
def lock file
@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:)
@@ -211,34 +317,75 @@ 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
#
# transform(options) { |row, index| ... }
# processing: :prepend | :normal (default) | :append
# condition: ruby expression as string (row is visible) or lambda { |row| ... }
#
def transform(options = {}, &block)
kind = options.fetch(:processing, :normal)
kind = :normal unless %i[prepend append].include?(kind)
add_step kind, options, &block
end
# map { |row, ctx| ... } -> hash, array of hashes or nil; turns chaining on
# map { |row, ctx| ... } -> hash, array of hashes or nil
# condition: ->(row) { ... } on map and filter: the step only for matching rows, the others are dropped
def map(options = {}, &block)
add_step :map, options, &block
end
# filter { |row, ctx| ... } -> keeps the row when the block is truthy; turns chaining on
# filter { |row, ctx| ... } -> keeps the row when the block is truthy
def filter(options = {}, &block)
add_step :filter, options, &block
end
#
# short steps, each one a map step
#
# 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
@@ -254,7 +401,7 @@ class DataPipeline
# validate layout -> fields of a DataPipeline::Layout
# validate { |row| 'soll ungleich haben' if ... } -> nil, message or array of messages
#
# a failing row raises ValidationError, handled by on_error; turns chaining on
# a failing row raises ValidationError, handled by on_error
# warn: true -> only a warning in the log and stats.warnings, the row is kept
# ignore: %w[A B] -> messages for these fields are left out (known gaps)
#
@@ -262,7 +409,7 @@ class DataPipeline
warn = options.delete(:warn) || false
ignore = Array(options.delete(:ignore)).map(&:to_s)
rules ||= options # validate 'konto' => { ... } arrives as keywords
@transformations << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
@steps << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
end
def read storage, options = {}
@@ -285,7 +432,7 @@ class DataPipeline
end
#
# loads a table once, for lookups in map/transform instead of one query per row
# loads a table once, for lookups in map instead of one query per row
#
# lookup pg, query: 'select ...', params: [...], key: 'rechnung_dbid' -> { key => [rows] }
# lookup pg, query: 'select ...', key: 'id', unique: true -> { key => row }
@@ -298,16 +445,18 @@ 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
(@lock = Lock.new(@settings[:lock])).acquire if @settings[:lock]
raise ArgumentError, 'write storage missing' if @targets.empty?
before = @transformations.select { |t| EDGE_BEFORE.include?(t[:kind]) }
after = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) }
body = @transformations - before - after
before = @steps.select { |t| t[:kind] == :header }
after = @steps.select { |t| t[:kind] == :trailer }
body = @steps - before - after
@settings[:streamed] ? run_streamed(before, body, after) : run_batch(before, body, after)
@@ -325,14 +474,25 @@ class DataPipeline
def add_step kind, options, &block
raise ArgumentError, "#{kind}: block missing" unless block
@transformations << { kind:, block:, options: }
if options.key?(:condition) && !options[:condition].respond_to?(:call)
raise ArgumentError, "#{kind}: condition must be a lambda ->(row) { ... }"
end
# without any transformation the source is copied unchanged
def chained?
@settings[:chained] || @transformations.empty? ||
@transformations.any? { |t| CHAINING.include?(t[:kind]) }
@steps << { 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
def write_row row
@@ -342,9 +502,12 @@ class DataPipeline
def finish
@stats.finished_at = Time.now
[@source, *@targets.map { |t| t[:storage] }, @reject_target].compact.uniq.each(&:close)
@lock&.release
self.class.runs << @stats
print_preview if @stats.dry_run && @stats.error.nil?
@logger.info "finish #{@stats}"
append_history
send_notifications
@on_finish.each do |hook|
hook.call(@stats)
rescue StandardError => e
@@ -352,6 +515,55 @@ 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 send_notifications
Array(@notifiers).each do |entry|
next if @stats.dry_run && !entry[:dry_run]
report = Report.new([@stats], title: entry[:title], params:)
next unless entry[:when].call(report)
entry[:notifier].deliver(report, include_rows: entry[:include_rows])
@logger.info "notified #{entry[:notifier].class.name.split('::').last}: #{report.event}"
rescue StandardError => e
@logger.error "notify #{entry[:notifier].class}: #{e.class}: #{e.message}"
end
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]
@@ -376,32 +588,24 @@ 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
results = {}.compare_by_identity
before.select { |t| t[:kind] == :prepend }.each { |t| results[t] = edge(t, nil, 0) }
rows = []
if chained?
counters = Array.new(body.size, 0)
data.each { |row| push(row, body, counters, data.length) { |out| rows << out } }
else
body.each { |step| rows.concat apply(step, data) }
end
dump 'Transformation#data', rows
ctx = Context.new(read: data.length, written: rows.length)
before.select { |t| t[:kind] == :header }.each { |t| results[t] = edge(t, ctx, 0) }
# parallel mode keeps the old count of source rows for append
count = chained? ? rows.length : data.length
new_data = []
before.each { |t| add_rows new_data, results[t] }
new_data.concat rows
after.each { |t| add_rows new_data, edge(t, ctx, count) }
@stats.written = rows.length
rows.each { |row| add_total(:written, row) } if @totals
check_reconcile
# header is computed after the body, so it knows the counters
ctx = Context.new(read: data.length, written: rows.length)
new_data = []
before.each { |t| add_rows new_data, t[:block].call(ctx) }
new_data.concat rows
after.each { |t| add_rows new_data, t[:block].call(ctx) }
dump 'Target#data', new_data
output(new_data) do |all|
@targets.each { |t| t[:storage].write(t[:only] ? all.select(&t[:only]) : all) }
@@ -410,13 +614,9 @@ class DataPipeline
# row by row from source.read { } to target.write_row, header only knows nothing yet
def run_streamed before, body, after
if !chained? && body.size > 1
raise ArgumentError, 'streamed: several transformations need chained (or map/filter)'
end
emit = ->(result) { output(add_rows([], result)) { |rows| rows.each { |row| write_row(row) } } }
before.each { |t| emit.call edge(t, Context.new, 0) }
before.each { |t| emit.call t[:block].call(Context.new) }
read = written = 0
counters = Array.new(body.size, 0)
@@ -426,8 +626,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
@@ -436,9 +638,10 @@ 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) }
after.each { |t| emit.call t[:block].call(ctx) }
@targets.each { |t| t[:storage].finalize }
end
@@ -453,11 +656,6 @@ class DataPipeline
end
end
# parallel mode: every step reads its own copy of the source rows
def apply step, rows
rows.each_with_index.flat_map { |row, index| safe_step_rows(step, row.dup, index, rows.length) }
end
# errors of a step for one row go to on_error
def safe_step_rows step, row, index, read
step_rows(step, row, index, read)
@@ -479,7 +677,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}##{@steps.index(step) + 1}"
end
# rows produced by one step for one input row
@@ -507,26 +706,13 @@ class DataPipeline
step[:block].call(row, Context.new(read:, index:)) ? [row] : []
when :map
add_rows [], step[:block].call(row, Context.new(read:, index:))
else
add_rows [], step[:block].call(row, index)
end
end
def edge step, ctx, count
case step[:kind]
when :prepend then step[:block].call(nil, 0)
when :append then step[:block].call(nil, count)
else step[:block].call(ctx)
end
end
# string conditions are evaluated with the local variable row
# condition: ->(row) { ... } on map and filter
def condition? step, row
condition = step[:options].fetch(:condition, true)
return true if condition == true
return condition.call(row) if condition.respond_to?(:call)
eval(condition)
condition = step[:options][:condition]
condition.nil? || condition.call(row)
end
# a hash is one row, an array are several rows, nil is no row
+59
View File
@@ -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
+157 -8
View File
@@ -18,8 +18,15 @@ class DataPipeline
# keys of a row are matched without regard to case, string or symbol.
#
# write container(:csv, file: ..., layout: layout) -> values in field order, unknown keys raise
# write container(:fixed, file: ..., layout: layout) -> fixed width lines (see FixedStorage)
# read container(:csv, ..., layout: layout) -> rows with the field names as keys
# validate layout -> required, length, format per field
#
# format: true converts values to the text format of the field type, before checking and writing:
# N numbers with the decimals of the length (13.2 -> 2), rounded half up; numeric texts too
# DAT Date, Time or 'YYYY-MM-DD' -> YYYYMMDD
# TIM Time or 'HH:MM:SS' -> HHMMSS
#
class Layout
Field = Struct.new(:name, :type, :length, :required) do
def key = name.downcase
@@ -30,13 +37,30 @@ class DataPipeline
after ||= 0
[before - after, after]
end
# characters in a fixed width file: N 13.2 -> 14 (11 digits, point, 2 decimals), DAT 8, TIM 6
def width
case type
when 'DAT' then 8
when 'TIM' then 6
when 'N'
return unless length
before, after = digits
before + after + (after.positive? ? 1 : 0)
else length&.to_i
end
end
end
NUMBER = /\A[+-]?\d+(\.\d+)?\z/
attr_reader :records
def initialize type_field: nil, strict: true, &block
def initialize type_field: nil, strict: true, format: false, &block
@type_field = type_field&.to_s&.downcase
@strict = strict
@format = format
@records = {}
instance_eval(&block) if block
end
@@ -56,23 +80,148 @@ class DataPipeline
def values row
fields = fields_for(row)
normalized = normalize(row)
if @strict
unknown = normalized.keys - fields.map(&:key)
raise LayoutError, "#{record_type(row)}: fields not in layout: #{unknown.join(', ')}" if unknown.any?
end
fields.map { |field| normalized[field.key] }
unknown!(row, fields, normalized)
fields.map { |field| output(field, normalized[field.key]) }
end
def errors row
fields = fields_for(row)
normalized = normalize(row)
fields.flat_map { |field| check(field, normalized[field.key]) }
fields.flat_map { |field| check(field, output(field, normalized[field.key])) }
rescue LayoutError => e
[e.message]
end
# value as text in the format of the field type (see format: true)
def format_value field, value
case value
when nil then nil
when Numeric then field.type == 'N' ? format_number(field, value) : value.to_s
when Date, Time then format_time(field, value)
else format_text(field, value.to_s)
end
end
# one line of a fixed width file: text left aligned, numbers right aligned with zeros, empty fields blank
def line row
fields = fields_for(row)
normalized = normalize(row)
unknown!(row, fields, normalized)
fields.map { |field| fixed(field, format_value(field, normalized[field.key])) }.join
end
# values of a line in field order (csv without header) -> row with the field names as keys
def parse values
type, fields = find_record { |fields| values[type_index(fields)].to_s.strip }
raise LayoutError, "record type #{values[0].inspect} not in layout (#{@records.keys.join(', ')})" unless fields
raise LayoutError, "#{type}: #{values.size} values, layout has #{fields.size} fields" if @strict && values.size > fields.size
fields.each_with_index.to_h { |field, i| [field.key, values[i]] }
end
# a line of a fixed width file -> row with the field names as keys, values without padding, blank -> nil
def parse_line line
type, fields = find_record { |fields| slices(line, fields)[type_index(fields)].to_s.strip }
raise LayoutError, "record type of #{line[0, 20].inspect} not in layout (#{@records.keys.join(', ')})" unless fields
fields.zip(slices(line, fields)).to_h do |field, text|
text = field.type == 'AN' ? text.to_s.rstrip : text.to_s.strip
[field.key, text.empty? ? nil : text]
end
end
# fixed width needs a length for every AN and N field
def check_widths!
missing = @records.flat_map { |type, fields| fields.reject(&:width).map { |f| "#{type}.#{f.name}" } }
raise LayoutError, "fixed width needs a length for: #{missing.join(', ')}" if missing.any?
self
end
private
def output field, value
@format ? format_value(field, value) : value
end
def unknown! row, fields, normalized
return unless @strict
unknown = normalized.keys - fields.map(&:key)
raise LayoutError, "#{record_type(row)}: fields not in layout: #{unknown.join(', ')}" if unknown.any?
end
def find_record
@records.find { |type, fields| yield(fields) == type }
end
def type_index fields
@type_field ? fields.index { |f| f.key == @type_field } || 0 : 0
end
def slices line, fields
offset = 0
fields.map do |field|
text = line[offset, field.width]
offset += field.width
text
end
end
# 12.345 -> '12.35' for N 11.2; floats by their shortest text, so 12.345 is not 12.3449999...
def format_number field, value
_before, after = field.length ? field.digits : [nil, 0]
number = value.is_a?(Float) ? value.to_s.to_r : value.to_r
number = number.round(after, half: :up)
sign = number.negative? ? '-' : ''
int, frac = number.abs.divmod(1)
return "#{sign}#{int}" if after.zero?
"#{sign}#{int}.#{(frac * 10**after).round.to_s.rjust(after, '0')}"
end
def format_time field, value
case field.type
when 'DAT' then value.strftime('%Y%m%d')
when 'TIM' then value.respond_to?(:hour) ? value.strftime('%H%M%S') : value.to_s
else value.respond_to?(:hour) ? value.iso8601 : value.to_s
end
end
# numeric texts get the decimals of the field, ISO dates and times lose their separators
def format_text field, text
case field.type
when 'N'
_before, after = field.length ? field.digits : [nil, 0]
return text unless text.match?(NUMBER) && after.positive?
decimals = text.split('.')[1].to_s
decimals.length > after ? text : "#{text.split('.')[0]}.#{decimals.ljust(after, '0')}"
when 'DAT' then text.match?(/\A\d{4}-\d{2}-\d{2}\z/) ? text.delete('-') : text
when 'TIM' then text.match?(/\A\d{2}:\d{2}:\d{2}\z/) ? text.delete(':') : text
else text
end
end
def fixed field, text
width = field.width
return ' ' * width if text.nil? || text.empty?
raise LayoutError, "#{field.name}: #{text.inspect} longer than #{width}" if text.length > width
case field.type
when 'N' then raise LayoutError, "#{field.name}: #{text.inspect} is not a number" unless text.match?(NUMBER)
when 'DAT', 'TIM' then raise LayoutError, "#{field.name}: #{text.inspect} needs #{width} digits" unless text.match?(/\A\d{#{width}}\z/)
else return text.ljust(width)
end
return text if field.type != 'N'
sign = text.start_with?('-') ? '-' : ''
digits = text.delete('+-')
raise LayoutError, "#{field.name}: #{text.inspect} longer than #{width}" if sign.length + digits.length > width
sign + digits.rjust(width - sign.length, '0')
end
def record_type row
@type_field ? normalize(row)[@type_field] : row.values.first
end
@@ -95,7 +244,7 @@ class DataPipeline
end
def check_number field, text
return ["#{field.name}: not a number"] unless text.match?(/\A[+-]?\d+(\.\d+)?\z/)
return ["#{field.name}: not a number"] unless text.match?(NUMBER)
return [] unless field.length
before, after = field.digits
+57
View File
@@ -0,0 +1,57 @@
require 'fileutils'
class DataPipeline
# another run holds the lock
class LockError < StandardError; end
#
# exclusive lock on a file, against overlapping runs (cron)
#
# lock 'data/.cudo.lock' -> in a pipeline: held from the start to the end of run
# etl run cudo.rb --lock=data/.lock -> for the whole script
#
# the lock is an flock on the file, released by the operating system when the process ends,
# also after a crash; the file stays and holds the pid of the last holder
#
class Lock
attr_reader :file
def initialize file
@file = file.to_s
end
def acquire
return self if @io
FileUtils.mkdir_p(File.dirname(@file))
io = File.open(@file, File::RDWR | File::CREAT, 0o644)
unless io.flock(File::LOCK_EX | File::LOCK_NB)
holder = io.read.strip
io.close
raise LockError, "#{@file} is locked#{" by pid #{holder}" unless holder.empty?}: another run is still active"
end
io.truncate(0)
io.write(Process.pid.to_s)
io.flush
@io = io
self
end
def release
return unless @io
@io.flock(File::LOCK_UN)
@io.close
@io = nil
end
def locked? = !@io.nil?
def synchronize
acquire
yield
ensure
release
end
end
end
+93
View File
@@ -0,0 +1,93 @@
require 'securerandom'
require_relative 'notifier'
require_relative '../storage/storage_plugin'
class DataPipeline
#
# e-mail over SMTP, text/plain UTF-8
#
# notify DataPipeline::Email.new(to: 'etl@example.com'), on: :failure
#
# options (defaults from ENV)
# to: one address or an array (required)
# from: sender, default ENV['SMTP_FROM'] or etl@<hostname>
# host: ENV['SMTP_HOST'], default localhost
# port: ENV['SMTP_PORT'], default 587 (465 with tls: true, 25 without starttls)
# username: ENV['SMTP_USERNAME'], password: ENV['SMTP_PASSWORD']; without username no login
# tls: true -> TLS from the start (port 465)
# starttls: :auto (default, if the server offers it), :always or false
# subject_prefix: default '[etl] '
# timeout: seconds for open and read (default 15)
#
class Email < Notifier
class Error < StandardError; end
def initialize to:, from: nil, host: nil, port: nil, username: nil, password: nil, tls: false,
starttls: :auto, subject_prefix: '[etl] ', timeout: 15
StoragePlugin.require_gem 'net/smtp', 'net-smtp'
@to = Array(to).map(&:to_s).reject(&:empty?)
raise ArgumentError, 'Email: to missing' if @to.empty?
@from = from || ENV['SMTP_FROM'] || "etl@#{Socket.gethostname}"
@host = host || ENV.fetch('SMTP_HOST', 'localhost')
@tls = tls
@starttls = tls || !starttls ? false : starttls.to_sym
@port = Integer(port || ENV['SMTP_PORT'] || default_port)
@username = username || ENV['SMTP_USERNAME']
@password = password || ENV['SMTP_PASSWORD']
@subject_prefix = subject_prefix
@timeout = timeout
end
def deliver report, include_rows: false
send_mail("#{@subject_prefix}#{report.subject}", body_of(report, include_rows))
end
# any text, e.g. from on_finish
def send_mail subject, body
message = build(subject, body)
smtp = Net::SMTP.new(@host, @port, tls: @tls, starttls: @starttls || false)
smtp.open_timeout = @timeout
smtp.read_timeout = @timeout
login = @username ? { user: @username, secret: @password, authtype: :login } : {}
smtp.start(helo: Socket.gethostname, **login) { |session| session.send_message(message, @from, *@to) }
message
rescue Net::SMTPError, SystemCallError, IOError, Timeout::Error, OpenSSL::SSL::SSLError => e
raise Error, "#{@host}:#{@port}: #{e.class}: #{e.message}"
end
private
def default_port
return 465 if @tls
@starttls ? 587 : 25
end
# rows of rejected lines only on request, they may hold customer data
def body_of report, include_rows
text = report.to_s
return text unless include_rows
rows = report.runs.flat_map { |stats| stats.errors.first(Report::MAX_ERRORS).map { |e| "#{e[:step]}: #{JSON.generate(e[:row])}" } }
rows.empty? ? text : "#{text}\n\nZeilen:\n#{rows.join("\n")}"
end
# RFC 5322 with a base64 body, the subject RFC 2047 encoded when it is not ascii
def build subject, body
encoded_subject = subject.ascii_only? ? subject : "=?UTF-8?B?#{[subject].pack('m0')}?="
[
"From: #{@from}",
"To: #{@to.join(', ')}",
"Subject: #{encoded_subject}",
"Date: #{Time.now.strftime('%a, %d %b %Y %H:%M:%S %z')}",
"Message-ID: <#{SecureRandom.uuid}@#{@from.split('@').last}>",
'MIME-Version: 1.0',
'Content-Type: text/plain; charset=UTF-8',
'Content-Transfer-Encoding: base64',
'',
[body].pack('m')
].join("\r\n")
end
end
end
+141
View File
@@ -0,0 +1,141 @@
require 'json'
require 'securerandom'
require 'socket'
class DataPipeline
#
# what a notifier gets: the stats of one or more runs (one script), with event, title and text lines
#
# event: run.failed (an abort), run.rejected (rejected rows), run.warnings, run.succeeded
# to_h: stable payload for webhooks (schema 1); rows of stats.errors only with include_rows: true,
# because they may hold customer data
#
class Report
EVENTS = %w[run.succeeded run.warnings run.rejected run.failed].freeze
MAX_ERRORS = 20
attr_reader :id, :title, :runs, :params, :host, :error
# error: an exception outside of a run (e.g. the script failed before or between pipelines)
def initialize runs, title: nil, params: {}, error: nil
@id = SecureRandom.uuid
@runs = Array(runs)
@title = title || @runs.filter_map(&:name).uniq.join(', ').then { |t| t.empty? ? 'datapipeline' : t }
@params = params.to_h
@host = Socket.gethostname
@error = error
end
# the worst event of all runs
def event
return 'run.failed' if @error
@runs.map { |stats| event_of(stats) }.max_by { |event| EVENTS.index(event) } || 'run.succeeded'
end
def failed? = event == 'run.failed'
def ok? = event == 'run.succeeded'
def status
{ 'run.failed' => 'abgebrochen', 'run.rejected' => 'mit verworfenen Zeilen',
'run.warnings' => 'mit Warnungen', 'run.succeeded' => 'ok' }.fetch(event)
end
def subject = "#{@title}: #{status}"
# text for mail, slack and logs: one block per run, the first errors without row data
def lines errors: 5
lines = ["#{subject} (#{@host})"]
lines << "Parameter: #{@params.map { |key, value| "#{key}=#{value}" }.join(' ')}" if @params.any?
lines << "Fehler: #{@error.class}: #{@error.message}" if @error
@runs.each do |stats|
lines << ''
lines << "#{stats.name || 'Pipeline'}#{" #{stats.version}" if stats.version}: #{stats}"
lines << "Start #{stats.started_at&.strftime('%d.%m.%Y %H:%M:%S')}, Dauer #{stats.duration}s"
stats.errors.first(errors).each { |e| lines << "- #{e[:step]}: #{e[:error]}" }
lines << "- … und #{stats.rejected - errors} weitere" if stats.rejected > errors
end
lines
end
def to_s = lines.join("\n")
def to_h include_rows: false
{
schema: 1, id: @id, event:, title: @title, host: @host, params: @params,
error: @error && "#{@error.class}: #{@error.message}",
runs: @runs.map do |stats|
stats.to_h.merge(errors: stats.errors.first(MAX_ERRORS).map { |e| include_rows ? e : e.except(:row) })
end
}
end
private
def event_of stats
return 'run.failed' if stats.error
return 'run.rejected' if stats.rejected.positive?
return 'run.warnings' if stats.warnings.to_i.positive?
'run.succeeded'
end
end
#
# a notifier only needs deliver(report, include_rows: false); see Webhook, Slack, Email
#
# notify DataPipeline::Webhook.new(url: ...), on: :problems
#
class Notifier
# when a notification is sent
WHEN = {
always: ->(_report) { true },
failure: ->(report) { report.failed? },
problems: ->(report) { !report.ok? }
}.freeze
TYPES = { webhook: 'DataPipeline::Webhook', slack: 'DataPipeline::Slack', email: 'DataPipeline::Email' }.freeze
# notifier from the connections file: type webhook, slack or email, the other keys are its options
def self.from_connection name
options = Connections.fetch(name)
type = options[:type].to_s.to_sym
klass = TYPES.fetch(type) { raise ArgumentError, "connection #{name.inspect}: type #{type} is no notifier (#{TYPES.keys.join(', ')})" }
Object.const_get(klass).new(**options.except(:type))
end
def self.condition on
WHEN.fetch(on.to_sym) { raise ArgumentError, "notify on: #{on} unknown, use #{WHEN.keys.join(', ')}" }
end
def deliver(_report, include_rows: false)
raise NotImplementedError, "#{self.class}#deliver is missing"
end
private
# POST with retries on 429/5xx (Retry-After honoured) and network errors; other 4xx raise at once
def post_with_retries url, body:, headers:, timeout:, retries:, retry_wait:, error:
attempt = 0
loop do
attempt += 1
begin
response = HTTP.timeout(timeout).headers(headers).post(url, body:)
rescue HTTP::Error, SystemCallError, IOError => e
raise error, "#{url}: #{e.class}: #{e.message}" if attempt > retries
sleep(retry_wait * attempt)
next
end
return response if response.status.success?
code = response.code
unless attempt <= retries && (code == 429 || code >= 500)
raise error, "#{url}: HTTP #{code} #{response.body.to_s[0, 200]}"
end
sleep(response.headers['Retry-After']&.to_f || retry_wait * attempt)
end
end
end
end
+66 -12
View File
@@ -1,34 +1,88 @@
require 'http'
require 'json'
require_relative 'notifier'
require_relative '../storage/storage_plugin'
class DataPipeline
#
# Slack.new(channel: '#etl').send_message('posting export done')
# notify DataPipeline::Slack.new(channel: '#etl'), on: :problems
# notify DataPipeline::Slack.new(webhook: ENV['SLACK_WEBHOOK']) # incoming webhook, no token needed
#
# options
# token: bot token, default ENV['SLACK_TOKEN']
# channel: default '#general'
# token: bot token for chat.postMessage, default ENV['SLACK_TOKEN'] (not needed with webhook)
# channel: default '#general' (with webhook the channel of the webhook)
# webhook: url of an incoming webhook instead of token and channel
# mention: e.g. '<!here>' or '<!subteam^ID>', only in reports of failed runs
# timeout: seconds per request (default 10)
# retries: repetitions on 429, 5xx and network errors (default 3)
# retry_wait: seconds, multiplied by the attempt (default 2)
# api_url: default https://slack.com/api/chat.postMessage
#
class Slack
class Slack < Notifier
API_URL = 'https://slack.com/api/chat.postMessage'.freeze
MAX_TEXT = 2900 # a section block holds 3000 characters
class Error < StandardError; end
DEFAULTS = {
token: nil,
channel: '#general'
channel: '#general',
webhook: nil,
mention: nil,
timeout: 10,
retries: 3,
retry_wait: 2,
api_url: API_URL
}.freeze
ICONS = { 'run.failed' => ':red_circle:', 'run.rejected' => ':large_orange_circle:',
'run.warnings' => ':large_yellow_circle:', 'run.succeeded' => ':large_green_circle:' }.freeze
# & < > have a meaning in slack texts
def self.escape(text) = text.to_s.gsub('&', '&amp;').gsub('<', '&lt;').gsub('>', '&gt;')
def initialize options = {}
StoragePlugin.require_gem 'http'
@options = DEFAULTS.merge(token: ENV['SLACK_TOKEN']).merge(options)
raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN)' unless @options[:token]
return if @options[:webhook] || @options[:token]
raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN, or webhook:)'
end
# returns the parsed slack response, raises when slack reports an error
# plain text (slack markup allowed); returns the parsed slack response, raises when slack reports an error
def send_message(content, verbose: false)
response = HTTP.auth("Bearer #{@options[:token]}")
.post(API_URL, json: { channel: @options[:channel], text: content })
body = JSON.parse(response.body.to_s)
puts JSON.pretty_generate(body) if verbose
raise "Slack#send_message failed: #{body['error']}" unless body['ok']
post({ text: content }, verbose:)
end
# the report as blocks: status line, then the runs as preformatted text
def deliver report, include_rows: false
icon = ICONS.fetch(report.event)
mention = report.failed? && @options[:mention] ? " #{@options[:mention]}" : ''
head = "#{icon} *#{self.class.escape(report.title)}*: #{report.status}#{mention}"
details = self.class.escape(report.lines.drop(1).join("\n").strip)
details = "#{details[0, MAX_TEXT]}\n…" if details.length > MAX_TEXT
blocks = [{ type: 'section', text: { type: 'mrkdwn', text: head } }]
blocks << { type: 'section', text: { type: 'mrkdwn', text: "```#{details}```" } } unless details.empty?
post({ text: "#{report.subject}#{mention}", blocks: })
end
private
def post payload, verbose: false
url = @options[:webhook] || @options[:api_url]
headers = { 'Content-Type' => 'application/json; charset=utf-8' }
unless @options[:webhook]
headers['Authorization'] = "Bearer #{@options[:token]}"
payload = payload.merge(channel: @options[:channel])
end
response = post_with_retries(url, body: JSON.generate(payload), headers:, timeout: @options[:timeout],
retries: @options[:retries], retry_wait: @options[:retry_wait], error: Error)
text = response.body.to_s
puts text if verbose
return text if @options[:webhook] # incoming webhooks answer "ok"
body = JSON.parse(text)
raise Error, "Slack#send_message failed: #{body['error']}" unless body['ok']
body
end
+63
View File
@@ -0,0 +1,63 @@
require 'openssl'
require_relative 'notifier'
require_relative '../storage/storage_plugin'
class DataPipeline
#
# POST of the report as json (Report#to_h, schema 1)
#
# notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET'])
#
# options
# url: target (required)
# secret: signs the body: X-DataPipeline-Signature: sha256=HMAC_SHA256(secret, "<timestamp>.<body>")
# with X-DataPipeline-Timestamp (unix seconds)
# headers: more headers, e.g. { 'Authorization' => 'Bearer ...' }
# timeout: seconds per request (default 10)
# retries: repetitions on 429, 5xx and network errors (default 3), Retry-After is honoured
# retry_wait: seconds, multiplied by the attempt (default 2)
#
# every request carries X-DataPipeline-Event and X-DataPipeline-Delivery (the report id, the same on retries)
#
class Webhook < Notifier
class Error < StandardError; end
def initialize url:, secret: nil, headers: {}, timeout: 10, retries: 3, retry_wait: 2
StoragePlugin.require_gem 'http'
@url = url.to_s
raise ArgumentError, 'Webhook: url missing' if @url.empty?
@secret = secret
@headers = headers.to_h { |key, value| [key.to_s, value.to_s] }
@timeout = timeout
@retries = retries
@retry_wait = retry_wait
end
def deliver report, include_rows: false
body = JSON.generate(report.to_h(include_rows:))
post_with_retries(@url, body:, headers: headers_for(report, body), timeout: @timeout, retries: @retries,
retry_wait: @retry_wait, error: Error)
end
# for receivers: true when signature and timestamp (max_age seconds) of a request are valid
def self.valid_signature? secret:, body:, timestamp:, signature:, max_age: 300
return false if timestamp.to_s.empty? || (Time.now.to_i - timestamp.to_i).abs > max_age
expected = "sha256=#{OpenSSL::HMAC.hexdigest('SHA256', secret, "#{timestamp}.#{body}")}"
expected.bytesize == signature.to_s.bytesize && OpenSSL.fixed_length_secure_compare(expected, signature.to_s)
end
private
def headers_for report, body
headers = { 'Content-Type' => 'application/json', 'User-Agent' => "datapipeline/#{VERSION}",
'X-DataPipeline-Event' => report.event, 'X-DataPipeline-Delivery' => report.id }.merge(@headers)
return headers unless @secret
timestamp = Time.now.to_i.to_s
signature = OpenSSL::HMAC.hexdigest('SHA256', @secret, "#{timestamp}.#{body}")
headers.merge('X-DataPipeline-Timestamp' => timestamp, 'X-DataPipeline-Signature' => "sha256=#{signature}")
end
end
end
+30 -4
View File
@@ -10,7 +10,12 @@ class DataPipeline
# columns: fixed column order for writing (and keys for reading without headers)
# without columns the values are written in hash order
# layout: DataPipeline::Layout, field order per record type (several record types in one file),
# unknown fields raise; no header line
# unknown fields raise; no header line. reading with headers: false: keys from the fields
# of the record type
# encoding: encoding of the file, e.g. 'ISO-8859-1' or 'Windows-1252' (default UTF-8);
# read as UTF-8 strings, written transcoded from UTF-8; a UTF-8 BOM is skipped when reading
# replace: '?' -> characters the encoding cannot hold are replaced when writing (default: error)
# bom: true -> write a UTF-8 BOM (Excel)
# counter: print a dot per written row
# verbose: print settings
#
@@ -44,11 +49,11 @@ class DataPipeline
def read(&block)
if block_given?
CSV.foreach(@file_name, **csv_options) { |row| yield to_hash(row) }
CSV.foreach(@file_name, **read_options) { |row| yield to_hash(row) }
return
end
CSV.read(@file_name, **csv_options).map { |row| to_hash(row) }
CSV.read(@file_name, **read_options).map { |row| to_hash(row) }
end
def write data
@@ -93,10 +98,30 @@ class DataPipeline
@options[:columns]
end
def utf8?
@options[:encoding].nil? || Encoding.find(@options[:encoding]) == Encoding::UTF_8
end
# csv: { encoding: ... } is passed unchanged (older scripts)
def read_options
return csv_options if csv_options.key?(:encoding)
csv_options.merge(encoding: utf8? ? 'bom|utf-8' : "#{@options[:encoding]}:utf-8")
end
def write_options
return csv_options if csv_options.key?(:encoding) || utf8?
options = csv_options.merge(encoding: @options[:encoding])
options.merge!(undef: :replace, invalid: :replace, replace: @options[:replace]) if @options[:replace]
options
end
def open_file
return if @file
@file = CSV.open(@file_name, 'w', **csv_options)
@file = CSV.open(@file_name, 'w', **write_options)
@file.to_io.write("\uFEFF") if @options[:bom] && utf8?
@header_written = false
return unless columns && csv_options[:headers] && !@options[:layout]
@@ -107,6 +132,7 @@ class DataPipeline
# rows with headers are CSV::Row, without headers plain arrays
def to_hash row
return row.to_h if row.is_a?(CSV::Row)
return @options[:layout].parse(row) if @options[:layout]
keys = columns&.map(&:to_s) || (0...row.length).map(&:to_s)
keys.zip(row).to_h
+77
View File
@@ -0,0 +1,77 @@
require_relative 'storage_plugin'
class DataPipeline
#
# fixed width file, one line per row, field widths from a DataPipeline::Layout
#
# options
# file: path of the file
# layout: DataPipeline::Layout, every AN and N field needs a length
# encoding: encoding of the file (default UTF-8), read as UTF-8, written transcoded
# replace: '?' -> characters the encoding cannot hold are replaced when writing (default: error)
# line_end: "\n" (default) or "\r\n"
#
# writing: values formatted as with Layout format: true, text left aligned with blanks, numbers right
# aligned with zeros (sign in front), empty fields blank; a value longer than its field raises LayoutError
# reading: record type from the type field, keys are the field names in lower case, values without padding,
# blank fields nil
#
class FixedStorage < StoragePlugin
def initialize **options, &block
@options = options
@file_name = options.fetch(:file) { raise ArgumentError, 'FixedStorage: option file missing' }
@layout = options.fetch(:layout) { raise ArgumentError, 'FixedStorage: option layout missing' }.check_widths!
@line_end = options.fetch(:line_end, "\n")
yield if block_given?
end
def read(&block)
return each_line.map { |line| @layout.parse_line(line) } unless block_given?
each_line { |line| yield @layout.parse_line(line) }
nil
end
def write data
data.each { |row| write_row row }
ensure
close
end
def write_row row
@file ||= File.open(@file_name, 'w', **write_options)
@file.write(@layout.line(row), @line_end)
end
def close
@file&.close
@file = nil
end
private
def utf8?
@options[:encoding].nil? || Encoding.find(@options[:encoding]) == Encoding::UTF_8
end
# empty lines are skipped
def each_line
return enum_for(:each_line) unless block_given?
encoding = utf8? ? 'bom|utf-8' : "#{@options[:encoding]}:utf-8"
File.foreach(@file_name, encoding:) do |line|
line = line.chomp
yield line unless line.strip.empty?
end
end
def write_options
return { encoding: 'utf-8' } if utf8?
options = { encoding: @options[:encoding] }
options.merge!(undef: :replace, invalid: :replace, replace: @options[:replace]) if @options[:replace]
options
end
end
end
+85 -3
View File
@@ -11,6 +11,9 @@ class DataPipeline
# typed: true -> values come back as ruby types (Integer, Float, Date, ...)
# false (default) -> all values as String, NULL as nil
# fetch_size: rows per fetch when reading with a block (default 1000)
# upsert: %w[belnr] -> insert ... on conflict (belnr) do update; needs a unique index on these columns
# batch_size: write_row collects rows and writes them in one transaction per batch (default 1000);
# :all -> one transaction for the whole stream, committed by finalize, rolled back by close
#
class PostgresStorage < StoragePlugin
def initialize **options, &block
@@ -45,9 +48,12 @@ class DataPipeline
each_row(@options[:query], params, &block)
end
# pending rows of write_row are written first, so a query sees them
def query options = {}
raise "Query for PostgresStorage:query missing" unless options[:query]
flush
result = @conn.exec_params(options[:query], options.fetch(:params, []))
return result.to_a unless block_given?
@@ -65,25 +71,88 @@ class DataPipeline
table_name, data = data.nil? ? [@options[:table], table_or_data] : [table_or_data, data]
raise "Table for PostgresStorage#write missing" unless table_name
@conn.transaction do
transaction do
truncate_once table_name
data.each { |row| insert table_name, row }
end
end
# rows are collected and written per batch_size in one transaction
def write_row row
raise "Table for PostgresStorage#write_row missing" unless @options[:table]
if batch_size == :all
begin_stream
truncate_once @options[:table]
insert @options[:table], row
else
(@pending ||= []) << row
flush if @pending.size >= batch_size
end
end
# end of a stream: remaining rows, commit of batch_size: :all
def finalize
flush
if @stream_open
@conn.exec('commit')
@stream_open = false
end
close
end
# pending rows are written; an open stream with batch_size: :all is rolled back (run aborted)
# an error while flushing does not hide the error that aborted the run
def close
@conn.close if @conn && !@conn.finished?
return unless @conn && !@conn.finished?
aborted_by = $!
begin
flush
rescue StandardError
raise unless aborted_by
end
@conn.exec('rollback') if @stream_open
@stream_open = false
@conn.close
end
private
def batch_size
size = @options.fetch(:batch_size, 1000)
size == :all ? :all : Integer(size)
end
def flush
return if @pending.nil? || @pending.empty?
rows = @pending
@pending = []
transaction do
truncate_once @options[:table]
rows.each { |row| insert @options[:table], row }
end
end
def begin_stream
return if @stream_open
raise 'PostgresStorage: batch_size: :all needs its own connection (no open transaction)' unless idle?
@conn.exec('begin')
@stream_open = true
end
# inside a transaction (e.g. a cursor of the same connection) no own transaction is opened
def transaction(&block)
idle? ? @conn.transaction(&block) : yield
end
def idle?
@conn.transaction_status == PG::PQTRANS_IDLE
end
# rows are fetched in batches inside a transaction; other queries on the
# same connection (lookups, subqueries) are possible while reading
def each_row sql, params
@@ -112,11 +181,24 @@ class DataPipeline
columns = row.keys.map { |col| @conn.quote_ident(col.to_s) }.join(',')
placeholders = (1..row.size).map { |i| "$#{i}" }.join(',')
@conn.exec_params(
"insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})",
"insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})#{upsert_clause(row)}",
row.values
)
end
# on conflict (keys) do update set <other columns> = excluded.<column>
def upsert_clause row
keys = Array(@options[:upsert]).map(&:to_s)
return '' if keys.empty?
others = row.keys.map(&:to_s) - keys
target = keys.map { |key| @conn.quote_ident(key) }.join(',')
return " on conflict (#{target}) do nothing" if others.empty?
updates = others.map { |col| "#{@conn.quote_ident(col)} = excluded.#{@conn.quote_ident(col)}" }.join(', ')
" on conflict (#{target}) do update set #{updates}"
end
# schema.table -> "schema"."table"
def quote_table name
name.to_s.split('.').map { |part| @conn.quote_ident(part) }.join('.')
+90 -2
View File
@@ -10,6 +10,10 @@ class DataPipeline
# create: true -> create the table on the first write, columns from the first row
# (Integer -> integer, Float -> real, else text)
# truncate: true -> delete all rows of the table before the first write
# upsert: %w[belnr] -> insert ... on conflict (belnr) do update; with create: true the unique index
# is created too, otherwise the table needs one on these columns
# batch_size: write_row collects rows and writes them in one transaction per batch (default 1000);
# :all -> one transaction for the whole stream, committed by finalize, rolled back by close
#
# rows come back as Hash<String, value> with sqlite types (Integer, Float, String, nil).
# the database file is opened on first use: a dry run creates no file.
@@ -31,9 +35,11 @@ class DataPipeline
query(query: @options[:query], params: @options.fetch(:params, []), &block)
end
# pending rows of write_row are written first, so a query sees them
def query options = {}
raise 'Query for SqliteStorage:query missing' unless options[:query]
flush
statement = db.prepare(options[:query])
result = statement.execute(*bind_args(options.fetch(:params, [])))
return result.map(&:to_h) unless block_given?
@@ -56,20 +62,47 @@ class DataPipeline
return if data.empty?
db.transaction do
transaction do
prepare_table table_name, data.first
data.each { |row| insert table_name, row }
end
end
# rows are collected and written per batch_size in one transaction
def write_row row
raise 'Table for SqliteStorage#write_row missing' unless @options[:table]
if batch_size == :all
begin_stream
prepare_table @options[:table], row
insert @options[:table], row
else
(@pending ||= []) << row
flush if @pending.size >= batch_size
end
end
# end of a stream: remaining rows, commit of batch_size: :all
def finalize
flush
if @stream_open
db.commit
@stream_open = false
end
close
end
# pending rows are written; an open stream with batch_size: :all is rolled back (run aborted);
# an error while flushing does not hide the error that aborted the run
def close
aborted_by = $!
begin
flush
rescue StandardError
raise unless aborted_by
end
@db.rollback if @stream_open && @db.transaction_active?
@stream_open = false
@db.close if @db && !@db.closed?
end
@@ -79,6 +112,38 @@ class DataPipeline
@db ||= SQLite3::Database.new(@database).tap { |d| d.results_as_hash = true }
end
def batch_size
size = @options.fetch(:batch_size, 1000)
size == :all ? :all : Integer(size)
end
def flush
return if @pending.nil? || @pending.empty?
rows = @pending
@pending = []
transaction do
rows.each do |row|
prepare_table @options[:table], row
insert @options[:table], row
end
end
end
def begin_stream
return if @stream_open
raise 'SqliteStorage: batch_size: :all needs its own connection (no open transaction)' if db.transaction_active?
db.transaction
@stream_open = true
end
# inside an open transaction no own transaction is opened
def transaction(&block)
db.transaction_active? ? yield : db.transaction(&block)
end
# create / truncate once per table, before its first row
def prepare_table table_name, row
return if @prepared[table_name]
@@ -86,6 +151,7 @@ class DataPipeline
if @options[:create]
columns = row.map { |key, value| "#{quote(key)} #{column_type(value)}" }.join(', ')
db.execute("create table if not exists #{quote_table(table_name)} (#{columns})")
create_upsert_index table_name if upsert_keys.any?
end
db.execute("delete from #{quote_table(table_name)}") if @options[:truncate]
@prepared[table_name] = true
@@ -108,7 +174,29 @@ class DataPipeline
columns = row.keys.map { |col| quote(col) }.join(',')
placeholders = (['?'] * row.size).join(',')
values = row.values.map { |v| v.is_a?(Date) || v.is_a?(Time) ? v.iso8601 : v }
db.execute("insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})", values)
db.execute("insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})#{upsert_clause(row)}", values)
end
def upsert_keys
Array(@options[:upsert]).map(&:to_s)
end
# on conflict (keys) do update set <other columns> = excluded.<column>
def upsert_clause row
return '' if upsert_keys.empty?
others = row.keys.map(&:to_s) - upsert_keys
target = upsert_keys.map { |key| quote(key) }.join(',')
return " on conflict (#{target}) do nothing" if others.empty?
" on conflict (#{target}) do update set #{others.map { |col| "#{quote(col)} = excluded.#{quote(col)}" }.join(', ')}"
end
# schema.table -> index schema.table_upsert on table (keys)
def create_upsert_index table_name
*schema, table = table_name.to_s.split('.')
index = [*schema, "#{table}_upsert"].map { |part| quote(part) }.join('.')
db.execute("create unique index if not exists #{index} on #{quote(table)} (#{upsert_keys.map { |k| quote(k) }.join(',')})")
end
def quote name
+1 -1
View File
@@ -1,5 +1,5 @@
class DataPipeline
VERSION = '2.0.0'.freeze
VERSION = '3.2.0'.freeze
# kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md
class Version
+86 -4
View File
@@ -1,5 +1,6 @@
require_relative '../test_helper'
require 'open3'
require_relative '../support/test_http_server'
# bin/etl as a subprocess: arguments, params, dry run, exit codes
class EtlCliTest < PipelineTest
@@ -14,8 +15,8 @@ class EtlCliTest < PipelineTest
"#{dir}/#{name}"
end
# like import.*.rb: own OptionParser for --from/--till, transform api
def test_legacy_script_gets_from_till_and_limit
# scripts with an own OptionParser for --from/--till
def test_script_with_own_option_parser_gets_from_till_and_limit
with_tmpdir do |dir|
path = script(dir, 'legacy.rb', <<~RUBY)
require 'optparse/date'
@@ -24,8 +25,8 @@ class EtlCliTest < PipelineTest
File.write('args.txt', "\#{opts[:from]} \#{opts[:till]}")
DataPipeline.new do
read container(:csv, file: 'in.csv')
transform(processing: :prepend) { |_, _| { 'h' => 'HEAD' } }
transform { |r, i| r.merge('i' => i) }
header { { 'h' => 'HEAD' } }
map { |r, ctx| r.merge('i' => ctx.index) }
write container(:csv, file: 'out.csv', csv: { headers: false, col_sep: '|' })
end.run
RUBY
@@ -86,4 +87,85 @@ class EtlCliTest < PipelineTest
assert_includes err, 'Usage: etl [run] PIPELINE.rb'
end
end
def test_lock_exit_code_3
with_tmpdir do |dir|
path = script(dir, 'locked.rb', <<~RUBY)
DataPipeline.new do
lock '.pipeline.lock'
read container(:csv, file: 'in.csv')
write container(:csv, file: 'out.csv')
end.run
RUBY
holder = DataPipeline::Lock.new("#{dir}/.script.lock").acquire
_out, err, status = etl(dir, 'run', path, '--lock=.script.lock')
assert_equal 3, status.exitstatus, err
assert_includes err, 'another run is still active'
refute File.exist?("#{dir}/out.csv")
holder.release
holder = DataPipeline::Lock.new("#{dir}/.pipeline.lock").acquire
_out, err, status = etl(dir, 'run', path, '--lock=.script.lock')
assert_equal 3, status.exitstatus, err
holder.release
_out, err, status = etl(dir, 'run', path, '--lock=.script.lock')
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
def test_notify_one_report_for_all_pipelines_of_the_script
server = TestHttpServer.new { [200, 'ok'] }
with_tmpdir do |dir|
Dir.mkdir("#{dir}/config")
File.write("#{dir}/config/connections.yml", "alerts:\n type: webhook\n url: #{server.url}/hook\n")
path = script(dir, 'zwei.rb', <<~RUBY)
2.times do |i|
DataPipeline.new do
name "teil\#{i}"
on_error :skip
read container(:csv, file: 'in.csv')
validate 'a' => { format: /\\A\\d+\\z/ } if i == 1
write container(:csv, file: "out\#{i}.csv")
end.run
end
RUBY
_out, err, status = etl(dir, 'run', path, '--notify=alerts')
assert_equal 2, status.exitstatus, err
assert_equal 1, server.requests.size
payload = JSON.parse(server.requests.last[:body])
assert_equal ['run.rejected', 'zwei', %w[teil0 teil1]], [payload['event'], payload['title'], payload['runs'].map { |r| r['name'] }]
_out, err, status = etl(dir, 'run', path, '--notify=alerts', '--notify-on=failure')
assert_equal 2, status.exitstatus, err
assert_equal 1, server.requests.size # nothing failed
broken = script(dir, 'kaputt.rb', "raise 'Skriptfehler'\n")
_out, _err, status = etl(dir, 'run', broken, '--notify=alerts')
assert_equal 1, status.exitstatus
payload = JSON.parse(server.requests.last[:body])
assert_equal ['run.failed', 'RuntimeError: Skriptfehler', []], payload.values_at('event', 'error', 'runs')
_out, err, status = etl(dir, 'run', path, '--notify=gibtsnicht')
assert_equal 1, status.exitstatus
assert_includes err, 'etl: --notify=gibtsnicht: connection "gibtsnicht" not in config/connections.yml (alerts)'
end
ensure
server&.stop
end
end
+21 -15
View File
@@ -4,16 +4,22 @@ require_relative '../test_helper'
class ChainedTest < PipelineTest
ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
def test_chained_transform_gets_previous_output
out = output(ROWS) { chained; transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
def test_each_step_gets_the_previous_output
out = output(ROWS) { map { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 2) } }
assert_equal ROWS.map { |r| r.merge('b' => 1, 'c' => 2) }, out
end
def test_one_to_many_then_index
out = output(ROWS) { chained; transform { |r| [r, r] }; transform { |r, i| r.merge('i' => i) } }
out = output(ROWS) { map { |r| [r, r] }; map { |r, ctx| r.merge('i' => ctx.index) } }
assert_equal (0...6).to_a, out.map { |r| r['i'] }
end
def test_nil_and_empty_array_drop_rows
assert_equal [{ 'a' => 1 }], output(ROWS) { map { |r| r['a'] == 1 ? r : nil } }
assert_equal ROWS, output(ROWS) { map { |r| [r, nil] } }
assert_equal [], output(ROWS) { map { [] } }
end
def test_map_and_filter
out = output(ROWS) do
map { |r| r.merge('d' => r['a'] * 2) }
@@ -23,12 +29,6 @@ class ChainedTest < PipelineTest
assert_equal [{ 'a' => 2, 'd' => 4, 'i' => 0 }, { 'a' => 3, 'd' => 6, 'i' => 1 }], out
end
def test_map_turns_chaining_on_for_transform
out = output(ROWS) { transform { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 1) } }
assert_equal 3, out.size
assert(out.all? { |r| r['b'] && r['c'] })
end
def test_header_and_trailer_with_context
out = output(ROWS) do
header { |c| { 'H' => c.read, 'W' => c.written } }
@@ -41,9 +41,14 @@ class ChainedTest < PipelineTest
assert_equal 6, out.size
end
def test_edge_order_follows_registration
out = output(ROWS) { header { { 'h' => 1 } }; transform(processing: :prepend) { { 'p' => 1 } }; map { |r| r } }
assert_equal [{ 'h' => 1 }, { 'p' => 1 }], out.first(2)
def test_several_headers_and_trailers_in_registration_order
out = output(ROWS) { header { { 'h' => 1 } }; trailer { { 't' => 1 } }; header { { 'h' => 2 } }; trailer { { 't' => 2 } } }
assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
end
def test_stable_order_of_many_steps
out = output(ROWS) { 20.times { |k| map { |r| r.merge("k#{k}" => k) } } }
assert_equal (0...20).to_a, out.first.values.drop(1)
end
def test_header_and_trailer_nil_and_arrays
@@ -51,12 +56,13 @@ class ChainedTest < PipelineTest
assert_equal ROWS + [{ 'x' => 1 }, { 'y' => 2 }], out
end
def test_condition_on_map
def test_condition_on_map_and_filter
assert_equal [{ 'a' => 1 }], output(ROWS) { map(condition: ->(r) { r['a'] == 1 }) { |r| r } }
assert_equal [{ 'a' => 3 }], output(ROWS) { filter(condition: ->(r) { r['a'] > 1 }) { |r| r['a'] == 3 } }
end
def test_chained_false
assert_equal 6, output(ROWS) { chained false; transform { |r| r }; transform { |r| r } }.size
def test_condition_must_be_a_lambda
assert_includes assert_raises(ArgumentError) { output(ROWS) { map(condition: "row['a'] == 1") { |r| r } } }.message, 'lambda'
end
def test_csv_end_to_end
-81
View File
@@ -1,81 +0,0 @@
require_relative '../test_helper'
#
# parallel mode (transform without chained/map/filter) must keep the output of v1.2.1,
# the version the CUDO importers were built on. expected values were recorded with v1.2.1.
#
class CompatTest < PipelineTest
ROWS = (1..6).map { |i| { 'id' => i, 'land' => i.even? ? 'DE' : 'AT' } }
def assert_output expected_json, &block
assert_equal expected_json, JSON.generate(output(ROWS, &block))
end
# pattern of import.customer.rb
def test_customer_pattern
assert_output '[{"SATYP":"FIRC","LFDNR":0},{"id":1,"land":"AT","lfdnr":"0000000001"},' \
'{"id":2,"land":"DE","lfdnr":"0000000002"},{"id":3,"land":"AT","lfdnr":"0000000003"},' \
'{"id":4,"land":"DE","lfdnr":"0000000004"},{"id":5,"land":"AT","lfdnr":"0000000005"},' \
'{"id":6,"land":"DE","lfdnr":"0000000006"},{"topic":"LARC","counter":"0000000006"}]' do
transform(processing: :prepend) { |_row, n| { 'SATYP': 'FIRC', 'LFDNR': n } }
transform(processing: :append) { |_row, n| { 'topic': 'LARC', 'counter': format('%010d', n) } }
transform { |row, _i| row.merge('lfdnr' => format('%010d', row['id'])) }
end
end
# pattern of import.posting.rb: closure counter, 1:n
def test_posting_pattern
expected = [{ SATYP: 'FIRC', LFDNR: 0 }]
lfdnr = 0
ROWS.each do |row|
expected << row.merge('lfdnr' => lfdnr += 1)
row['id'].times { |k| expected << { 'pos' => k, 'lfdnr' => lfdnr += 1 } }
end
expected << { SATYPE: 'LARC', SUMME: '0000000027' }
assert_output JSON.generate(expected) do
cnt = 1
transform(processing: :prepend) { |_row, _n| { 'SATYP': 'FIRC', 'LFDNR': 0 } }
transform do |row|
out = [row.merge('lfdnr' => cnt)]
cnt += 1
row['id'].times do |k|
out << { 'pos' => k, 'lfdnr' => cnt }
cnt += 1
end
out
end
transform(processing: :append) { |_row| { 'SATYPE': 'LARC', 'SUMME': format('%010d', cnt - 1) } }
end
end
def test_append_registered_before_prepend
assert_output '[{"h":0},{"id":1,"land":"AT"},{"id":2,"land":"DE"},{"id":3,"land":"AT"},' \
'{"id":4,"land":"DE"},{"id":5,"land":"AT"},{"id":6,"land":"DE"},{"t":6}]' do
transform(processing: :append) { |_, n| { 't' => n } }
transform { |r| r }
transform(processing: :prepend) { |_, n| { 'h' => n } }
end
end
def test_condition_string
assert_output '[{"id":2,"land":"DE","i":1},{"id":4,"land":"DE","i":3},{"id":6,"land":"DE","i":5}]' do
transform(condition: "row['land'] == 'DE'") { |r, i| r.merge('i' => i) }
end
end
def test_two_parallel_transforms
expected = ROWS.map { |r| r.merge('a' => 1) } + ROWS.map { |r| r.merge('b' => 1) }
assert_output JSON.generate(expected) do
transform { |r| r.merge('a' => 1) }
transform { |r| [r.merge('b' => 1), nil] }
end
end
def test_nil_and_empty_array
assert_output '[{"id":1,"land":"AT"}]' do
transform { |r| r['id'] > 3 ? nil : [] }
transform { |r| r['id'] == 1 ? r : nil }
end
end
end
@@ -0,0 +1,98 @@
require_relative '../test_helper'
require 'time'
# 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
+50
View File
@@ -71,4 +71,54 @@ class LayoutTest < PipelineTest
assert_equal [{ 'a' => '1' }], target.output
assert_equal 1, stats.rejected
end
FORMATTED = DataPipeline::Layout.new(format: true) do
record 'K', [['SATYP', 'AN', 2, true], ['DATUM', 'DAT'], ['ZEIT', 'TIM'], ['BETRAG', 'N', '11.2'], ['NR', 'N', 5], ['TEXT', 'AN', 6]]
end
def test_format_values_by_type
row = { 'satyp' => 'K', 'datum' => Date.new(2026, 10, 9), 'zeit' => Time.new(2026, 10, 9, 7, 5, 3), 'betrag' => 12.345, 'nr' => 7, 'text' => 'abc' }
assert_equal ['K', '20261009', '070503', '12.35', '7', 'abc'], FORMATTED.values(row)
assert_equal ['K', '20261009', '070503', '3.50', nil, nil], FORMATTED.values({ 'satyp' => 'K', 'datum' => '2026-10-09', 'zeit' => '07:05:03', 'betrag' => '3.5' })
assert_equal '-0.01', FORMATTED.values({ 'satyp' => 'K', 'betrag' => -0.005 })[3]
assert_equal '0.33', FORMATTED.values({ 'satyp' => 'K', 'betrag' => Rational(1, 3) })[3]
assert_equal '1.234', FORMATTED.values({ 'satyp' => 'K', 'betrag' => '1.234' })[3] # texts are not rounded
assert_equal ['K', nil, nil, nil, nil, nil], FORMATTED.values({ 'satyp' => 'K' })
assert_equal [nil, 12.345], [LAYOUT.values({ 'satyp' => 'HEAD' })[1], DataPipeline::Layout.new { record 'K', [['S'], ['B', 'N', '5.2']] }.values({ 's' => 'K', 'b' => 12.345 })[1]]
end
def test_errors_check_the_formatted_values
assert_empty FORMATTED.errors({ 'satyp' => 'K', 'datum' => Date.today, 'betrag' => 0.1 + 0.2 })
assert_equal ['BETRAG: more than 2 decimals'], DataPipeline::Layout.new { record 'K', [['S'], ['BETRAG', 'N', '11.2']] }.errors({ 's' => 'K', 'betrag' => 0.1 + 0.2 })
end
def test_fixed_line_and_parse_line
row = { 'satyp' => 'K', 'datum' => Date.new(2026, 10, 9), 'betrag' => -12.5, 'nr' => 7, 'text' => 'Müller' }
line = FORMATTED.line(row)
assert_equal 'K 20261009 -00000012.5000007Müller', line
assert_equal 2 + 8 + 6 + 12 + 5 + 6, line.length
assert_equal({ 'satyp' => 'K', 'datum' => '20261009', 'zeit' => nil, 'betrag' => '-00000012.50', 'nr' => '00007', 'text' => 'Müller' },
FORMATTED.parse_line(line))
assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.line(row.merge('text' => 'zu lang')) }.message, 'longer than 6'
assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.line(row.merge('nr' => 'x')) }.message, 'not a number'
assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.line(row.merge('extra' => 1)) }.message, 'extra'
assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.parse_line('XX') }.message, 'not in layout'
end
def test_check_widths
layout = DataPipeline::Layout.new { record 'A', [['SATYP', 'AN', 1], ['TEXT'], ['NR', 'N'], ['D', 'DAT']] }
assert_equal 'fixed width needs a length for: A.TEXT, A.NR', assert_raises(DataPipeline::LayoutError) { layout.check_widths! }.message
assert_same FORMATTED, FORMATTED.check_widths!
end
def test_parse_values_and_csv_read_with_layout
assert_equal({ 'satyp' => 'POS', 'nr' => '1', 'text' => nil, 'betrag' => nil }, LAYOUT.parse(%w[POS 1]))
assert_includes assert_raises(DataPipeline::LayoutError) { LAYOUT.parse(%w[HEAD 1 2 3]) }.message, '4 values'
with_tmpdir do |dir|
File.write("#{dir}/in.txt", "HEAD|20251103|\nPOS|1|abc|1.50\n")
rows = DataPipeline::CsvStorage.new(file: "#{dir}/in.txt", layout: LAYOUT, csv: { headers: false, col_sep: '|' }).read
assert_equal [{ 'satyp' => 'HEAD', 'datum' => '20251103', 'zeit' => nil },
{ 'satyp' => 'POS', 'nr' => '1', 'text' => 'abc', 'betrag' => '1.50' }], rows
end
end
end
+47
View File
@@ -0,0 +1,47 @@
require_relative '../test_helper'
class LockTest < PipelineTest
def test_second_lock_on_the_same_file_fails
with_tmpdir do |dir|
first = DataPipeline::Lock.new("#{dir}/sub/.lock").acquire
assert first.locked?
error = assert_raises(DataPipeline::LockError) { DataPipeline::Lock.new("#{dir}/sub/.lock").acquire }
assert_includes error.message, "pid #{Process.pid}"
first.release
DataPipeline::Lock.new("#{dir}/sub/.lock").synchronize { assert true }
end
end
def test_pipeline_holds_the_lock_during_run_and_releases_it
with_tmpdir do |dir|
file = "#{dir}/.lock"
inside = nil
run_pipeline([{ 'a' => 1 }]) do
lock file
map { |row| inside = (DataPipeline::Lock.new(file).acquire rescue $!); row }
end
assert_kind_of DataPipeline::LockError, inside
DataPipeline::Lock.new(file).synchronize { assert true } # released after run
end
end
def test_locked_pipeline_aborts_with_lock_error_and_writes_nothing
with_tmpdir do |dir|
holder = DataPipeline::Lock.new("#{dir}/.lock").acquire
target = MemoryStorage.new
finished = nil
assert_raises(DataPipeline::LockError) do
DataPipeline.new do
logger PipelineHelper::QUIET
lock "#{dir}/.lock"
read MemoryStorage.new(rows: [{ 'a' => 1 }])
write target
on_finish { |stats| finished = stats }
end.run
end
assert_nil target.written
assert_kind_of DataPipeline::LockError, finished.error
holder.release
end
end
end
+4 -4
View File
@@ -41,8 +41,8 @@ class OperationsTest < PipelineTest
assert_raises(RuntimeError) { run_pipeline(ROWS) { on_error :skip; header { raise 'h' }; map { |r| r } } }
end
def test_skip_in_parallel_mode
target, stats = run_pipeline(ROWS) { on_error :skip; transform { |r| r['id'] == '1' ? raise('x') : r } }
def test_skip_in_map
target, stats = run_pipeline(ROWS) { on_error :skip; map { |r| r['id'] == '1' ? raise('x') : r } }
assert_equal 4, target.output.size
assert_equal 1, stats.rejected
end
@@ -67,8 +67,8 @@ class OperationsTest < PipelineTest
assert_empty DataPipeline::Validation.errors({ 'a' => '' }, 'a' => { max: 1, type: :integer })
end
def test_validate_turns_chaining_on
target, = run_pipeline(ROWS) { on_error :skip; transform { |r| r.merge('x' => 1) }; validate RULES }
def test_validate_gets_the_output_of_the_previous_step
target, = run_pipeline(ROWS) { on_error :skip; map { |r| r.merge('x' => 1) }; validate RULES }
assert_equal 3, target.output.size
assert(target.output.all? { |r| r['x'] })
end
+3 -40
View File
@@ -1,17 +1,11 @@
require_relative '../test_helper'
# dsl, settings, plugin lookup, parallel mode
# dsl, settings, plugin lookup
class PipelineCoreTest < PipelineTest
ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
def settings_of(pipeline) = pipeline.instance_variable_get(:@settings)
def test_nil_filters_rows
assert_equal [{ 'a' => 1 }], output(ROWS) { transform { |r| r['a'] == 1 ? r : nil } }
assert_equal ROWS, output(ROWS) { transform { |r| [r, nil] } }
assert_equal ROWS, output(ROWS) { transform(processing: :prepend) { nil }; transform { |r| r }; transform(processing: :append) { nil } }
end
def test_dsl_switches
assert_equal false, settings_of(DataPipeline.new { verbose false })[:verbose]
assert_equal true, settings_of(DataPipeline.new { verbose })[:verbose]
@@ -73,39 +67,8 @@ class PipelineCoreTest < PipelineTest
assert_equal Array.new(8) { |i| (i % 3) + 1 }, reads
end
def test_parallel_mode_is_side_by_side
out = output(ROWS) { transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
assert_equal 6, out.size
end
def test_parallel_mode_copies_rows_per_step
out = output(ROWS) { transform { |r| r['x'] = 1; r }; transform { |r| r.key?('x') ? nil : r } }
assert_equal 6, out.size
assert_equal [{ 'a' => 10 }, { 'a' => 20 }, { 'a' => 30 }], output(ROWS) { transform { |r| [r.transform_values! { |v| v * 10 }] } }
end
def test_stable_order_and_all_prepends_appends
out = output(ROWS) { 20.times { |k| transform { |r| r['a'] == 1 ? { 'k' => k } : nil } } }
assert_equal (0...20).to_a, out.map { |r| r['k'] }
out = output(ROWS) do
transform(processing: :prepend) { { 'h' => 1 } }
transform(processing: :prepend) { { 'h' => 2 } }
transform { |r| r }
transform(processing: :append) { { 't' => 1 } }
transform(processing: :append) { { 't' => 2 } }
end
assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
end
def test_append_count_parallel_vs_chained
assert_equal({ 'n' => 3 }, output(ROWS) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
assert_equal({ 'n' => 6 }, output(ROWS) { chained; transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
end
def test_conditions
assert_equal [{ 'a' => 1 }, { 'a' => 3 }], output(ROWS) { transform(condition: ->(r) { r['a'] != 2 }) { |r| r } }
assert_equal [{ 'a' => 2 }], output(ROWS) { transform(condition: "row['a'] == 2") { |r| r } }
def test_removed_dsl_of_version_2
%i[transform chained].each { |name| assert_raises(NoMethodError) { DataPipeline.new { send(name) { |r| r } } } }
end
def test_copy_without_steps
+47
View File
@@ -0,0 +1,47 @@
require_relative '../test_helper'
# rename, select, cast, default
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
end
+5 -14
View File
@@ -21,23 +21,14 @@ class StreamingTest < PipelineTest
assert_equal({ 'T' => 4, 'R' => 3 }, stream.last)
end
def test_stream_prepend_append
def test_stream_several_steps_one_to_many
out = output(ROWS, streamed: true) do
chained
transform(processing: :prepend) { |_, n| { 'p' => n } }
transform { |r| [r, r] }
transform(processing: :append) { |_, n| { 'n' => n } }
map { |r| [r, r] }
map { |r, ctx| r.merge('i' => ctx.index) }
trailer { |ctx| { 'n' => ctx.written } }
end
assert_equal({ 'p' => 0 }, out.first)
assert_equal({ 'n' => 6 }, out.last)
assert_equal 8, out.size
end
def test_stream_parallel_single_step_and_error_for_several
out = output(ROWS, streamed: true) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }
assert_equal({ 'n' => 3 }, out.last)
error = assert_raises(ArgumentError) { output(ROWS, streamed: true) { transform { |r| r }; transform { |r| r } } }
assert_includes error.message, 'chained'
assert_equal (0...6).to_a, out.first(6).map { |r| r['i'] }
end
def test_stream_condition_and_copy
+33 -4
View File
@@ -4,15 +4,15 @@ require 'open3'
require 'fileutils'
#
# the examples of README.md section 1: the code in the documentation is the file,
# the examples of README.md section 2: the code in the documentation is the file,
# and every example runs and produces the documented result
#
class DocExamplesTest < PipelineTest
DOC = File.read(File.join(ROOT, 'README.md'))
EXAMPLES = DOC.scan(/<!-- beispiel: (\S+) -->\n```ruby\n(.*?)\n```/m)
def run_example dir, name, *args
env = { 'BUNDLE_GEMFILE' => File.join(ROOT, 'Gemfile') }
def run_example dir, name, *args, env: {}
env = { 'BUNDLE_GEMFILE' => File.join(ROOT, 'Gemfile') }.merge(env)
Open3.capture3(env, RbConfig.ruby, File.join(ROOT, 'bin/etl'), 'run', File.join(ROOT, name), "--out=#{dir}", *args,
chdir: dir)
end
@@ -20,7 +20,7 @@ class DocExamplesTest < PipelineTest
def test_documented_code_equals_files
assert_equal %w[examples/kunden_import.rb examples/rechnungen_satzarten.rb examples/bewegungen_inkrementell.rb
examples/eigenes_plugin.rb examples/rest_lesen.rb examples/rest_schreiben.rb
examples/rest_zu_rest.rb], EXAMPLES.map(&:first)
examples/rest_zu_rest.rb examples/buchungen_export.rb], EXAMPLES.map(&:first)
EXAMPLES.each do |file, code|
assert_equal File.read(File.join(ROOT, file)).rstrip, code.rstrip, "#{file} differs from README.md"
end
@@ -152,4 +152,33 @@ class DocExamplesTest < PipelineTest
end
end
end
def test_buchungen_export
with_tmpdir do |dir|
env = { 'BUCHHALTUNG_DB' => "#{dir}/buchhaltung.sqlite3", 'SLACK_TOKEN' => nil }
out, err, status = run_example(dir, 'examples/buchungen_export.rb', '--dry-run', env:)
assert_equal 0, status.exitstatus, err
refute File.exist?("#{dir}/buchhaltung.sqlite3")
2.times do
out, err, status = run_example(dir, 'examples/buchungen_export.rb', env:)
assert_equal 0, status.exitstatus, err
end
assert_includes out, 'buchungen_export: read=7 written=7 rejected=0 betrag=0/0'
db = DataPipeline::SqliteStorage.new(database: "#{dir}/buchhaltung.sqlite3", query: 'select * from buchungen order by belnr, buzei')
rows = db.read
db.close
assert_equal 7, rows.size # upsert: the second run updates instead of inserting
assert_equal({ 'belnr' => '5100000002', 'buzei' => 3, 'datum' => '2026-10-02', 'konto' => '2000', 'shkzg' => 'H',
'dmbtr' => '409.80', 'betrag' => -409.8, 'buchungstext' => 'Papeterie Zürcher AG' }, rows[4])
assert_equal 'ohne Text', rows[5]['buchungstext']
csv = File.read("#{dir}/buchungen_sap.csv", encoding: 'Windows-1252').encode('UTF-8')
assert_includes csv, "5100000001;1;2026-10-01;1020;S;1250.00;Zahlung Müller GmbH\n"
history = DataPipeline.history("#{dir}/runs.ndjson", name: 'buchungen_export')
assert_equal [true, true, true], history.map { |run| run['ok'] }
assert_equal [true, false, false], history.map { |run| run['dry_run'] }
end
end
end
+30
View File
@@ -86,4 +86,34 @@ class CsvStorageTest < PipelineTest
assert_equal "FIRC|0\nDOHD|AR1|\nDOPS|1.50\nLARC|0000000002\n", File.read("#{dir}/c.ok")
end
end
def test_encoding_read_and_write
with_tmpdir do |dir|
DataPipeline::CsvStorage.new(file: "#{dir}/l1.csv", encoding: 'ISO-8859-1').write([{ 'name' => 'Müller', 'ort' => 'Zürich' }])
assert_equal "name,ort\nM\xFCller,Z\xFCrich\n".b, File.binread("#{dir}/l1.csv")
rows = DataPipeline::CsvStorage.new(file: "#{dir}/l1.csv", encoding: 'ISO-8859-1').read
assert_equal [{ 'name' => 'Müller', 'ort' => 'Zürich' }], rows
assert_equal Encoding::UTF_8, rows.first['name'].encoding
end
end
def test_encoding_error_or_replace
with_tmpdir do |dir|
assert_raises(Encoding::UndefinedConversionError) do
DataPipeline::CsvStorage.new(file: "#{dir}/a.csv", encoding: 'ISO-8859-1').write([{ 'x' => '5 €' }])
end
DataPipeline::CsvStorage.new(file: "#{dir}/b.csv", encoding: 'ISO-8859-1', replace: '?').write([{ 'x' => '5 €' }])
assert_equal "x\n5 ?\n", File.read("#{dir}/b.csv")
DataPipeline::CsvStorage.new(file: "#{dir}/c.csv", encoding: 'Windows-1252').write([{ 'x' => '5 €' }])
assert_equal "x\n5 \x80\n".b, File.binread("#{dir}/c.csv")
end
end
def test_bom_written_and_skipped
with_tmpdir do |dir|
DataPipeline::CsvStorage.new(file: "#{dir}/bom.csv", bom: true).write([{ 'a' => '1' }])
assert_equal "\uFEFFa\n1\n", File.read("#{dir}/bom.csv", encoding: 'utf-8')
assert_equal [{ 'a' => '1' }], DataPipeline::CsvStorage.new(file: "#{dir}/bom.csv").read
end
end
end
+63
View File
@@ -0,0 +1,63 @@
require_relative '../test_helper'
class FixedStorageTest < PipelineTest
LAYOUT = DataPipeline::Layout.new do
record 'KOPF', [['SATYP', 'AN', 4, true], ['BELNR', 'AN', 6, true], ['DATUM', 'DAT', 8]]
record 'POS', [['SATYP', 'AN', 4, true], ['TEXT', 'AN', 10], ['BETRAG', 'N', '7.2']]
end
ROWS = [{ 'satyp' => 'KOPF', 'belnr' => 'R1', 'datum' => Date.new(2026, 10, 9) },
{ 'satyp' => 'POS', 'text' => 'Müller', 'betrag' => 12.5 },
{ 'satyp' => 'POS', 'betrag' => -3 }].freeze
def test_write_and_read_back
with_tmpdir do |dir|
DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT).write(ROWS)
assert_equal "KOPFR1 20261009\nPOS Müller 00012.50\nPOS -0003.00\n", File.read("#{dir}/out.txt")
assert_equal [{ 'satyp' => 'KOPF', 'belnr' => 'R1', 'datum' => '20261009' },
{ 'satyp' => 'POS', 'text' => 'Müller', 'betrag' => '00012.50' },
{ 'satyp' => 'POS', 'text' => nil, 'betrag' => '-0003.00' }],
DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT).read
end
end
def test_encoding_line_end_and_streaming_read
with_tmpdir do |dir|
DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT, encoding: 'ISO-8859-1', line_end: "\r\n").write(ROWS.first(2))
assert_equal "KOPFR1 20261009\r\nPOS M\xFCller 00012.50\r\n".b, File.binread("#{dir}/out.txt")
rows = []
DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT, encoding: 'ISO-8859-1').read { |row| rows << row }
assert_equal 'Müller', rows.last['text']
end
end
def test_pipeline_with_container_and_dry_run
with_tmpdir do |dir|
rows = ROWS.map(&:dup)
out = capture_stdout do
DataPipeline.new do
logger PipelineHelper::QUIET
dry_run
read MemoryStorage.new(rows:)
write container(:fixed, file: "#{dir}/dry.txt", layout: LAYOUT)
end.run
end
refute File.exist?("#{dir}/dry.txt")
assert_includes out, 'DRY RUN'
DataPipeline.new do
logger PipelineHelper::QUIET
read MemoryStorage.new(rows:)
write container(:fixed, file: "#{dir}/out.txt", layout: LAYOUT)
end.run
assert_equal 3, File.readlines("#{dir}/out.txt").size
end
end
def test_options_and_lengths_required
assert_raises(ArgumentError) { DataPipeline::FixedStorage.new(layout: LAYOUT) }
assert_raises(ArgumentError) { DataPipeline::FixedStorage.new(file: 'x') }
layout = DataPipeline::Layout.new { record 'A', [['SATYP', 'AN']] }
assert_raises(DataPipeline::LayoutError) { DataPipeline::FixedStorage.new(file: 'x', layout:) }
end
end
+206
View File
@@ -0,0 +1,206 @@
require_relative '../test_helper'
require_relative '../support/test_http_server'
require_relative '../support/test_smtp_server'
# Report, Webhook, Slack, Email and notify against local test servers
class NotificationTest < PipelineTest
ROWS = [{ 'id' => '1', 'konto' => '1000' }, { 'id' => '2', 'konto' => '' }].freeze
# collects reports, for notify without network
class MemoryNotifier < DataPipeline::Notifier
attr_reader :reports
def initialize(fail: false) = (@reports = []; @fail = fail)
def deliver report, include_rows: false
raise 'kaputt' if @fail
@reports << [report, include_rows]
end
end
def teardown
@server&.stop
DataPipeline.connections_file = nil
super
end
def stats_with_rejected
run_pipeline(ROWS.map(&:dup)) { name 'export'; on_error :skip; validate 'konto' => { required: true } }.last
end
def report(stats = stats_with_rejected) = DataPipeline::Report.new([stats], params: { from: '2026-10-01' })
def test_report_event_lines_and_payload_without_rows
report = report()
assert_equal ['run.rejected', 'export: mit verworfenen Zeilen'], [report.event, report.subject]
assert_includes report.to_s, '- validate#1: konto: missing'
assert_includes report.to_s, 'Parameter: from=2026-10-01'
payload = report.to_h
assert_equal [1, 'run.rejected', 'export'], payload.values_at(:schema, :event, :title)
assert_equal [{ step: 'validate#1', error: 'konto: missing' }], payload[:runs].first[:errors]
assert_equal({ 'id' => '2', 'konto' => '' }, report.to_h(include_rows: true)[:runs].first[:errors].first[:row])
end
def test_report_worst_event_of_several_runs_and_script_error
ok = run_pipeline(ROWS.map(&:dup)).last
assert_equal 'run.succeeded', DataPipeline::Report.new([ok]).event
assert_equal 'run.rejected', DataPipeline::Report.new([ok, stats_with_rejected]).event
failed = DataPipeline::Report.new([ok], title: 'skript', error: RuntimeError.new('kaputt'))
assert_equal %w[run.failed skript], [failed.event, failed.title]
assert_includes failed.to_s, 'Fehler: RuntimeError: kaputt'
end
def test_webhook_signed_post
@server = TestHttpServer.new { [200, 'ok'] }
report = report()
DataPipeline::Webhook.new(url: "#{@server.url}/hook", secret: 's3cret', headers: { 'X-Team' => 'etl' }).deliver(report)
request = @server.requests.last
headers = request[:headers]
assert_equal ['/hook', 'run.rejected', report.id, 'etl'],
[request[:path], headers['x-datapipeline-event'], headers['x-datapipeline-delivery'], headers['x-team']]
assert_equal 1, JSON.parse(request[:body])['schema']
refute_includes request[:body], '"row"'
assert DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body],
timestamp: headers['x-datapipeline-timestamp'],
signature: headers['x-datapipeline-signature'])
refute DataPipeline::Webhook.valid_signature?(secret: 'falsch', body: request[:body],
timestamp: headers['x-datapipeline-timestamp'],
signature: headers['x-datapipeline-signature'])
refute DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body], timestamp: (Time.now.to_i - 600).to_s,
signature: headers['x-datapipeline-signature'])
end
def test_webhook_retries_429_and_5xx_but_not_4xx
answers = [[503, 'busy'], [429, 'slow', { 'Retry-After' => '0' }], [200, 'ok']]
@server = TestHttpServer.new { answers.shift }
DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report)
assert_equal 3, @server.requests.size
assert_equal 1, @server.requests.map { |r| r[:headers]['x-datapipeline-delivery'] }.uniq.size
@server.stop
@server = TestHttpServer.new { [400, 'bad'] }
error = assert_raises(DataPipeline::Webhook::Error) { DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report) }
assert_includes error.message, 'HTTP 400'
assert_equal 1, @server.requests.size
end
def test_webhook_network_error_after_retries
port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } }
error = assert_raises(DataPipeline::Webhook::Error) do
DataPipeline::Webhook.new(url: "http://127.0.0.1:#{port}", retries: 1, retry_wait: 0, timeout: 2).deliver(report)
end
assert_includes error.message, '127.0.0.1'
end
def test_slack_api_blocks_escape_and_mention
@server = TestHttpServer.new { [200, { 'ok' => true }] }
slack = DataPipeline::Slack.new(token: 'xoxb-1', channel: '#etl', api_url: @server.url, mention: '<!here>')
failed = DataPipeline::Report.new([stats_with_rejected], title: 'A & <B>', error: RuntimeError.new('x < y'))
slack.deliver(failed)
request = @server.requests.last
body = JSON.parse(request[:body])
assert_equal ['Bearer xoxb-1', '#etl'], [request[:headers]['authorization'], body['channel']]
assert_equal ':red_circle: *A &amp; &lt;B&gt;*: abgebrochen <!here>', body['blocks'][0]['text']['text']
assert_includes body['blocks'][1]['text']['text'], 'RuntimeError: x &lt; y'
slack.deliver(report)
refute_includes JSON.parse(@server.requests.last[:body])['blocks'][0]['text']['text'], '<!here>' # mention only on failure
end
def test_slack_error_and_incoming_webhook
@server = TestHttpServer.new { |req| req[:path] == '/api' ? [200, { 'ok' => false, 'error' => 'not_in_channel' }] : [200, 'ok'] }
error = assert_raises(DataPipeline::Slack::Error) { DataPipeline::Slack.new(token: 't', api_url: "#{@server.url}/api").send_message('x') }
assert_includes error.message, 'not_in_channel'
token = ENV.delete('SLACK_TOKEN')
hook = DataPipeline::Slack.new(webhook: "#{@server.url}/hook")
assert_equal 'ok', hook.deliver(report)
request = @server.requests.last
assert_nil request[:headers]['authorization']
refute JSON.parse(request[:body]).key?('channel')
ensure
ENV['SLACK_TOKEN'] = token if token
end
def test_email_over_smtp_with_login
@server = TestSmtpServer.new
mail = DataPipeline::Email.new(to: %w[a@example.com b@example.com], from: 'etl@example.com', host: '127.0.0.1',
port: @server.port, starttls: false, username: 'user', password: 'geheim')
mail.deliver(DataPipeline::Report.new([stats_with_rejected], title: 'Grösse Ä'))
sent = @server.mails.last
assert_equal ['etl@example.com', %w[a@example.com b@example.com], %w[user geheim]], sent.values_at(:from, :to, :auth)
head, body = sent[:data].split("\r\n\r\n", 2)
assert_includes head, "Subject: =?UTF-8?B?#{['[etl] Grösse Ä: mit verworfenen Zeilen'].pack('m0')}?="
assert_includes head, 'Content-Type: text/plain; charset=UTF-8'
text = body.unpack1('m').force_encoding('UTF-8')
assert_includes text, '- validate#1: konto: missing'
refute_includes text, 'Zeilen:'
end
def test_email_rows_on_request_and_errors
@server = TestSmtpServer.new
mail = DataPipeline::Email.new(to: 'a@example.com', host: '127.0.0.1', port: @server.port, starttls: false)
mail.deliver(report, include_rows: true)
assert_includes @server.mails.last[:data].split("\r\n\r\n", 2).last.unpack1('m').force_encoding('UTF-8'), '{"id":"2","konto":""}'
assert_nil @server.mails.last[:auth]
assert_raises(ArgumentError) { DataPipeline::Email.new(to: []) }
port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } }
assert_raises(DataPipeline::Email::Error) { DataPipeline::Email.new(to: 'a@b', host: '127.0.0.1', port:, starttls: false).send_mail('x', 'y') }
end
def test_notify_conditions_dry_run_and_rows
always = MemoryNotifier.new
failure = MemoryNotifier.new
problems = MemoryNotifier.new
rows = MemoryNotifier.new
run_pipeline(ROWS.map(&:dup)) do
name 'export'
on_error :skip
validate 'konto' => { required: true }
notify always, on: :always
notify failure, on: :failure
notify problems
notify rows, include_rows: true, title: 'Export Konten'
end
assert_equal [1, 0, 1], [always, failure, problems].map { |n| n.reports.size }
assert_equal ['Export Konten', true], [rows.reports.first[0].title, rows.reports.first[1]]
dry = MemoryNotifier.new
dry_too = MemoryNotifier.new
capture_stdout { run_pipeline(ROWS.map(&:dup), dry_run: true) { notify dry, on: :always; notify dry_too, on: :always, dry_run: true } }
assert_equal [0, 1], [dry.reports.size, dry_too.reports.size]
assert_raises(ArgumentError) { run_pipeline { notify Object.new } }
assert_raises(ArgumentError) { run_pipeline { notify MemoryNotifier.new, on: :sometimes } }
end
def test_notify_error_is_only_logged
log = StringIO.new
stats = DataPipeline.new do
logger Logger.new(log)
read MemoryStorage.new(rows: ROWS.map(&:dup))
write MemoryStorage.new
notify MemoryNotifier.new(fail: true), on: :always
end.run
assert stats.ok?
assert_includes log.string, 'notify NotificationTest::MemoryNotifier: RuntimeError: kaputt'
end
def test_notifier_from_connections_file
with_tmpdir do |dir|
File.write("#{dir}/c.yml", <<~YAML)
alerts: { type: webhook, url: "http://127.0.0.1:9/hook", secret: "<%= 'abc' %>" }
mail: { type: email, to: [etl@example.com], host: smtp.example.com }
db: { type: sqlite, database: x.sqlite3 }
YAML
DataPipeline.connections_file = "#{dir}/c.yml"
pipeline = DataPipeline.new { logger PipelineHelper::QUIET }
assert_kind_of DataPipeline::Webhook, pipeline.notifier(:alerts)
assert_equal 'abc', pipeline.notifier(:alerts).instance_variable_get(:@secret)
assert_kind_of DataPipeline::Email, pipeline.notifier(:mail)
assert_includes assert_raises(ArgumentError) { pipeline.notifier(:db) }.message, 'is no notifier'
end
end
end
+80
View File
@@ -130,4 +130,84 @@ class PostgresStorageTest < PipelineTest
error = assert_raises(PG::ConnectionBad) { DataPipeline::PostgresStorage.new(host: 'localhost', port: 1, database: 'x') }
assert_includes error.message, 'Error connection to database:'
end
def test_upsert_updates_existing_rows
with_pg_database do |db|
pg_exec db, 'create table u (belnr text primary key, betrag numeric, text text)'
writer = DataPipeline::PostgresStorage.new(**db, table: 'u', upsert: %w[belnr])
writer.write([{ 'belnr' => 'A', 'betrag' => 1, 'text' => 'alt' }, { 'belnr' => 'B', 'betrag' => 2, 'text' => 'b' }])
writer.write([{ 'belnr' => 'A', 'betrag' => 5, 'text' => 'neu' }])
writer.write_row({ 'belnr' => 'B' })
writer.write_row({ 'belnr' => 'C', 'betrag' => 3 })
writer.close
assert_equal [%w[A 5 neu], %w[B 2 b], ['C', '3', nil]],
pg_exec(db, 'select belnr, betrag::int::text as betrag, text from u order by belnr').map(&:values)
end
end
def test_write_row_in_batches
with_pg_database do |db|
pg_exec db, 'create table b (id int)'
writer = DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: 2)
3.times { |i| writer.write_row({ 'id' => i }) }
assert_equal '2', pg_exec(db, 'select count(*) from b').first['count'] # third row still pending
writer.finalize
assert_equal '3', pg_exec(db, 'select count(*) from b').first['count']
end
end
def test_close_writes_pending_rows
with_pg_database do |db|
pg_exec db, 'create table b (id int)'
writer = DataPipeline::PostgresStorage.new(**db, table: 'b')
writer.write_row({ 'id' => 1 })
writer.close
assert_equal '1', pg_exec(db, 'select count(*) from b').first['count']
end
end
def test_batch_size_all_is_one_transaction_for_the_stream
with_pg_database do |db|
pg_exec db, 'create table b (id int); insert into b values (99)'
aborted = DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: :all, truncate: true)
2.times { |i| aborted.write_row({ 'id' => i }) }
aborted.close # without finalize: rolled back, truncate too
assert_equal ['99'], pg_exec(db, 'select id from b').map { |r| r['id'] }
done = DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: :all, truncate: true)
2.times { |i| done.write_row({ 'id' => i }) }
done.finalize
assert_equal %w[0 1], pg_exec(db, 'select id from b order by id').map { |r| r['id'] }
end
end
def test_streamed_pipeline_aborted_with_batch_size_all_writes_nothing
with_pg_database do |db|
pg_exec db, 'create table b (id int)'
assert_raises(RuntimeError) do
DataPipeline.new do
logger PipelineHelper::QUIET
streamed
read MemoryStorage.new(rows: Array.new(5) { |i| { 'id' => i } })
map { |row| row['id'] == 3 ? raise('kaputt') : row }
write DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: :all)
end.run
end
assert_equal '0', pg_exec(db, 'select count(*) from b').first['count']
end
end
def test_streamed_copy_on_the_same_connection_as_the_cursor
with_pg_database do |db|
pg_exec db, 'create table k as select i as id from generate_series(1, 25) i; create table c (id int)'
pg = DataPipeline::PostgresStorage.new(**db, query: 'select id from k', table: 'c', fetch_size: 10, batch_size: 7)
DataPipeline.new do
logger PipelineHelper::QUIET
streamed
read pg
write pg
end.run
assert_equal '25', pg_exec(db, 'select count(*) from c').first['count']
end
end
end
+67
View File
@@ -92,4 +92,71 @@ class SqliteStorageTest < PipelineTest
assert_equal 'Deutschland', names['DE']['name']
end
end
def test_upsert_with_created_unique_index
with_tmpdir do |dir|
writer = DataPipeline::SqliteStorage.new(database: "#{dir}/u.sqlite3", table: 'u', create: true, upsert: %w[belnr])
writer.write([{ 'belnr' => 'A', 'betrag' => 1.0 }, { 'belnr' => 'B', 'betrag' => 2.0 }])
writer.write([{ 'belnr' => 'A', 'betrag' => 5.0 }])
writer.write_row({ 'belnr' => 'B' })
assert_equal [['A', 5.0], ['B', 2.0]], writer.query(query: 'select belnr, betrag from u order by belnr').map(&:values)
writer.close
end
end
def count(file) = DataPipeline::SqliteStorage.new(database: file).then { |db| db.query(query: 'select count(*) as n from b').first['n'].tap { db.close } }
def test_write_row_in_batches_and_query_sees_pending_rows
with_tmpdir do |dir|
file = "#{dir}/b.sqlite3"
writer = DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true, batch_size: 2)
3.times { |i| writer.write_row({ 'id' => i }) }
assert_equal 2, count(file) # third row still pending
assert_equal 3, writer.query(query: 'select count(*) as n from b').first['n']
writer.write_row({ 'id' => 3 })
writer.finalize
assert_equal 4, count(file)
end
end
def test_close_writes_pending_rows
with_tmpdir do |dir|
writer = DataPipeline::SqliteStorage.new(database: "#{dir}/b.sqlite3", table: 'b', create: true)
writer.write_row({ 'id' => 1 })
writer.close
assert_equal 1, count("#{dir}/b.sqlite3")
end
end
def test_batch_size_all_is_one_transaction_for_the_stream
with_tmpdir do |dir|
file = "#{dir}/b.sqlite3"
DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true).tap { |w| w.write([{ 'id' => 99 }]); w.close }
aborted = DataPipeline::SqliteStorage.new(database: file, table: 'b', truncate: true, batch_size: :all)
2.times { |i| aborted.write_row({ 'id' => i }) }
aborted.close # without finalize: rolled back, truncate too
assert_equal 1, count(file)
done = DataPipeline::SqliteStorage.new(database: file, table: 'b', truncate: true, batch_size: :all)
2.times { |i| done.write_row({ 'id' => i }) }
done.finalize
assert_equal 2, count(file)
end
end
def test_streamed_pipeline_aborted_with_batch_size_all_writes_nothing
with_tmpdir do |dir|
file = "#{dir}/b.sqlite3"
assert_raises(RuntimeError) do
DataPipeline.new do
logger PipelineHelper::QUIET
streamed
read MemoryStorage.new(rows: Array.new(5) { |i| { 'id' => i } })
map { |row| row['id'] == 3 ? raise('kaputt') : row }
write DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true, batch_size: :all)
end.run
end
refute DataPipeline::SqliteStorage.new(database: file).query(query: "select name from sqlite_master where name = 'b'").any?
end
end
end
+4 -3
View File
@@ -5,7 +5,7 @@ require 'json'
#
# minimal http server for DataPipeline::RestStorage tests, one request per connection
# server = TestHttpServer.new { |req| [200, { 'items' => [] }] }
# req: { method:, path:, query:, headers:, body: } -> [status, hash/array or string]
# req: { method:, path:, query:, headers:, body: } -> [status, hash/array or string, optional headers]
#
class TestHttpServer
attr_reader :requests
@@ -40,9 +40,10 @@ class TestHttpServer
request = { method:, path:, query: URI.decode_www_form(query.to_s).to_h, headers:, body: }
@requests << request
status, payload = @handler.call(request)
status, payload, extra = @handler.call(request)
text = payload.is_a?(String) ? payload : JSON.generate(payload)
client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n" \
extra = Hash(extra).map { |key, value| "#{key}: #{value}\r\n" }.join
client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n#{extra}" \
"Content-Length: #{text.bytesize}\r\nConnection: close\r\n\r\n#{text}"
rescue StandardError => e
warn "TestHttpServer: #{e.class}: #{e.message}"
+59
View File
@@ -0,0 +1,59 @@
require 'socket'
#
# minimal smtp server for DataPipeline::Email tests: EHLO, AUTH LOGIN, MAIL, RCPT, DATA, QUIT
# server = TestSmtpServer.new
# server.mails -> [{ from:, to: [...], data:, auth: [user, password] }]
#
class TestSmtpServer
attr_reader :mails
def initialize
@mails = []
@server = TCPServer.new('127.0.0.1', 0)
@thread = Thread.new { loop { serve(@server.accept) } }
end
def port = @server.addr[1]
def stop
@thread.kill
@server.close
end
private
def serve client
mail = { to: [] }
client.write "220 test ESMTP\r\n"
while (line = client.gets)
case line
when /\AEHLO/i then client.write "250-test\r\n250 AUTH LOGIN PLAIN\r\n"
when /\AAUTH LOGIN/i
client.write "334 VXNlcm5hbWU6\r\n"
user = client.gets.unpack1('m')
client.write "334 UGFzc3dvcmQ6\r\n"
mail[:auth] = [user, client.gets.unpack1('m')]
client.write "235 ok\r\n"
when /\AMAIL FROM:<(.*)>/i then mail[:from] = $1; client.write "250 ok\r\n"
when /\ARCPT TO:<(.*)>/i then mail[:to] << $1; client.write "250 ok\r\n"
when /\ADATA/i
client.write "354 go\r\n"
data = +''
while (part = client.gets) && part != ".\r\n"
data << part
end
mail[:data] = data
@mails << mail
mail = { to: [] }
client.write "250 queued\r\n"
when /\AQUIT/i then client.write "221 bye\r\n"; break
else client.write "250 ok\r\n"
end
end
rescue StandardError => e
warn "TestSmtpServer: #{e.class}: #{e.message}"
ensure
client.close
end
end