/
declare.py
467 lines (400 loc) · 21.1 KB
/
declare.py
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
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
"""
This module hosts functions to convert DataJoint table definitions into mysql table definitions, and to
declare the corresponding mysql tables.
"""
import re
import pyparsing as pp
import logging
from .errors import DataJointError, _support_filepath_types, FILEPATH_FEATURE_SWITCH
from .attribute_adapter import get_adapter
from .utils import OrderedDict
UUID_DATA_TYPE = 'binary(16)'
MAX_TABLE_NAME_LENGTH = 64
CONSTANT_LITERALS = {'CURRENT_TIMESTAMP'} # SQL literals to be used without quotes (case insensitive)
EXTERNAL_TABLE_ROOT = '~external'
TYPE_PATTERN = {k: re.compile(v, re.I) for k, v in dict(
INTEGER=r'((tiny|small|medium|big|)int|integer)(\s*\(.+\))?(\s+unsigned)?(\s+auto_increment)?|serial$',
DECIMAL=r'(decimal|numeric)(\s*\(.+\))?(\s+unsigned)?$',
FLOAT=r'(double|float|real)(\s*\(.+\))?(\s+unsigned)?$',
STRING=r'(var)?char\s*\(.+\)$',
ENUM=r'enum\s*\(.+\)$',
BOOL=r'bool(ean)?$', # aliased to tinyint(1)
TEMPORAL=r'(date|datetime|time|timestamp|year)(\s*\(.+\))?$',
INTERNAL_BLOB=r'(tiny|small|medium|long|)blob$',
EXTERNAL_BLOB=r'blob@(?P<store>[a-z]\w*)$',
INTERNAL_ATTACH=r'attach$',
EXTERNAL_ATTACH=r'attach@(?P<store>[a-z]\w*)$',
FILEPATH=r'filepath@(?P<store>[a-z]\w*)$',
UUID=r'uuid$',
ADAPTED=r'<.+>$'
).items()}
# custom types are stored in attribute comment
SPECIAL_TYPES = {'UUID', 'INTERNAL_ATTACH', 'EXTERNAL_ATTACH', 'EXTERNAL_BLOB', 'FILEPATH', 'ADAPTED'}
NATIVE_TYPES = set(TYPE_PATTERN) - SPECIAL_TYPES
EXTERNAL_TYPES = {'EXTERNAL_ATTACH', 'EXTERNAL_BLOB', 'FILEPATH'} # data referenced by a UUID in external tables
SERIALIZED_TYPES = {'EXTERNAL_ATTACH', 'INTERNAL_ATTACH', 'EXTERNAL_BLOB', 'INTERNAL_BLOB'} # requires packing data
assert set().union(SPECIAL_TYPES, EXTERNAL_TYPES, SERIALIZED_TYPES) <= set(TYPE_PATTERN)
def match_type(attribute_type):
try:
return next(category for category, pattern in TYPE_PATTERN.items() if pattern.match(attribute_type))
except StopIteration:
raise DataJointError("Unsupported attribute type {type}".format(type=attribute_type)) from None
logger = logging.getLogger(__name__)
def build_foreign_key_parser_old():
# old-style foreign key parser. Superceded by expression-based syntax. See issue #436
# This will be deprecated in a future release.
left = pp.Literal('(').suppress()
right = pp.Literal(')').suppress()
attribute_name = pp.Word(pp.srange('[a-z]'), pp.srange('[a-z0-9_]'))
new_attrs = pp.Optional(left + pp.delimitedList(attribute_name) + right).setResultsName('new_attrs')
arrow = pp.Literal('->').suppress()
lbracket = pp.Literal('[').suppress()
rbracket = pp.Literal(']').suppress()
option = pp.Word(pp.srange('[a-zA-Z]'))
options = pp.Optional(lbracket + pp.delimitedList(option) + rbracket).setResultsName('options')
ref_table = pp.Word(pp.alphas, pp.alphanums + '._').setResultsName('ref_table')
ref_attrs = pp.Optional(left + pp.delimitedList(attribute_name) + right).setResultsName('ref_attrs')
return new_attrs + arrow + options + ref_table + ref_attrs
def build_foreign_key_parser():
arrow = pp.Literal('->').suppress()
lbracket = pp.Literal('[').suppress()
rbracket = pp.Literal(']').suppress()
option = pp.Word(pp.srange('[a-zA-Z]'))
options = pp.Optional(lbracket + pp.delimitedList(option) + rbracket).setResultsName('options')
ref_table = pp.restOfLine.setResultsName('ref_table')
return arrow + options + ref_table
def build_attribute_parser():
quoted = pp.QuotedString('"') ^ pp.QuotedString("'")
colon = pp.Literal(':').suppress()
attribute_name = pp.Word(pp.srange('[a-z]'), pp.srange('[a-z0-9_]')).setResultsName('name')
data_type = (pp.Combine(pp.Word(pp.alphas) + pp.SkipTo("#", ignore=quoted))
^ pp.QuotedString('<', endQuoteChar='>', unquoteResults=False)).setResultsName('type')
default = pp.Literal('=').suppress() + pp.SkipTo(colon, ignore=quoted).setResultsName('default')
comment = pp.Literal('#').suppress() + pp.restOfLine.setResultsName('comment')
return attribute_name + pp.Optional(default) + colon + data_type + comment
def build_index_parser():
left = pp.Literal('(').suppress()
right = pp.Literal(')').suppress()
unique = pp.Optional(pp.CaselessKeyword('unique')).setResultsName('unique')
index = pp.CaselessKeyword('index').suppress()
attribute_name = pp.Word(pp.srange('[a-z]'), pp.srange('[a-z0-9_]'))
return unique + index + left + pp.delimitedList(attribute_name).setResultsName('attr_list') + right
foreign_key_parser_old = build_foreign_key_parser_old()
foreign_key_parser = build_foreign_key_parser()
attribute_parser = build_attribute_parser()
index_parser = build_index_parser()
def is_foreign_key(line):
"""
:param line: a line from the table definition
:return: true if the line appears to be a foreign key definition
"""
arrow_position = line.find('->')
return arrow_position >= 0 and not any(c in line[:arrow_position] for c in '"#\'')
def compile_foreign_key(line, context, attributes, primary_key, attr_sql, foreign_key_sql, index_sql):
"""
:param line: a line from a table definition
:param context: namespace containing referenced objects
:param attributes: list of attribute names already in the declaration -- to be updated by this function
:param primary_key: None if the current foreign key is made from the dependent section. Otherwise it is the list
of primary key attributes thus far -- to be updated by the function
:param attr_sql: list of sql statements defining attributes -- to be updated by this function.
:param foreign_key_sql: list of sql statements specifying foreign key constraints -- to be updated by this function.
:param index_sql: list of INDEX declaration statements, duplicate or redundant indexes are ok.
"""
# Parse and validate
from .table import Table
from .expression import Projection
obsolete = False # See issue #436. Old style to be deprecated in a future release
try:
result = foreign_key_parser.parseString(line)
except pp.ParseException:
try:
result = foreign_key_parser_old.parseString(line)
except pp.ParseBaseException as err:
raise DataJointError('Parsing error in line "%s". %s.' % (line, err)) from None
else:
obsolete = True
try:
ref = eval(result.ref_table, context)
except NameError if obsolete else Exception:
raise DataJointError('Foreign key reference %s could not be resolved' % result.ref_table)
options = [opt.upper() for opt in result.options]
for opt in options: # check for invalid options
if opt not in {'NULLABLE', 'UNIQUE'}:
raise DataJointError('Invalid foreign key option "{opt}"'.format(opt=opt))
is_nullable = 'NULLABLE' in options
is_unique = 'UNIQUE' in options
if is_nullable and primary_key is not None:
raise DataJointError('Primary dependencies cannot be nullable in line "{line}"'.format(line=line))
if obsolete:
if not isinstance(ref, type) or not issubclass(ref, Table):
raise DataJointError('Foreign key reference %r must be a valid query' % result.ref_table)
if isinstance(ref, type) and issubclass(ref, Table):
ref = ref()
# check that dependency is of supported type
if (not isinstance(ref, (Table, Projection)) or len(ref.restriction) or
(isinstance(ref, Projection) and (not isinstance(ref._arg, Table) or len(ref._arg.restriction)))):
raise DataJointError('Dependency "%s" is not supported (yet). Use a base table or its projection.' %
result.ref_table)
if obsolete:
# for backward compatibility with old-style dependency declarations. See issue #436
if not isinstance(ref, Table):
DataJointError('Dependency "%s" is not supported. Check documentation.' % result.ref_table)
if not all(r in ref.primary_key for r in result.ref_attrs):
raise DataJointError('Invalid foreign key attributes in "%s"' % line)
try:
raise DataJointError('Duplicate attributes "{attr}" in "{line}"'.format(
attr=next(attr for attr in result.new_attrs if attr in attributes), line=line))
except StopIteration:
pass # the normal outcome
# Match the primary attributes of the referenced table to local attributes
new_attrs = list(result.new_attrs)
ref_attrs = list(result.ref_attrs)
# special case, the renamed attribute is implicit
if new_attrs and not ref_attrs:
if len(new_attrs) != 1:
raise DataJointError('Renamed foreign key must be mapped to the primary key in "%s"' % line)
if len(ref.primary_key) == 1:
# if the primary key has one attribute, allow implicit renaming
ref_attrs = ref.primary_key
else:
# if only one primary key attribute remains, then allow implicit renaming
ref_attrs = [attr for attr in ref.primary_key if attr not in attributes]
if len(ref_attrs) != 1:
raise DataJointError('Could not resolve which primary key attribute should be referenced in "%s"' % line)
if len(new_attrs) != len(ref_attrs):
raise DataJointError('Mismatched attributes in foreign key "%s"' % line)
if ref_attrs:
# convert to projected dependency
ref = ref.proj(**dict(zip(new_attrs, ref_attrs)))
# declare new foreign key attributes
base = ref._arg if isinstance(ref, Projection) else ref # base reference table
for attr, ref_attr in zip(ref.primary_key, base.primary_key):
if attr not in attributes:
attributes.append(attr)
if primary_key is not None:
primary_key.append(attr)
attr_sql.append(
base.heading[ref_attr].sql.replace(ref_attr, attr, 1).replace('NOT NULL ', '', int(is_nullable)))
# declare the foreign key
foreign_key_sql.append(
'FOREIGN KEY (`{fk}`) REFERENCES {ref} (`{pk}`) ON UPDATE CASCADE ON DELETE RESTRICT'.format(
fk='`,`'.join(ref.primary_key),
pk='`,`'.join(base.primary_key),
ref=base.full_table_name))
# declare unique index
if is_unique:
index_sql.append('UNIQUE INDEX ({attrs})'.format(attrs=','.join("`%s`" % attr for attr in ref.primary_key)))
def prepare_declare(definition, context):
# split definition into lines
definition = re.split(r'\s*\n\s*', definition.strip())
# check for optional table comment
table_comment = definition.pop(0)[1:].strip() if definition[0].startswith('#') else ''
if table_comment.startswith(':'):
raise DataJointError('Table comment must not start with a colon ":"')
in_key = True # parse primary keys
primary_key = []
attributes = []
attribute_sql = []
foreign_key_sql = []
index_sql = []
external_stores = []
for line in definition:
if not line or line.startswith('#'): # ignore additional comments
pass
elif line.startswith('---') or line.startswith('___'):
in_key = False # start parsing dependent attributes
elif is_foreign_key(line):
compile_foreign_key(line, context, attributes,
primary_key if in_key else None,
attribute_sql, foreign_key_sql, index_sql)
elif re.match(r'^(unique\s+)?index[^:]*$', line, re.I): # index
compile_index(line, index_sql)
else:
name, sql, store = compile_attribute(line, in_key, foreign_key_sql, context)
if store:
external_stores.append(store)
if in_key and name not in primary_key:
primary_key.append(name)
if name not in attributes:
attributes.append(name)
attribute_sql.append(sql)
return table_comment, primary_key, attribute_sql, foreign_key_sql, index_sql, external_stores
def declare(full_table_name, definition, context):
"""
Parse declaration and generate the SQL CREATE TABLE code
:param full_table_name: full name of the table
:param definition: DataJoint table definition
:param context: dictionary of objects that might be referred to in the table
:return: SQL CREATE TABLE statement, list of external stores used
"""
table_name = full_table_name.strip('`').split('.')[1]
if len(table_name) > MAX_TABLE_NAME_LENGTH:
raise DataJointError(
'Table name `{name}` exceeds the max length of {max_length}'.format(
name=table_name,
max_length=MAX_TABLE_NAME_LENGTH))
table_comment, primary_key, attribute_sql, foreign_key_sql, index_sql, external_stores = prepare_declare(
definition, context)
if not primary_key:
raise DataJointError('Table must have a primary key')
return (
'CREATE TABLE IF NOT EXISTS %s (\n' % full_table_name +
',\n'.join(attribute_sql + ['PRIMARY KEY (`' + '`,`'.join(primary_key) + '`)'] + foreign_key_sql + index_sql) +
'\n) ENGINE=InnoDB, COMMENT "%s"' % table_comment), external_stores
def _make_attribute_alter(new, old, primary_key):
"""
:param new: new attribute declarations
:param old: old attribute declarations
:param primary_key: primary key attributes
:return: list of SQL ALTER commands
"""
# parse attribute names
name_regexp = re.compile(r"^`(?P<name>\w+)`")
original_regexp = re.compile(r'COMMENT "{\s*(?P<name>\w+)\s*}')
matched = ((name_regexp.match(d), original_regexp.search(d)) for d in new)
new_names = OrderedDict((d.group('name'), n and n.group('name')) for d, n in matched)
old_names = [name_regexp.search(d).group('name') for d in old]
# verify that original names are only used once
renamed = set()
for v in new_names.values():
if v:
if v in renamed:
raise DataJointError('Alter attempted to rename attribute {%s} twice.' % v)
renamed.add(v)
# verify that all renamed attributes existed in the old definition
try:
raise DataJointError(
"Attribute {} does not exist in the original definition".format(
next(attr for attr in renamed if attr not in old_names)))
except StopIteration:
pass
# dropping attributes
to_drop = [n for n in old_names if n not in renamed and n not in new_names]
sql = ['DROP `%s`' % n for n in to_drop]
old_names = [name for name in old_names if name not in to_drop]
# add or change attributes in order
prev = None
for new_def, (new_name, old_name) in zip(new, new_names.items()):
if new_name not in primary_key:
after = None # if None, then must include the AFTER clause
if prev:
try:
idx = old_names.index(old_name or new_name)
except ValueError:
after = prev[0]
else:
if idx >= 1 and old_names[idx - 1] != (prev[1] or prev[0]):
after = prev[0]
if new_def not in old or after:
sql.append('{command} {new_def} {after}'.format(
command=("ADD" if (old_name or new_name) not in old_names else
"MODIFY" if not old_name else
"CHANGE `%s`" % old_name),
new_def=new_def,
after="" if after is None else "AFTER `%s`" % after))
prev = new_name, old_name
return sql
def alter(definition, old_definition, context):
"""
:param definition: new table definition
:param old_definition: current table definition
:param context: the context in which to evaluate foreign key definitions
:return: string SQL ALTER command, list of new stores used for external storage
"""
table_comment, primary_key, attribute_sql, foreign_key_sql, index_sql, external_stores = prepare_declare(
definition, context)
table_comment_, primary_key_, attribute_sql_, foreign_key_sql_, index_sql_, external_stores_ = prepare_declare(
old_definition, context)
# analyze differences between declarations
sql = list()
if primary_key != primary_key_:
raise NotImplementedError('table.alter cannot alter the primary key (yet).')
if foreign_key_sql != foreign_key_sql_:
raise NotImplementedError('table.alter cannot alter foreign keys (yet).')
if index_sql != index_sql_:
raise NotImplementedError('table.alter cannot alter indexes (yet)')
if attribute_sql != attribute_sql_:
sql.extend(_make_attribute_alter(attribute_sql, attribute_sql_, primary_key))
if table_comment != table_comment_:
sql.append('COMMENT="%s"' % table_comment)
return sql, [e for e in external_stores if e not in external_stores_]
def compile_index(line, index_sql):
match = index_parser.parseString(line)
index_sql.append('{unique} index ({attrs})'.format(
unique=match.unique,
attrs=','.join('`%s`' % a for a in match.attr_list)))
def substitute_special_type(match, category, foreign_key_sql, context):
"""
:param match: dict containing with keys "type" and "comment" -- will be modified in place
:param category: attribute type category from TYPE_PATTERN
:param foreign_key_sql: list of foreign key declarations to add to
:param context: context for looking up user-defined attribute_type adapters
"""
if category == 'UUID':
match['type'] = UUID_DATA_TYPE
elif category == 'INTERNAL_ATTACH':
match['type'] = 'LONGBLOB'
elif category in EXTERNAL_TYPES:
if category == 'FILEPATH' and not _support_filepath_types():
raise DataJointError("""
The filepath data type is disabled until complete validation.
To turn it on as experimental feature, set the environment variable
{env} = TRUE or upgrade datajoint.
""".format(env=FILEPATH_FEATURE_SWITCH))
match['store'] = match['type'].split('@', 1)[1]
match['type'] = UUID_DATA_TYPE
foreign_key_sql.append(
"FOREIGN KEY (`{name}`) REFERENCES `{{database}}`.`{external_table_root}_{store}` (`hash`) "
"ON UPDATE RESTRICT ON DELETE RESTRICT".format(external_table_root=EXTERNAL_TABLE_ROOT, **match))
elif category == 'ADAPTED':
adapter = get_adapter(context, match['type'])
match['type'] = adapter.attribute_type
category = match_type(match['type'])
if category in SPECIAL_TYPES:
# recursive redefinition from user-defined datatypes.
substitute_special_type(match, category, foreign_key_sql, context)
else:
assert False, 'Unknown special type'
def compile_attribute(line, in_key, foreign_key_sql, context):
"""
Convert attribute definition from DataJoint format to SQL
:param line: attribution line
:param in_key: set to True if attribute is in primary key set
:param foreign_key_sql: the list of foreign key declarations to add to
:param context: context in which to look up user-defined attribute type adapterss
:returns: (name, sql, is_external) -- attribute name and sql code for its declaration
"""
try:
match = attribute_parser.parseString(line + '#', parseAll=True)
except pp.ParseException as err:
raise DataJointError('Declaration error in position {pos} in line:\n {line}\n{msg}'.format(
line=err.args[0], pos=err.args[1], msg=err.args[2])) from None
match['comment'] = match['comment'].rstrip('#')
if 'default' not in match:
match['default'] = ''
match = {k: v.strip() for k, v in match.items()}
match['nullable'] = match['default'].lower() == 'null'
if match['nullable']:
if in_key:
raise DataJointError('Primary key attributes cannot be nullable in line "%s"' % line)
match['default'] = 'DEFAULT NULL' # nullable attributes default to null
else:
if match['default']:
quote = (match['default'].split('(')[0].upper() not in CONSTANT_LITERALS
and match['default'][0] not in '"\'')
match['default'] = 'NOT NULL DEFAULT ' + ('"%s"' if quote else "%s") % match['default']
else:
match['default'] = 'NOT NULL'
match['comment'] = match['comment'].replace('"', '\\"') # escape double quotes in comment
if match['comment'].startswith(':'):
raise DataJointError('An attribute comment must not start with a colon in comment "{comment}"'.format(**match))
category = match_type(match['type'])
if category in SPECIAL_TYPES:
match['comment'] = ':{type}:{comment}'.format(**match) # insert custom type into comment
substitute_special_type(match, category, foreign_key_sql, context)
if category in SERIALIZED_TYPES and match['default'] not in {'DEFAULT NULL', 'NOT NULL'}:
raise DataJointError(
'The default value for a blob or attachment attributes can only be NULL in:\n{line}'.format(line=line))
sql = ('`{name}` {type} {default}' + (' COMMENT "{comment}"' if match['comment'] else '')).format(**match)
return match['name'], sql, match.get('store')