Class: Deimos::SchemaBackends::AvroBase

Inherits:
Base
  • Object
show all
Defined in:
lib/deimos/schema_backends/avro_base.rb,
sig/defs.rbs

Overview

Encode / decode using Avro, either locally or via schema registry.

Direct Known Subclasses

AvroLocal, AvroSchemaRegistry, AvroValidation

Instance Attribute Summary collapse

Attributes inherited from Base

#key_schema, #namespace, #registry_info, #schema

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#coerce, #decode, #decode_payload, #encode, #encode_payload, #inspect

Constructor Details

#initialize(schema:, namespace:, registry_info: nil) ⇒ AvroBase

Returns a new instance of AvroBase.

Parameters:

  • schema: (String, Symbol)
  • namespace: (String) 


14
15
16
17
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 14

def initialize(schema:, namespace:, registry_info: nil)
  super
  @schema_store = SchemaRegistry::AvroSchemaStore.new(path: Deimos.config.schema.path)
end

Instance Attribute Details

#schema_store ⇒ Object

Returns the value of attribute schema_store.

Returns:

  • (Object) 


1432
1433
1434
 
# File 'sig/defs.rbs', line 1432

def schema_store
  @schema_store
end

Class Method Details

.content_type ⇒ String

Returns:

  • (String) 


103
104
105
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 103

def self.content_type
  'avro/binary'
end

.field_type(avro_schema) ⇒ String

Converts Avro::Schema::NamedSchema's to String form for generated YARD docs. Recursively handles the typing for Arrays, Maps and Unions.

@param avro_schema

@return — A string representation of the Type of this SchemaField

Parameters:

Returns:

  • (String) 


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
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 117

def self.field_type(avro_schema)
  case avro_schema.type_sym
  when :string, :boolean
    avro_schema.type_sym.to_s.titleize
  when :int, :long
    'Integer'
  when :float, :double
    'Float'
  when :record, :enum
    schema_classname(avro_schema)
  when :array
    arr_t = field_type(Deimos::SchemaField.new('n/a', avro_schema.items).type)
    "Array<#{arr_t}>"
  when :map
    map_t = field_type(Deimos::SchemaField.new('n/a', avro_schema.values).type)
    "Hash<String, #{map_t}>"
  when :union
    types = avro_schema.schemas.map do |t|
      field_type(Deimos::SchemaField.new('n/a', t).type)
    end
    types.join(', ')
  when :null
    'nil'
  end
end

.mock_backend ⇒ Symbol

Returns:

  • (Symbol) 


98
99
100
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 98

def self.mock_backend
  :avro_validation
end

.schema_base_class(schema) ⇒ Avro::Schema::NamedSchema

Returns the base type of this schema. Decodes Arrays, Maps and Unions

@param schema

Parameters:

Returns:



146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 146

def self.schema_base_class(schema)
  case schema.type_sym
  when :array
    schema_base_class(schema.items)
  when :map
    schema_base_class(schema.values)
  when :union
    schema.schemas.map(&method(:schema_base_class)).
      reject { |s| s.type_sym == :null }.first
  else
    schema
  end
end

.schema_classname(schema) ⇒ String

@param schema — A named schema

Parameters:

Returns:

  • (String) 


109
110
111
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 109

def self.schema_classname(schema)
  schema.name.underscore.camelize.singularize
end

Instance Method Details

#coerce_field(field, value) ⇒ Object

Parameters:

Returns:

  • (Object) 


72
73
74
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 72

def coerce_field(field, value)
  AvroSchemaCoercer.new(avro_schema).coerce_type(field.type, value)
end

#decode_key(payload, key_id) ⇒ String

Parameters:

  • payload (::Hash[untyped, untyped])
  • key_id (String, Symbol)

Returns:

  • (String) 


41
42
43
44
45
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 41

def decode_key(payload, key_id)
  @key_schema ||= generate_key_schema(key_id)
  field_name = _field_name_from_schema(@key_schema)
  decode(payload, schema: @key_schema['name'])[field_name]
end

#encode_key(key_id, key, topic: nil) ⇒ String

Parameters:

  • key_id (String, Symbol)
  • key (String, ::Hash[untyped, untyped])
  • topic: (String, nil) (defaults to: nil)

Returns:

  • (String) 


29
30
31
32
33
34
35
36
37
38
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 29

def encode_key(key_id, key, topic: nil)
  begin
    @key_schema ||= @schema_store.find("#{@schema}_key")
  rescue SchemaRegistry::SchemaNotFoundError
    @key_schema = generate_key_schema(key_id)
  end
  field_name = _field_name_from_schema(@key_schema)
  payload = key.is_a?(Hash) ? key : { field_name => key }
  encode(payload, schema: @key_schema['name'], topic: topic, is_key: true)
end

#generate_key_schema(field_name) ⇒ Object 



160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 160

def generate_key_schema(field_name)
  key_field = avro_schema.fields.find { |f| f.name == field_name.to_s }
  name = _key_schema_name(@schema)
  key_schema = {
    'type' => 'record',
    'name' => name,
    'namespace' => @namespace,
    'doc' => "Key for #{@namespace}.#{@schema} - autogenerated by Deimos",
    'fields' => [
      {
        'name' => field_name,
        'type' => key_field.type.type_sym.to_s
      }
    ]
  }
  @schema_store.add_schema(key_schema)
  @key_schema = key_schema
end

#load_schema ⇒ Avro::Schema

Returns:



93
94
95
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 93

def load_schema
  avro_schema
end

#schema_fields ⇒ ::Array[SchemaField]

Returns:



77
78
79
80
81
82
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 77

def schema_fields
  avro_schema.fields.map do |field|
    enum_values = field.type.type == 'enum' ? field.type.symbols : []
    SchemaField.new(field.name, field.type, enum_values, field.default)
  end
end

#sql_type(field) ⇒ Symbol

:nodoc:

Parameters:

Returns:

  • (Symbol) 


48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 48

def sql_type(field)
  type = field.type.type
  return type if %w(array map record).include?(type)

  if type == :union
    non_null = field.type.schemas.reject { |f| f.type == :null }
    if non_null.size > 1
      warn("WARNING: #{field.name} has more than one non-null type. Picking the first for the SQL type.")
    end
    return non_null.first.type
  end
  return type.to_sym if %w(float boolean).include?(type)
  return :integer if type == 'int'
  return :bigint if type == 'long'

  if type == 'double'
    warn('Avro `double` type turns into SQL `float` type. Please ensure you have the correct `limit` set.')
    return :float
  end

  :string
end

#supports_class_generation? ⇒ Boolean

Returns:

  • (Boolean) 


24
25
26
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 24

def supports_class_generation?
  true
end

#supports_key_schemas? ⇒ Boolean

Returns:

  • (Boolean) 


19
20
21
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 19

def supports_key_schemas?
  true
end

#validate(payload, schema:) ⇒ void

This method returns an undefined value.

Parameters:

  • payload (::Hash[untyped, untyped])
  • schema: (String, Symbol) 


85
86
87
88
89
 
# File 'lib/deimos/schema_backends/avro_base.rb', line 85

def validate(payload, schema:)
  Avro::SchemaValidator.validate!(avro_schema(schema), payload,
                                  recursive: true,
                                  fail_on_extra_fields: true)
end