Class: RedshiftConnector::Connector

Inherits:
Object
  • Object
show all
Defined in:
lib/redshift-connector/connector.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(exporter:, importer:, logger:) ⇒ Connector

Returns a new instance of Connector.



109
110
111
112
113
 
# File 'lib/redshift-connector/connector.rb', line 109

def initialize(exporter:, importer:, logger:)
  @exporter = exporter
  @importer = importer
  @logger = logger
end

Class Method Details

.transport_all(strategy: 'rename', schema:, table:, src_table: table, dest_table: table, columns:, bucket: nil, txn_id:, filter:, logger: RedshiftConnector.logger, quiet: false) ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/redshift-connector/connector.rb', line 70

def Connector.transport_all(
    strategy: 'rename',
    schema:,
    table:,
    src_table: table,
    dest_table: table,
    columns:,
    bucket: nil,
    txn_id:,
    filter:,
    logger: RedshiftConnector.logger,
    quiet: false
)
  bucket = bucket ? S3Bucket.get(bucket) : S3Bucket.default
  logger = NullLogger.new if quiet
  bundle = S3DataFileBundle.for_table(
    bucket: bucket,
    schema: schema,
    table: table,
    txn_id: txn_id,
    filter: filter,
    logger: logger
  )
  exporter = Exporter.for_table(
    bundle: bundle,
    schema: schema,
    table: table,
    columns: columns,
    logger: logger
  )
  importer = Importer.get_rebuild_class(strategy).new(
    dao: table.classify.constantize,
    bundle: bundle,
    columns: columns,
    logger: logger
  )
  new(exporter: exporter, importer: importer, logger: logger)
end

.transport_delta(schema:, table: nil, src_table: table, dest_table: table, condition:, columns:, delete_cond: nil, upsert_columns: nil, bucket: nil, txn_id:, filter:, logger: RedshiftConnector.logger, quiet: false) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/redshift-connector/connector.rb', line 8

def Connector.transport_delta(
    schema:,
    table: nil,
    src_table: table,
    dest_table: table,
    condition:,
    columns:,
    delete_cond: nil,
    upsert_columns: nil,
    bucket: nil,
    txn_id:, filter:,
    logger: RedshiftConnector.logger,
    quiet: false
)
  unless src_table and dest_table
    raise ArgumentError, "missing :table, :src_table or :dest_table"
  end
  bucket = bucket ? S3Bucket.get(bucket) : S3Bucket.default
  logger = NullLogger.new if quiet
  bundle = S3DataFileBundle.for_table(
    bucket: bucket,
    schema: schema,
    table: src_table,
    txn_id: txn_id,
    filter: filter,
    logger: logger
  )
  exporter = Exporter.for_table_delta(
    bundle: bundle,
    schema: schema,
    table: src_table,
    columns: columns,
    condition: condition,
    logger: logger
  )
  if delete_cond and upsert_columns
    raise ArgumentError, "delete_cond and upsert_columns are exclusive"
  end
  dao = dest_table.classify.constantize
  importer =
    if delete_cond
      Importer::InsertDelta.new(
        dao: dao,
        bundle: bundle,
        columns: columns,
        delete_cond: delete_cond,
        logger: logger
      )
    elsif upsert_columns
      Importer::Upsert.new(
        dao: dao,
        bundle: bundle,
        columns: columns,
        upsert_columns: upsert_columns,
        logger: logger
      )
    else
      raise ArgumentError, "either of delete_cond or upsert_columns is required for transport_delta"
    end
  new(exporter: exporter, importer: importer, logger: logger)
end

Instance Method Details

#execute ⇒ Object 



127
128
129
130
 
# File 'lib/redshift-connector/connector.rb', line 127

def execute
  export(forced: export_forced?) if export_enabled?
  import if import_enabled?
end

#export(forced: false) ⇒ Object 



132
133
134
135
136
137
138
139
 
# File 'lib/redshift-connector/connector.rb', line 132

def export(forced: false)
  @logger.info "==== export task =================================================="
  if not forced and @exporter.completed?
    @logger.info "export task is already executed; skip"
  else
    @exporter.execute
  end
end

#export_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/redshift-connector/connector.rb', line 115

def export_enabled?
  not ENV['IMPORT_ONLY']
end

#export_forced? ⇒ Boolean

Returns:

  • (Boolean) 


119
120
121
 
# File 'lib/redshift-connector/connector.rb', line 119

def export_forced?
  !! (ENV['EXPORT_ONLY'] or ENV['FORCE'])
end

#import ⇒ Object 



141
142
143
144
 
# File 'lib/redshift-connector/connector.rb', line 141

def import
  @logger.info "==== import task =================================================="
  @importer.execute
end

#import_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


123
124
125
 
# File 'lib/redshift-connector/connector.rb', line 123

def import_enabled?
  not ENV['EXPORT_ONLY']
end