Module: StructuredDataToSql::IOSupport

Defined in:
lib/structured_data_to_sql/io_support.rb

Defined Under Namespace

Classes: CountingWriter

Class Method Summary collapse

Class Method Details

.discover(source, patterns:, stream_extension:, input_gzip: false) ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/structured_data_to_sql/io_support.rb', line 46

def discover(source, patterns:, stream_extension:, input_gzip: false)
  temporary_files = []
  source_items = readable_io?(source) ? [source] : Array(source)
  candidates =
    source_items.flat_map do |item|
      if readable_io?(item)
        suffix = "#{stream_extension}#{input_gzip ? ".gz" : ""}"
        tempfile = Tempfile.new(["structured-data-to-sql-input-", suffix])
        tempfile.binmode
        IO.copy_stream(item, tempfile)
        tempfile.close
        temporary_files << tempfile
        Pathname(tempfile.path)
      else
        path = Pathname(item)
        raise InputError, "Input not found: #{path}" unless path.exist?

        if path.directory?
          patterns.flat_map { |pattern| path.glob(pattern).to_a }
        else
          path
        end
      end
    end
  [candidates.flatten.uniq.sort_by(&:to_s), temporary_files]
rescue SystemCallError, IOError => e
  temporary_files&.each(&:unlink)
  raise InputError, e.message
end

.estimated_input_bytes(path) ⇒ Object

Bytes the streaming reader will hand out for path: the file size for plain files, the gzip ISIZE trailer (uncompressed length modulo 2**32, lifted until it is at least the compressed size) for .gz files. One four-byte read per file, no decompression, never raises.



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/structured_data_to_sql/io_support.rb', line 80

def estimated_input_bytes(path)
  pathname = Pathname(path)
  size = pathname.size
  return size unless pathname.extname == ".gz" && size >= 18

  trailer = File.binread(pathname.to_s, 4, size - 4)
  estimate = trailer.unpack1("V")
  estimate += 2**32 while estimate < size
  estimate
rescue StandardError
  begin
    Pathname(path).size
  rescue StandardError
    0
  end
end

.readable_io?(object) ⇒ Boolean

Returns:

  • (Boolean) 


36
37
38
39
 
# File 'lib/structured_data_to_sql/io_support.rb', line 36

def readable_io?(object)
  object.respond_to?(:read) && !object.is_a?(String) &&
    !object.is_a?(Pathname)
end

.unique_adjacent_path(path) ⇒ Object 



97
98
99
100
101
102
 
# File 'lib/structured_data_to_sql/io_support.rb', line 97

def unique_adjacent_path(path)
  pathname = Pathname(path)
  pathname.dirname.join(
    ".#{pathname.basename}.tmp-#{Process.pid}-#{SecureRandom.hex(8)}"
  )
end

.validate_output_target!(target) ⇒ Object

Raises:



104
105
106
107
108
109
110
111
112
 
# File 'lib/structured_data_to_sql/io_support.rb', line 104

def validate_output_target!(target)
  return if writable_io?(target)

  path = Pathname(target)
  return unless path.directory?

  raise OutputError,
        "Output path is a directory; provide a file path: #{path}"
end

.with_output(target, gzip:, atomic: true) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/structured_data_to_sql/io_support.rb', line 114

def with_output(target, gzip:, atomic: true)
  validate_output_target!(target)
  if writable_io?(target)
    counting_target = CountingWriter.new(target)
    if gzip
      writer = Zlib::GzipWriter.new(counting_target)
      begin
        yield writer
      ensure
        writer.finish
      end
    else
      yield counting_target
    end
    return counting_target.bytes_written
  end

  path = Pathname(target)
  temporary_path = atomic ? unique_adjacent_path(path) : path
  begin
    if gzip
      Zlib::GzipWriter.open(temporary_path.to_s) do |gzip_writer|
        yield gzip_writer
      end
    else
      File.open(temporary_path, "w:utf-8") do |file_writer|
        yield file_writer
      end
    end
    File.rename(temporary_path, path) if atomic
    File.size(path)
  rescue SystemCallError, IOError, Zlib::Error => e
    raise OutputError, e.message
  ensure
    if atomic && temporary_path && temporary_path.exist?
      FileUtils.rm_f(temporary_path)
    end
  end
end

.writable_io?(object) ⇒ Boolean

Returns:

  • (Boolean) 


41
42
43
44
 
# File 'lib/structured_data_to_sql/io_support.rb', line 41

def writable_io?(object)
  object.respond_to?(:write) && !object.is_a?(String) &&
    !object.is_a?(Pathname)
end