Changed SQLAlchemy record manager to write many records in one execute.
@@ -57,15 +57,10 @@ class SQLAlchemyRecordManager(SQLRecordManager):
return compiled
def write_records(self, records, tracking_kwargs=None):
- try:
- with self.session.bind.begin() as connection:
- if tracking_kwargs:
- # Insert tracking record.
- params = {c: tracking_kwargs[c] for c in self.tracking_record_field_names}
- connection.execute(self.insert_tracking_record, **params)
-
+ all_params = []
+ statement = None
if records:
- # Insert event and notification records.
+ # Prepare to insert event and notification records.
statement = self.insert_values
if hasattr(self.record_class, 'id'):
all_ids = set((r.id for r in records))
@@ -83,7 +78,6 @@ class SQLAlchemyRecordManager(SQLRecordManager):
# application_name. We need values and don't have them.
raise ProgrammingError("record ID not set when required")
- all_params = []
for record in records:
# Params for stored item itself (e.g. event).
params = {
@@ -98,15 +92,40 @@ class SQLAlchemyRecordManager(SQLRecordManager):
if hasattr(self.record_class, 'pipeline_id'):
params['pipeline_id'] = self.pipeline_id
if hasattr(self.record_class, 'id'):
+ if record.id == ' ':
+ params['id'] = None
+ params['pipeline_id'] = None # clumsy, a bit
+ else:
params['id'] = record.id
if hasattr(record, 'causal_dependencies'):
params['causal_dependencies'] = record.causal_dependencies
all_params.append(params)
+ elif not tracking_kwargs:
+ # Don't bother if there is nothing to write.
+ return
+
+ # if tracking_kwargs:
+ # tracking_msg = str(id(self)) + (
+ # " {pipeline_id}:{notification_id} {application_name} <- {upstream_application_name}"
+ # ).format(**tracking_kwargs)
+ # else:
+ # tracking_msg = ''
+ try:
+
+ with self.session.bind.begin() as connection:
+ if tracking_kwargs:
+ # Insert tracking record.
+ connection.execute(self.insert_tracking_record, **tracking_kwargs)
+ # print("Inserted tracking: {}".format(tracking_msg))
+
+ if all_params:
connection.execute(statement, all_params)
except IntegrityError as e:
+ # if 'notification_tracking' in str(e):
+ # print("Failed tracking: {}: {}".format(tracking_msg, e))
self.raise_record_integrity_error(e)
except DBAPIError as e:
@@ -196,7 +215,7 @@ class SQLAlchemyRecordManager(SQLRecordManager):
query = query.filter(position_field == position)
return query.one()
except (NoResultFound, MultipleResultsFound):
- raise IndexError
+ raise IndexError(self.application_name, sequence_id, position)
def filter_by(self, **kwargs):
return self.orm_query().filter_by(**kwargs)
