35 lines
1.4 KiB
Ruby
35 lines
1.4 KiB
Ruby
#
|
|
# Von API zu API: neue Bestellungen über Folge-Links lesen und einzeln per PUT archivieren
|
|
# Ein zweiter Lauf ohne neue Bestellungen sendet nichts.
|
|
#
|
|
# ruby examples/demo_api.rb &
|
|
# bin/etl run examples/rest_zu_rest.rb [--url=http://127.0.0.1:4567] [--out=tmp]
|
|
#
|
|
require 'fileutils'
|
|
require 'datapipeline'
|
|
|
|
API = DataPipeline.params.fetch(:url, 'http://127.0.0.1:4567')
|
|
AUTH = "Bearer #{ENV.fetch('DEMO_API_TOKEN', 'demo-token')}"
|
|
out = DataPipeline.params.fetch(:out, 'tmp')
|
|
FileUtils.mkdir_p(out)
|
|
|
|
DataPipeline.new do
|
|
streamed
|
|
|
|
# höchste Bestellnummer des letzten erfolgreichen Laufs
|
|
stand = watermark :bestellungen, column: 'id', file: "#{out}/.state.json", initial: 0
|
|
|
|
# die Antwort nennt die nächste Seite selbst: {"items": [...], "next": "/api/bestellungen?cursor=2"}
|
|
read container(:rest, url: "#{API}/api/bestellungen", auth: AUTH, root: 'items',
|
|
next: ->(body) { body['next'] && "#{API}#{body['next']}" })
|
|
|
|
filter { |row| row['id'] > stand.value.to_i }
|
|
map { |row| row.merge('archiviert' => Date.today.iso8601) }
|
|
|
|
# eine url je Bestellung: PUT /api/archiv/<id>
|
|
write container(:rest, url: "#{API}/api/archiv", auth: AUTH, write_method: :put,
|
|
write_url: ->(row) { "#{API}/api/archiv/#{row['id']}" })
|
|
|
|
on_finish { |stats| puts "archiviert: #{stats.written}, Stand jetzt: #{stand.current}" }
|
|
end.run
|