Class: AvroTurf::ConfluentSchemaRegistry

Inherits:
Object
  • Object
show all
Defined in:
lib/avro_turf/confluent_schema_registry.rb

Constant Summary collapse

CONTENT_TYPE = 
"application/vnd.schemaregistry.v1+json"

Instance Method Summary collapse

Constructor Details

#initialize(url, schema_context: nil, logger: Logger.new($stdout), proxy: nil, user: nil, password: nil, ssl_ca_file: nil, client_cert: nil, client_chain: nil, client_key: nil, client_key_pass: nil, client_cert_data: nil, client_chain_data: nil, client_key_data: nil, path_prefix: nil, connect_timeout: nil, resolv_resolver: nil, retry_limit: nil) ⇒ ConfluentSchemaRegistry

Returns a new instance of ConfluentSchemaRegistry.



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
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 8

def initialize(
  url,
  schema_context: nil,
  logger: Logger.new($stdout),
  proxy: nil,
  user: nil,
  password: nil,
  ssl_ca_file: nil,
  client_cert: nil,
  client_chain: nil,
  client_key: nil,
  client_key_pass: nil,
  client_cert_data: nil,
  client_chain_data: nil,
  client_key_data: nil,
  path_prefix: nil,
  connect_timeout: nil,
  resolv_resolver: nil,
  retry_limit: nil
)
  @path_prefix = path_prefix
  @schema_context_prefix = schema_context.nil? ? "" : ":.#{schema_context}:"
  @schema_context_options = schema_context.nil? ? {} : {query: {subject: @schema_context_prefix}}
  @logger = logger
  headers = Excon.defaults[:headers].merge(
    "Content-Type" => CONTENT_TYPE
  )
  params = {
    headers: headers,
    user: user,
    password: password,
    proxy: proxy,
    ssl_ca_file: ssl_ca_file,
    client_cert: client_cert,
    client_chain: client_chain,
    client_key: client_key,
    client_key_pass: client_key_pass,
    client_cert_data: client_cert_data,
    client_chain_data: client_chain_data,
    client_key_data: client_key_data,
    resolv_resolver: resolv_resolver,
    connect_timeout: connect_timeout,
    retry_limit: retry_limit
  }
  # Remove nil params to allow Excon to use its default values
  params.reject! { |_, v| v.nil? }
  @connection = Excon.new(
    url,
    params
  )
end

Instance Method Details

#check(subject, schema) ⇒ Object

Check if a schema exists. Returns nil if not found.



97
98
99
100
101
102
103
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 97

def check(subject, schema)
  data = post("/subjects/#{@schema_context_prefix}#{subject}",
    expects: [200, 404],
    body: {schema: schema.to_s}.to_json,
    idempotent: true)
  data unless data.has_key?("error_code")
end

#compatibility_issues(subject, schema, version = "latest") ⇒ Object

Check for specific schema compatibility issues Returns:

  • nil if the subject or version does not exist

  • a list of compatibility issues

docs.confluent.io/platform/current/schema-registry/develop/api.html#sr-api-compatibility



122
123
124
125
126
127
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 122

def compatibility_issues(subject, schema, version = "latest")
  data = post("/compatibility/subjects/#{@schema_context_prefix}#{subject}/versions/#{version}",
    expects: [200, 404], body: {schema: schema.to_s}.to_json, query: {verbose: true}, idempotent: true)

  data.fetch("messages", []) unless data.has_key?("error_code")
end

#compatible?(subject, schema, version = "latest") ⇒ Boolean

Check if a schema is compatible with the stored version. Returns:

  • true if compatible

  • nil if the subject or version does not exist

  • false if incompatible

docs.confluent.io/3.1.2/schema-registry/docs/api.html#compatibility

Returns:

  • (Boolean) 


111
112
113
114
115
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 111

def compatible?(subject, schema, version = "latest")
  data = post("/compatibility/subjects/#{@schema_context_prefix}#{subject}/versions/#{version}",
    expects: [200, 404], body: {schema: schema.to_s}.to_json, idempotent: true)
  data.fetch("is_compatible", false) unless data.has_key?("error_code")
end

#fetch(id) ⇒ Object 



60
61
62
63
64
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 60

def fetch(id)
  @logger.info "Fetching schema with id #{id}"
  data = get("/schemas/ids/#{id}", idempotent: true, **@schema_context_options)
  data.fetch("schema")
end

#global_config ⇒ Object

Get global config



130
131
132
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 130

def global_config
  get("/config", idempotent: true)
end

#register(subject, schema) ⇒ Object 



66
67
68
69
70
71
72
73
74
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 66

def register(subject, schema)
  data = post("/subjects/#{@schema_context_prefix}#{subject}/versions", body: {schema: schema.to_s}.to_json)

  id = data.fetch("id")

  @logger.info "Registered schema for subject `#{@schema_context_prefix}#{subject}`; id = #{id}"

  id
end

#schema_subject_versions(schema_id) ⇒ Object

Get the subject and version for a schema id



92
93
94
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 92

def schema_subject_versions(schema_id)
  get("/schemas/ids/#{schema_id}/versions", idempotent: true, **@schema_context_options)
end

#subject_config(subject) ⇒ Object

Get config for subject



140
141
142
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 140

def subject_config(subject)
  get("/config/#{@schema_context_prefix}#{subject}", idempotent: true)
end

#subject_version(subject, version = "latest") ⇒ Object

Get a specific version for a subject



87
88
89
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 87

def subject_version(subject, version = "latest")
  get("/subjects/#{@schema_context_prefix}#{subject}/versions/#{version}", idempotent: true)
end

#subject_versions(subject) ⇒ Object

List all versions for a subject



82
83
84
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 82

def subject_versions(subject)
  get("/subjects/#{@schema_context_prefix}#{subject}/versions", idempotent: true)
end

#subjects ⇒ Object

List all subjects



77
78
79
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 77

def subjects
  get("/subjects", idempotent: true)
end

#update_global_config(config) ⇒ Object

Update global config



135
136
137
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 135

def update_global_config(config)
  put("/config", body: config.to_json, idempotent: true)
end

#update_subject_config(subject, config) ⇒ Object

Update config for subject



145
146
147
 
# File 'lib/avro_turf/confluent_schema_registry.rb', line 145

def update_subject_config(subject, config)
  put("/config/#{@schema_context_prefix}#{subject}", body: config.to_json, idempotent: true)
end