datajoint-python/datajoint/declare.py at main · atlab/datajoint-python · GitHub

Repository navigation

History

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(

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')

datajoint-python/datajoint/declare.py at main · atlab/datajoint-python · GitHub

Repository navigation

Expand file tree

Breadcrumbs

declare.py

Latest commit

History

Breadcrumbs

declare.py

File metadata and controls