-
Notifications
You must be signed in to change notification settings - Fork 5
/
schemas.cljc
298 lines (273 loc) · 12 KB
/
schemas.cljc
1
2
3
4
5
6
7
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
69
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
108
109
110
111
112
113
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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
(ns deercreeklabs.lancaster.schemas
(:require
[camel-snake-kebab.core :as csk]
[clojure.string :as str]
[deercreeklabs.baracus :as ba]
[deercreeklabs.lancaster.fingerprint :as fingerprint]
[deercreeklabs.lancaster.impl :as impl]
[deercreeklabs.lancaster.resolution :as resolution]
[deercreeklabs.lancaster.pcf :as pcf]
[deercreeklabs.lancaster.utils :as u]
[deercreeklabs.log-utils :as lu :refer [debugs]]
#?(:clj [primitive-math :as pm])
[schema.core :as s :include-macros true]
[taoensso.timbre :as timbre :refer [debugf errorf infof]]))
#?(:clj (pm/use-primitive-operators))
(def WrappedData [(s/one s/Keyword "schema-name")
(s/one s/Any "data")])
(def RecordFieldDef [(s/one s/Keyword "field-name")
(s/one (s/protocol u/ILancasterSchema) "field-schema")
(s/optional s/Any "field-default")])
(defrecord LancasterSchema
[schema-name edn-schema json-schema parsing-canonical-form
fingerprint64 plumatic-schema serializer deserializer default-data-size
*name->serializer *name->deserializer *pcf->resolving-deserializer]
u/ILancasterSchema
(serialize [this data]
(let [os (impl/make-output-stream default-data-size)]
(u/serialize this os data)
(u/to-byte-array os)))
(serialize [this os data]
(serializer os data []))
(deserialize [this writer-pcf is]
(if (= writer-pcf parsing-canonical-form)
(deserializer is)
(if-let [rd (@*pcf->resolving-deserializer writer-pcf)]
(rd is)
(let [rd (resolution/make-resolving-deserializer writer-pcf this
*name->deserializer)]
(swap! *pcf->resolving-deserializer assoc writer-pcf rd)
(rd is)))))
(wrap [this data]
[schema-name data])
(get-edn-schema [this]
edn-schema)
(get-json-schema [this]
json-schema)
(get-parsing-canonical-form [this]
parsing-canonical-form)
(get-fingerprint64 [this]
fingerprint64)
(get-plumatic-schema [this]
plumatic-schema))
(defmulti make-edn-schema u/first-arg-dispatch)
(defmulti validate-schema-args u/first-arg-dispatch)
(defn edn-schema->lancaster-schema [schema-type edn-schema]
(let [name->edn-schema (u/make-name->edn-schema edn-schema)
avro-schema (if (u/avro-primitive-types schema-type)
(name schema-type)
(u/edn-schema->avro-schema edn-schema))
json-schema (u/edn->json-string avro-schema)
parsing-canonical-form (pcf/avro-schema->pcf avro-schema)
fingerprint64 (fingerprint/fingerprint64 parsing-canonical-form)
plumatic-schema (u/edn-schema->plumatic-schema edn-schema
name->edn-schema)
*name->serializer (atom {})
*name->deserializer (atom {})
serializer (u/make-serializer edn-schema name->edn-schema
*name->serializer)
deserializer (u/make-deserializer edn-schema *name->deserializer)
default-data-size (u/make-default-data-size edn-schema
name->edn-schema)
*pcf->resolving-deserializer (atom {})
schema-name (u/get-schema-name edn-schema)]
(->LancasterSchema
schema-name edn-schema json-schema parsing-canonical-form
fingerprint64 plumatic-schema serializer deserializer default-data-size
*name->serializer *name->deserializer *pcf->resolving-deserializer)))
(defn get-name-or-schema [edn-schema *names]
(let [schema-name (u/get-schema-name edn-schema)]
(if (@*names schema-name)
schema-name
(do
(swap! *names conj schema-name)
edn-schema))))
(defn fix-repeated-schemas
([edn-schema]
(fix-repeated-schemas edn-schema (atom #{})))
([edn-schema *names]
(case (u/get-avro-type edn-schema)
:enum (get-name-or-schema edn-schema *names)
:fixed (get-name-or-schema edn-schema *names)
:array (update edn-schema :items #(fix-repeated-schemas % *names))
:map (update edn-schema :values #(fix-repeated-schemas % *names))
:union (mapv #(fix-repeated-schemas % *names) edn-schema)
:record (let [name-or-schema (get-name-or-schema edn-schema *names)
fix-field (fn [field]
(update field :type
#(fix-repeated-schemas % *names)))]
(if (map? name-or-schema)
(update edn-schema :fields #(mapv fix-field %))
name-or-schema))
edn-schema)))
(defn make-schema
([schema-type ns-name schema-name args]
(let [name-kw (keyword ns-name schema-name)]
(make-schema schema-type name-kw args)))
([schema-type name-kw args]
(when (and (u/avro-named-types schema-type)
(not (keyword? name-kw)))
(let [fn-name (str "make-" (name schema-type) "-schema")]
(throw (ex-info (str "First arg to " fn-name " must be a name keyword."
"The keyword can be namespaced or not.")
{:given-name-kw name-kw}))))
(when-not (u/avro-primitive-types schema-type)
(validate-schema-args schema-type args))
(let [edn-schema (if (u/avro-primitive-types schema-type)
schema-type
(-> (make-edn-schema schema-type name-kw args)
(fix-repeated-schemas)))]
(edn-schema->lancaster-schema schema-type edn-schema ))))
(defn merge-record-schemas [name-kw schemas]
(when-not (keyword? name-kw)
(throw (ex-info (str "First arg to merge-record-schemas must be a name "
"keyword. The keyword can be namespaced or not.")
{:given-name-kw name-kw})))
(when-not (sequential? schemas)
(throw (ex-info (str "Second arg to merge-record-schemas must be a "
"sequence of record schema objects.")
{:given-schemas schemas})))
(doseq [schema schemas]
(when (or (not (instance? LancasterSchema schema))
(not (= :record (:type (u/get-edn-schema schema)))))
(throw (ex-info (str "Second arg to merge-record-schemas must be a "
"sequence of record schema objects.")
{:bad-schema schema}))))
(let [fields (mapcat #(:fields (u/get-edn-schema %)) schemas)
edn-schema {:name name-kw
:type :record
:fields fields}]
(edn-schema->lancaster-schema :record edn-schema)))
(defn make-primitive-schema [schema-kw]
(make-schema schema-kw nil nil))
(defn schema-or-kw? [x]
(or (instance? LancasterSchema x)
(keyword? x)))
(defmethod validate-schema-args :record
[schema-type fields]
(when-not (sequential? fields)
(throw (ex-info (str "Second arg to make-record-schema must be a sequence "
"of field definitions.")
{:given-fields fields})))
(doseq [field fields]
(let [[name-kw field-schema default] field]
(when-not (keyword? name-kw)
(throw (ex-info "First arg in field definition must be a name keyword."
{:given-name-kw name-kw})))
(when-not (schema-or-kw? field-schema)
(throw
(ex-info (str "Second arg in field definition must be a schema object "
"or a name keyword.")
{:given-field-schema field-schema})))
(when default
(try
(u/serialize field-schema (impl/make-output-stream 100) default)
(catch #?(:clj Exception :cljs js/Error) e
(let [ex-msg (lu/get-exception-msg e)]
(if (str/includes? ex-msg "not a valid")
(throw
(ex-info
(str "Default value for field `" name-kw "` is invalid. "
ex-msg)
(u/sym-map name-kw default ex-msg)))))))))))
(defmethod validate-schema-args :enum
[schema-type symbols]
(when-not (sequential? symbols)
(throw (ex-info (str "Second arg to make-enum-schema must be a sequence "
"of keywords.")
{:given-symbols symbols})))
(doseq [symbol symbols]
(when-not (keyword? symbol)
(throw (ex-info "All symbols in an enum must be keywords."
{:given-symbol symbol})))))
(defmethod validate-schema-args :fixed
[schema-type size]
(when-not (integer? size)
(throw (ex-info (str "Second arg to make-fixed-schema (size) must be an "
"integer.")
{:given-size size}))))
(defmethod validate-schema-args :array
[schema-type items-schema]
(when-not (schema-or-kw? items-schema)
(throw
(ex-info (str "Second arg to make-array-schema must be a schema object "
"or a name keyword.")
{:given-items-schema items-schema}))))
(defmethod validate-schema-args :map
[schema-type values-schema]
(when-not (schema-or-kw? values-schema)
(throw
(ex-info (str "Second arg to make-map-schema must be a schema object "
"or a name keyword.")
{:given-values-schema values-schema}))))
(defmethod validate-schema-args :union
[schema-type member-schemas]
(when-not (sequential? member-schemas)
(throw (ex-info (str "Second arg to make-union-schema must be a sequence "
"of member schema objects or name keywords.")
{:given-member-schemas member-schemas})))
(doseq [member-schema member-schemas]
(when-not (schema-or-kw? member-schema)
(throw
(ex-info (str "All member schemas in a union must be schema objects "
"or name keywords.")
{:bad-member-schema member-schema}))))
(when (u/illegal-union? (map u/get-edn-schema
;; Name keywords & named schemas are always okay
(remove keyword? member-schemas)))
(throw (ex-info "Illegal union schema." {:schema member-schemas}))))
(defn make-record-field [field]
(when-not (#{2 3} (count field))
(throw
(ex-info (str "Record field definition must have 2 or 3 parameters. ("
"[field-name field-schema] or "
"[field-name field-schema field-default]).\n"
" Got " (count field) " parameters.\n"
" Bad field definition: " field)
{:bad-field-def field})))
(let [[field-name field-schema field-default] field
field-edn-schema (if (keyword? field-schema)
field-schema
(u/get-edn-schema field-schema))]
(when-not (keyword? field-name)
(throw
(ex-info (str "Field names must be keywords. Bad field name: "
field-name)
(u/sym-map field-name field-schema field-default field))))
{:name field-name
:type field-edn-schema
:default (u/get-default-data field-edn-schema field-default)}))
(defmethod make-edn-schema :record
[schema-type name-kw fields]
(let [name-kw (u/qualify-name-kw name-kw)
fields (binding [u/**enclosing-namespace** (namespace name-kw)]
(mapv make-record-field fields))
edn-schema {:name name-kw
:type :record
:fields fields}]
edn-schema))
(defmethod make-edn-schema :enum
[schema-type name-kw symbols]
(let [name-kw (u/qualify-name-kw name-kw)
edn-schema {:name name-kw
:type :enum
:symbols symbols}]
edn-schema))
(defmethod make-edn-schema :fixed
[schema-type name-kw size]
(let [name-kw (u/qualify-name-kw name-kw)
edn-schema {:name name-kw
:type :fixed
:size size}]
edn-schema))
(defmethod make-edn-schema :array
[schema-type name-kw items]
{:type :array
:items (u/ensure-edn-schema items)})
(defmethod make-edn-schema :map
[schema-type name-kw values]
{:type :map
:values (u/ensure-edn-schema values)})
(defmethod make-edn-schema :union
[schema-type name-kw member-schemas]
(mapv u/ensure-edn-schema member-schemas))