FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Merge pull request #377 from dimitri-yatsenko/master · MMathisLab/datajoint-python@76e3d56 · GitHub

  1. MMathisLab
  2. datajoint-python

Repository navigation

Commit 76e3d56

Browse files
eywalker
authored
Merge pull request datajoint#377 from dimitri-yatsenko/master
Add external storage feature and multiple bug fixes
2 parents 8ea75a6 + 7627569 commit 76e3d56

27 files changed

Lines changed: 721 additions & 249 deletions

‎.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,3 +16,5 @@ dj_local_conf.json
1616
build/
1717
.coverage
1818
./tests/.coverage
19+
./tests/dj-store/*
20+
*.log

‎datajoint/autopopulate.py‎

Lines changed: 44 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
"""autopopulate containing the dj.AutoPopulate class. See `dj.AutoPopulate` for more info."""
22
import logging
33
import datetime
4+
import traceback
45
import random
6+
from tqdm import tqdm
7+
from itertools import count
58
from pymysql import OperationalError
6-
from .relational_operand import RelationalOperand, AndList
9+
from .relational_operand import RelationalOperand, AndList, U
710
from . import DataJointError
811
from . import key as KEY
912
from .base_relation import FreeRelation
@@ -18,15 +21,15 @@ class AutoPopulate:
1821
"""
1922
AutoPopulate is a mixin class that adds the method populate() to a Relation class.
2023
Auto-populated relations must inherit from both Relation and AutoPopulate,
21-
must define the property `key_source`, and must define the callback method _make_tuples.
24+
must define the property `key_source`, and must define the callback method `make`.
2225
"""
2326
_key_source = None
2427

2528
@property
2629
def key_source(self):
2730
"""
2831
:return: the relation whose primary key values are passed, sequentially, to the
29-
`_make_tuples` method when populate() is called.The default value is the
32+
``make`` method when populate() is called.The default value is the
3033
join of the parent relations. Users may override to change the granularity
3134
or the scope of populate() calls.
3235
"""
@@ -40,13 +43,15 @@ def key_source(self):
4043
self._key_source *= FreeRelation(self.connection, parents.pop(0)).proj()
4144
return self._key_source
4245

43-
def _make_tuples(self, key):
46+
47+
def make(self, key):
4448
"""
45-
Derived classes must implement method _make_tuples that fetches data from tables that are
49+
Derived classes must implement method `make` that fetches data from tables that are
4650
above them in the dependency hierarchy, restricting by the given key, computes dependent
4751
attributes, and inserts the new tuples into self.
4852
"""
49-
raise NotImplementedError('Subclasses of AutoPopulate must implement the method "_make_tuples"')
53+
raise NotImplementedError('Subclasses of AutoPopulate must implement the method `make`')
54+
5055

5156
@property
5257
def target(self):
@@ -63,16 +68,19 @@ def _job_key(self, key):
6368
"""
6469
return key
6570

66-
def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False, order="original", limit=None):
71+
def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False,
72+
order="original", limit=None, max_calls=None, display_progress=False):
6773
"""
68-
rel.populate() calls rel._make_tuples(key) for every primary key in self.key_source
74+
rel.populate() calls rel.make(key) for every primary key in self.key_source
6975
for which there is not already a tuple in rel.
7076
7177
:param restrictions: a list of restrictions each restrict (rel.key_source - target.proj())
7278
:param suppress_errors: suppresses error if true
7379
:param reserve_jobs: if true, reserves job to populate in asynchronous fashion
7480
:param order: "original"|"reverse"|"random" - the order of execution
75-
:param limit: if not None, populates at max that many keys
81+
:param display_progress: if True, report progress_bar
82+
:param limit: if not None, checks at most that many keys
83+
:param max_calls: if not None, populates at max that many keys
7684
"""
7785
if self.connection.in_transaction:
7886
raise DataJointError('Populate cannot be called during a transaction.')
@@ -84,11 +92,20 @@ def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False, ord
8492
todo = self.key_source
8593
if not isinstance(todo, RelationalOperand):
8694
raise DataJointError('Invalid key_source value')
87-
todo = todo.proj() & AndList(restrictions)
95+
todo = (todo & AndList(restrictions)).proj()
8896

89-
error_list = [] if suppress_errors else None
97+
# raise error if the populated target lacks any attributes from the primary key of key_source
98+
try:
99+
raise DataJointError(
100+
'The populate target lacks attribute %s from the primary key of key_source' % next(
101+
name for name in todo.heading if name not in self.target.heading))
102+
except StopIteration:
103+
pass
104+
105+
todo -= self.target
90106

91-
jobs = self.connection.jobs[self.target.database] if reserve_jobs else None
107+
error_list = [] if suppress_errors else None
108+
jobs = self.connection.schemas[self.target.database].jobs if reserve_jobs else None
92109

93110
# define and setup signal handler for SIGTERM
94111
if reserve_jobs:
@@ -97,15 +114,20 @@ def handler(signum, frame):
97114
raise SystemExit('SIGTERM received')
98115
old_handler = signal.signal(signal.SIGTERM, handler)
99116

100-
todo -= self.target
101117
keys = todo.fetch(KEY, limit=limit)
102118
if order == "reverse":
103119
keys.reverse()
104120
elif order == "random":
105121
random.shuffle(keys)
106122

123+
call_count = count()
107124
logger.info('Found %d keys to populate' % len(keys))
108-
for key in keys:
125+
126+
make = self._make_tuples if hasattr(self, '_make_tuples') else self.make
127+
128+
for key in (tqdm(keys) if display_progress else keys):
129+
if max_calls is not None and call_count >= max_calls:
130+
break
109131
if not reserve_jobs or jobs.reserve(self.target.table_name, self._job_key(key)):
110132
self.connection.start_transaction()
111133
if key in self.target: # already populated
@@ -114,8 +136,9 @@ def handler(signum, frame):
114136
jobs.complete(self.target.table_name, self._job_key(key))
115137
else:
116138
logger.info('Populating: ' + str(key))
139+
next(call_count)
117140
try:
118-
self._make_tuples(dict(key))
141+
make(dict(key))
119142
except (KeyboardInterrupt, SystemExit, Exception) as error:
120143
try:
121144
self.connection.cancel_transaction()
@@ -124,7 +147,8 @@ def handler(signum, frame):
124147
if reserve_jobs:
125148
# show error name and error message (if any)
126149
error_message = ': '.join([error.__class__.__name__, str(error)]).strip(': ')
127-
jobs.error(self.target.table_name, self._job_key(key), error_message=error_message)
150+
jobs.error(self.target.table_name, self._job_key(key),
151+
error_message=error_message, error_stack=traceback.format_exc())
128152

129153
if not suppress_errors or isinstance(error, SystemExit):
130154
raise
@@ -139,17 +163,18 @@ def handler(signum, frame):
139163
# place back the original signal handler
140164
if reserve_jobs:
141165
signal.signal(signal.SIGTERM, old_handler)
142-
143166
return error_list
144167

145168
def progress(self, *restrictions, display=True):
146169
"""
147170
report progress of populating this table
148171
:return: remaining, total -- tuples to be populated
149172
"""
150-
todo = self.key_source & AndList(restrictions)
173+
todo = (self.key_source & AndList(restrictions)).proj()
174+
if any(name not in self.target.heading for name in todo.heading):
175+
raise DataJointError('The populated target must have all the attributes of the key source')
151176
total = len(todo)
152-
remaining = len(todo.proj() - self.target)
177+
remaining = len(todo - self.target)
153178
if display:
154179
print('%-20s' % self.__class__.__name__,
155180
'Completed %d of %d (%2.1f%%) %s' % (

‎datajoint/base_relation.py‎

Lines changed: 60 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ class BaseRelation(RelationalOperand):
3030
_context = None
3131
database = None
3232
_log_ = None
33+
_external_table = None
3334

3435
# -------------- required by RelationalOperand ----------------- #
3536
@property
@@ -50,14 +51,21 @@ def context(self):
5051

5152
def declare(self):
5253
"""
53-
Loads the table heading. If the table is not declared, use self.definition to declare
54+
Use self.definition to declare the table in the database
5455
"""
5556
try:
56-
self.connection.query(
57-
declare(self.full_table_name, self.definition, self._context))
57+
sql, uses_external = declare(self.full_table_name, self.definition, self._context)
58+
if uses_external:
59+
# trigger the creation of the external hash lookup for the current schema
60+
external_table = self.connection.schemas[self.database].external_table
61+
sql = sql.format(external_table=external_table.full_table_name)
62+
self.connection.query(sql)
5863
except pymysql.OperationalError as error:
64+
# skip if no create privilege
5965
if error.args[0] == server_error_codes['command denied']:
6066
logger.warning(error.args[1])
67+
else:
68+
raise
6169
else:
6270
self._log('Declared ' + self.full_table_name)
6371

@@ -114,6 +122,12 @@ def _log(self):
114122
self._log_ = Log(self.connection, database=self.database)
115123
return self._log_
116124

125+
@property
126+
def external_table(self):
127+
if self._external_table is None:
128+
self._external_table = self.connection.schemas[self.database].external_table
129+
return self._external_table
130+
117131
def insert1(self, row, **kwargs):
118132
"""
119133
Insert one data record or one Mapping (like a dict).
@@ -142,60 +156,29 @@ def insert(self, rows, replace=False, skip_duplicates=False, ignore_extra_fields
142156
warnings.warn('Use of `ignore_errors` in `insert` and `insert1` is deprecated. Use try...except... '
143157
'to explicitly handle any errors', stacklevel=2)
144158

145-
# handle query safely - if skip_duplicates=True, wraps the query with transaction and checks for warning
146-
def safe_query(*args, **kwargs):
147-
if skip_duplicates:
148-
# check if there is already an open transaction
149-
open_transaction = self.connection.in_transaction
150-
if not open_transaction:
151-
self.connection.start_transaction()
152-
try:
153-
with warnings.catch_warnings(record=True) as ws:
154-
warnings.simplefilter('always')
155-
self.connection.query(*args, suppress_warnings=False, **kwargs)
156-
for w in ws:
157-
if w.message.args[0] != server_error_codes['duplicate entry']:
158-
raise InternalError(w.message.args)
159-
except:
160-
if not open_transaction:
161-
try:
162-
self.connection.cancel_transaction()
163-
except OperationalError:
164-
pass
165-
raise
166-
else:
167-
if not open_transaction:
168-
self.connection.commit_transaction()
169-
else:
170-
self.connection.query(*args, **kwargs)
171-
159+
heading = self.heading
172160
if isinstance(rows, RelationalOperand):
173-
# INSERT FROM SELECT - build alternate field-narrowing query (only) when needed
174-
if ignore_extra_fields and not all(name in self.heading.names for name in rows.heading.names):
175-
query = 'INSERT{ignore} INTO {table} ({fields}) SELECT {fields} FROM ({select}) as `__alias`'.format(
176-
ignore=" IGNORE" if skip_duplicates else "",
177-
table=self.full_table_name,
178-
fields='`'+'`,`'.join(self.heading.names)+'`',
179-
select=rows.make_sql())
180-
else:
181-
query = 'INSERT{ignore} INTO {table} ({fields}) {select}'.format(
182-
ignore=" IGNORE" if skip_duplicates else "",
161+
# insert from select
162+
if not ignore_extra_fields:
163+
try:
164+
raise DataJointError(
165+
"Attribute %s not found. To ignore extra attributes in insert, set ignore_extra_fields=True." %
166+
next(name for name in rows.heading if name not in heading))
167+
except StopIteration:
168+
pass
169+
fields = list(name for name in heading if name in rows.heading)
170+
171+
query = '{command} INTO {table} ({fields}) {select}{duplicate}'.format(
172+
command='REPLACE' if replace else 'INSERT',
173+
fields='`' + '`,`'.join(fields) + '`',
183174
table=self.full_table_name,
184-
fields='`'+'`,`'.join(rows.heading.names)+'`',
185-
select=rows.make_sql())
186-
try:
187-
safe_query(query)
188-
except (InternalError, IntegrityError) as err:
189-
if err.args[0] == server_error_codes['unknown column']:
190-
# args[1] -> Unknown column 'extra' in 'field list'
191-
raise DataJointError('{} : To ignore extra fields, set ignore_extra_fields=True in insert.'.format(err.args[1]))
192-
elif err.args[0] == server_error_codes['duplicate entry']:
193-
raise DataJointError('{} : To ignore duplicate entries, set skip_duplicates=True in insert.'.format(err.args[1]))
194-
else:
195-
raise
175+
select=rows.make_sql(select_fields=fields),
176+
duplicate=(' ON DUPLICATE KEY UPDATE `{pk}`=`{pk}`'.format(pk=self.primary_key[0])
177+
if skip_duplicates else '')
178+
)
179+
self.connection.query(query)
196180
return
197181

198-
heading = self.heading
199182
if heading.attributes is None:
200183
logger.warning('Could not access table {table}'.format(table=self.full_table_name))
201184
return
@@ -218,16 +201,18 @@ def make_placeholder(name, value):
218201
"""
219202
if ignore_extra_fields and name not in heading:
220203
return None
221-
if heading[name].is_blob:
222-
value = pack(value)
223-
placeholder = '%s'
204+
if heading[name].is_external:
205+
placeholder, value = '%s', self.external_table.put(heading[name].type, value)
206+
elif heading[name].is_blob:
207+
if value is None:
208+
placeholder, value = 'NULL', None
209+
else:
210+
placeholder, value = '%s', pack(value)
224211
elif heading[name].numeric:
225212
if value is None or value == '' or np.isnan(np.float(value)): # nans are turned into NULLs
226-
placeholder = 'NULL'
227-
value = None
213+
placeholder, value = 'NULL', None
228214
else:
229-
placeholder = '%s'
230-
value = str(int(value) if isinstance(value, bool) else value)
215+
placeholder, value = '%s', (str(int(value) if isinstance(value, bool) else value))
231216
else:
232217
placeholder = '%s'
233218
return name, placeholder, value
@@ -284,13 +269,15 @@ def check_fields(fields):
284269
rows = list(make_row_to_insert(row) for row in rows)
285270
if rows:
286271
try:
287-
safe_query(
288-
"{command} INTO {destination}(`{fields}`) VALUES {placeholders}".format(
289-
command='REPLACE' if replace else 'INSERT IGNORE' if skip_duplicates else 'INSERT',
290-
destination=self.from_clause,
291-
fields='`,`'.join(field_list),
292-
placeholders=','.join('(' + ','.join(row['placeholders']) + ')' for row in rows)),
293-
args=list(itertools.chain.from_iterable((v for v in r['values'] if v is not None) for r in rows)))
272+
query = "{command} INTO {destination}(`{fields}`) VALUES {placeholders}{duplicate}".format(
273+
command='REPLACE' if replace else 'INSERT',
274+
destination=self.from_clause,
275+
fields='`,`'.join(field_list),
276+
placeholders=','.join('(' + ','.join(row['placeholders']) + ')' for row in rows),
277+
duplicate=(' ON DUPLICATE KEY UPDATE `{pk}`=`{pk}`'.format(pk=self.primary_key[0])
278+
if skip_duplicates else ''))
279+
self.connection.query(query, args=list(
280+
itertools.chain.from_iterable((v for v in r['values'] if v is not None) for r in rows)))
294281
except (OperationalError, InternalError, IntegrityError) as err:
295282
if err.args[0] == server_error_codes['command denied']:
296283
raise DataJointError('Command denied: %s' % err.args[1]) from None
@@ -304,7 +291,6 @@ def check_fields(fields):
304291
else:
305292
raise
306293

307-
308294
def delete_quick(self):
309295
"""
310296
Deletes the table without cascading and without user prompt. If this table has any dependent
@@ -338,7 +324,7 @@ def delete(self):
338324
# restrict by self
339325
if self.restrictions:
340326
restrict_by_me.add(self.full_table_name)
341-
restrictions[self.full_table_name].append(self.restrictions.simplify()) # copy own restrictions
327+
restrictions[self.full_table_name].append(self.restrictions) # copy own restrictions
342328
# restrict by renamed nodes
343329
restrict_by_me.update(table for table in delete_list if table.isdigit()) # restrict by all renamed nodes
344330
# restrict by tables restricted by a non-primary semijoin
@@ -454,7 +440,7 @@ def describe(self):
454440
if attr.name in fk_props['referencing_attributes']:
455441
do_include = False
456442
if attributes_thus_far.issuperset(fk_props['referencing_attributes']):
457-
# simple foreign keys
443+
# simple foreign key
458444
parents.pop(parent_name)
459445
if not parent_name.isdigit():
460446
definition += '-> {class_name}\n'.format(
@@ -473,9 +459,10 @@ def describe(self):
473459
attributes_declared.update(fk_props['referencing_attributes'])
474460
if do_include:
475461
attributes_declared.add(attr.name)
462+
name = attr.name.lstrip('_') # for external
476463
definition += '%-20s : %-28s # %s\n' % (
477-
attr.name if attr.default is None else '%s=%s' % (attr.name, attr.default),
478-
'%s%s' % (attr.type, 'auto_increment' if attr.autoincrement else ''), attr.comment)
464+
name if attr.default is None else '%s=%s' % (name, attr.default),
465+
'%s%s' % (attr.type, ' auto_increment' if attr.autoincrement else ''), attr.comment)
479466
print(definition)
480467
return definition
481468

@@ -547,7 +534,7 @@ def lookup_class_name(name, context, depth=3):
547534
except AttributeError:
548535
pass # not a UserRelation -- cannot have part tables.
549536
else:
550-
for part in (getattr(member, p) for p in parts):
537+
for part in (getattr(member, p) for p in parts if hasattr(member, p)):
551538
if inspect.isclass(part) and issubclass(part, BaseRelation) and part.full_table_name == name:
552539
return '.'.join([node['context_name'], member_name, part.__name__]).lstrip('.')
553540
elif node['depth'] > 0 and inspect.ismodule(member) and member.__name__ != 'datajoint':

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL