`` element. Note that ``i`` will be a 1-based index value for
the row within its table. The meaning of ``row`` is basically not
defined; it depends on the type of data the grid deals with.
"""
if callable(self.row_attrs):
attrs = self.row_attrs(row, i)
else:
attrs = dict(self.row_attrs)
if self.mergeable:
attrs['data-uuid'] = row.uuid
return attrs
def get_cell_attrs(self, row, column):
"""
Returns a dictionary of HTML attributes which should be applied to the
``| `` element in which the given row and column "intersect".
"""
if callable(self.cell_attrs):
return self.cell_attrs(row, column)
return self.cell_attrs
def get_main_actions(self):
"""
Return a list of 'main' actions for the grid.
"""
actions = []
if self.viewable and self.has_perm('view'):
actions.append(self.make_grid_action_view())
return actions
def make_grid_action_view(self):
return self.make_action('view', icon='eye', url=self.default_view_url())
def default_view_url(self):
if self.use_index_links:
return self.get_view_index_url
def get_view_index_url(self, row, i):
route = '{}.view_index'.format(self.get_route_prefix())
return '{}?index={}'.format(self.request.route_url(route), self.first_visible_grid_index + i - 1)
def get_more_actions(self):
"""
Return a list of 'more' actions for the grid.
"""
actions = []
# Edit
if self.editable and self.has_perm('edit'):
actions.append(self.make_grid_action_edit())
# Delete
if self.deletable and self.has_perm('delete'):
actions.append(self.make_grid_action_delete())
return actions
def make_grid_action_edit(self):
return self.make_action('edit', icon='edit', url=self.default_edit_url)
def make_grid_action_clone(self):
return self.make_action('clone', icon='object-ungroup',
url=self.default_clone_url)
def make_grid_action_delete(self):
kwargs = {}
if self.delete_confirm == 'simple':
kwargs['click_handler'] = 'deleteObject'
return self.make_action('delete', icon='trash', url=self.default_delete_url, **kwargs)
def default_edit_url(self, obj, i=None):
"""
Return the default "edit" URL for the given object, if
applicable. This first checks :meth:`editable_instance()` for
the object, and will only return a URL if the object is deemed
editable.
:param obj: A top-level record/object, of the type normally
handled by this master view.
:param i: Optional row index within a grid.
:returns: The "edit object" URL as string, or ``None``.
"""
if self.editable_instance(obj):
return self.request.route_url('{}.edit'.format(self.get_route_prefix()),
**self.get_action_route_kwargs(obj))
def default_clone_url(self, row, i=None):
return self.request.route_url('{}.clone'.format(self.get_route_prefix()),
**self.get_action_route_kwargs(row))
def default_delete_url(self, row, i=None):
if self.deletable_instance(row):
return self.request.route_url('{}.delete'.format(self.get_route_prefix()),
**self.get_action_route_kwargs(row))
def make_action(self, key, url=None, factory=None, **kwargs):
"""
Make a new :class:`GridAction` instance for the current grid.
"""
if url is None:
route = '{}.{}'.format(self.get_route_prefix(), key)
url = lambda r, i: self.request.route_url(route, **self.get_action_route_kwargs(r))
if not factory:
factory = grids.GridAction
return factory(key, url=url, **kwargs)
def get_action_route_kwargs(self, obj):
"""
Get a dict of route kwargs for the given object.
This is called from various other "convenience" URL
generators, e.g. :meth:`default_edit_url()`.
It inspects the given object, as well as the "model key" (as
returned by :meth:`get_model_key()`), and returns a dict of
appropriate route kwargs for the object.
Most typically, the model key is just ``uuid`` and so this
would effectively return ``{'uuid': obj.uuid}``.
But composite model keys are supported too, so if the model
key is ``(parent_id, child_id)`` this might instead return
``{'parent_id': obj.parent_id, 'child_id': obj.child_id}``.
Such kwargs would then be fed into ``route_url()`` as needed,
for example to get a "view product URL"::
kw = self.get_action_route_kwargs(product)
url = self.request.route_url('products.view', **kw)
:param obj: A top-level record/object, of the type normally
handled by this master view.
:returns: A dict of route kwargs for the object.
"""
keys = self.get_model_key(as_tuple=True)
if keys:
try:
return dict([(key, obj[key])
for key in keys])
except TypeError:
return dict([(key, getattr(obj, key))
for key in keys])
# TODO: sanity check, is the above all we need..?
log.warning("yes we still do the code below sometimes")
try:
mapper = orm.object_mapper(obj)
except orm.exc.UnmappedInstanceError:
try:
if isinstance(self.model_key, str):
return {self.model_key: obj[self.model_key]}
return dict([(key, obj[key])
for key in self.model_key])
except TypeError:
return {self.model_key: getattr(obj, self.model_key)}
else:
pkeys = get_primary_keys(obj)
keys = list(pkeys)
values = [getattr(obj, k) for k in keys]
return dict(zip(keys, values))
def get_data(self, session=None):
"""
Generate the base data set for the grid. This typically will be a
SQLAlchemy query against the view's model class, but subclasses may
override this to support arbitrary data sets.
Note that if your view is typical and uses a SA model, you should not
override this methid, but override :meth:`query()` instead.
"""
if session is None:
session = self.Session()
return self.query(session)
def query(self, session):
"""
Produce the initial/base query for the master grid. By default this is
simply a query against the model class, but you may override as
necessary to apply any sort of pre-filtering etc. This is useful if
say, you don't ever want to show records of a certain type to non-admin
users. You would modify the base query to hide what you wanted,
regardless of the user's filter selections.
"""
model_class = self.get_model_class()
query = session.query(model_class)
# only show "local only" objects, unless global access allowed
if self.secure_global_objects:
if not self.has_perm('view_global'):
query = query.filter(model_class.local_only == True)
for supp in self.iter_view_supplements():
query = supp.get_grid_query(query)
return query
def get_effective_query(self, session=None, **kwargs):
return self.get_effective_data(session=session, **kwargs)
# TODO: should rename to checkable?
def checkbox(self, instance):
"""
Returns a boolean indicating whether ot not a checkbox should be
rendererd for the given row. Default implementation returns ``True``
in all cases.
"""
return True
def checked(self, instance):
"""
Returns a boolean indicating whether ot not a checkbox should be
checked by default, for the given row. Default implementation returns
``False`` in all cases.
"""
return False
def download_results_path(self, user_uuid, filename=None,
typ='results', makedirs=False):
"""
Returns an absolute path for the "results" data file, specific to the
given user UUID.
"""
route_prefix = self.get_route_prefix()
path = os.path.join(self.rattail_config.datadir(), 'downloads',
typ, route_prefix,
user_uuid[:2], user_uuid[2:])
if makedirs and not os.path.exists(path):
os.makedirs(path)
if filename:
path = os.path.join(path, filename)
return path
def download_results_filename(self, fmt):
"""
Must return an appropriate "download results" filename for the given
format. E.g. ``'products.csv'``
"""
route_prefix = self.get_route_prefix()
if fmt == 'csv':
return '{}.csv'.format(route_prefix)
if fmt == 'xlsx':
return '{}.xlsx'.format(route_prefix)
def download_results_supported_formats(self):
# TODO: default formats should be configurable?
return OrderedDict([
('xlsx', "Excel (XLSX)"),
('csv', "CSV"),
])
def download_results_default_format(self):
# TODO: default format should be configurable
return 'xlsx'
def download_results(self):
"""
View for saving current (filtered) data results into a file, and
downloading that file.
"""
route_prefix = self.get_route_prefix()
user_uuid = self.request.user.uuid
# POST means generate a new results file for download
if self.request.method == 'POST':
# make sure a valid format was requested
supported = self.download_results_supported_formats()
if not supported:
self.request.session.flash("There are no supported download formats!",
'error')
return self.redirect(self.get_index_url())
fmt = self.request.POST.get('fmt')
if not fmt:
fmt = self.download_results_default_format() or list(supported)[0]
if fmt not in supported:
self.request.session.flash("Unsupported download format: {}".format(fmt),
'error')
return self.redirect(self.get_index_url())
# parse field list if one was given
fields = self.request.POST.get('fields')
if fields:
fields = fields.split(',')
# start thread to actually do work / report progress
key = '{}.download_results'.format(route_prefix)
progress = self.make_progress(key)
results = self.get_effective_data()
thread = Thread(target=self.download_results_thread,
args=(results, fmt, fields, user_uuid, progress))
thread.start()
# show user the progress page
return self.render_progress(progress, {
'cancel_url': self.get_index_url(),
'cancel_msg': "Download was canceled.",
})
# not POST, so just download a file (if specified)
filename = self.request.GET.get('filename')
if not filename:
return self.redirect(self.get_index_url())
path = self.download_results_path(user_uuid, filename)
return self.file_response(path)
def download_results_thread(self, results, fmt, fields, user_uuid, progress):
"""
Thread target, which invokes :meth:`download_results_generate()` to
officially generate the data file which is then to be downloaded.
"""
route_prefix = self.get_route_prefix()
session = self.make_isolated_session()
try:
# create folder(s) for output; make sure file doesn't exist
filename = self.download_results_filename(fmt)
path = self.download_results_path(user_uuid, filename, makedirs=True)
if os.path.exists(path):
os.remove(path)
# generate file for download
results = results.with_session(session)
self.download_results_setup(fields, progress=progress)
self.download_results_generate(session, results, path, fmt, fields,
progress=progress)
session.commit()
except Exception as error:
msg = "failed to generate results file for download!"
log.warning(msg, exc_info=True)
session.rollback()
if progress:
progress.session.load()
progress.session['error'] = True
progress.session['error_msg'] = "{}: {}".format(
msg, simple_error(error))
progress.session.save()
return
finally:
session.close()
if progress:
progress.session.load()
progress.session['complete'] = True
progress.session['success_url'] = self.get_index_url()
progress.session['extra_session_bits'] = {
'{}.results.generated'.format(route_prefix): path,
}
progress.session.save()
def download_results_setup(self, fields, progress=None):
"""
Perform any up-front caching or other setup required, just prior to
generating a new results data file for download.
"""
def download_results_generate(self, session, results, path, fmt, fields, progress=None):
"""
This method is responsible for actually generating the data file for a
"download results" operation, according to the given params.
"""
if fmt == 'csv':
csv_file = open(path, 'wt', encoding='utf_8')
writer = csv.DictWriter(csv_file, fields)
writer.writeheader()
def write(obj, i):
data = self.download_results_normalize(obj, fields, fmt=fmt)
csvrow = self.download_results_coerce_csv(data, fields)
writer.writerow(csvrow)
self.progress_loop(write, results.all(), progress,
message="Writing data to CSV file")
csv_file.close()
elif fmt == 'xlsx':
writer = ExcelWriter(path, fields,
sheet_title=self.get_model_title_plural())
writer.write_header()
xlrows = []
def write(obj, i):
data = self.download_results_normalize(obj, fields, fmt=fmt)
row = self.download_results_coerce_xlsx(data, fields)
xlrow = [row[field] for field in fields]
xlrows.append(xlrow)
self.progress_loop(write, results.all(), progress,
message="Collecting data for Excel")
def finalize(x, i):
writer.write_rows(xlrows)
writer.auto_freeze()
writer.auto_filter()
writer.auto_resize()
writer.save()
self.progress_loop(finalize, [1], progress,
message="Writing Excel file to disk")
def download_results_fields_available(self, **kwargs):
"""
Return the list of fields which are *available* to be written to
download file. Default field list will be constructed from the
underlying table columns.
"""
fields = []
mapper = orm.class_mapper(self.model_class)
for prop in mapper.iterate_properties:
if isinstance(prop, orm.ColumnProperty):
fields.append(prop.key)
return fields
def download_results_fields_default(self, fields, **kwargs):
"""
Return the default list of fields to be written to download file.
Unless you override, all "available" fields will be included by
default.
"""
return fields
def download_results_normalize(self, obj, fields, **kwargs):
"""
Normalize the given object into a data dict, for use when writing to
the results file for download.
"""
data = {}
for field in fields:
value = getattr(obj, field, None)
# make timestamps zone-aware
if isinstance(value, datetime.datetime):
value = localtime(self.rattail_config, value,
from_utc=not self.has_local_times)
data[field] = value
return data
def download_results_coerce_csv(self, data, fields, **kwargs):
"""
Coerce the given data dict record, to a "row" dict suitable for use
when writing directly to CSV file. Each value in the dict should be a
string type.
"""
csvrow = dict(data)
for field in fields:
value = csvrow.get(field)
if value is None:
value = ''
else:
value = str(value)
csvrow[field] = value
return csvrow
def download_results_coerce_xlsx(self, data, fields, **kwargs):
"""
Coerce the given data dict record, to a "row" dict suitable for use
when writing directly to XLSX file.
"""
data = dict(data)
for key in data:
value = data[key]
# make timestamps local, "zone-naive"
if isinstance(value, datetime.datetime):
value = localtime(self.rattail_config, value, tzinfo=False)
data[key] = value
return data
def results_csv(self):
"""
Download current list results as CSV.
"""
results = self.get_effective_data()
# start thread to actually do work / generate progress data
route_prefix = self.get_route_prefix()
key = '{}.results_csv'.format(route_prefix)
progress = self.make_progress(key)
thread = Thread(target=self.results_csv_thread,
args=(results, self.request.user.uuid, progress))
thread.start()
# send user to progress page
return self.render_progress(progress, {
'cancel_url': self.get_index_url(),
'cancel_msg': "CSV download was canceled.",
})
def results_csv_session(self):
return self.make_isolated_session()
def results_csv_thread(self, results, user_uuid, progress):
"""
Thread target, responsible for actually generating the CSV file which
is to be presented for download.
"""
route_prefix = self.get_route_prefix()
session = self.results_csv_session()
try:
# create folder(s) for output; make sure file doesn't exist
path = os.path.join(self.rattail_config.datadir(), 'downloads',
'results-csv', route_prefix,
user_uuid[:2], user_uuid[2:])
if not os.path.exists(path):
os.makedirs(path)
path = os.path.join(path, '{}.csv'.format(route_prefix))
if os.path.exists(path):
os.remove(path)
results = results.with_session(session).all()
fields = self.get_csv_fields()
csv_file = open(path, 'wt', encoding='utf_8')
writer = csv.DictWriter(csv_file, fields)
writer.writeheader()
def write(obj, i):
writer.writerow(self.get_csv_row(obj, fields))
self.progress_loop(write, results, progress,
message="Collecting data for CSV")
csv_file.close()
session.commit()
except Exception as error:
msg = "generating CSV file for download failed!"
log.warning(msg, exc_info=True)
session.rollback()
session.close()
if progress:
progress.session.load()
progress.session['error'] = True
progress.session['error_msg'] = "{}: {}".format(
msg, simple_error(error))
progress.session.save()
return
finally:
session.close()
if progress:
progress.session.load()
progress.session['complete'] = True
progress.session['success_url'] = self.get_index_url()
progress.session['extra_session_bits'] = {
'{}.results_csv.generated'.format(route_prefix): True,
}
progress.session.save()
def results_csv_download(self):
route_prefix = self.get_route_prefix()
user_uuid = self.request.user.uuid
path = os.path.join(self.rattail_config.datadir(), 'downloads',
'results-csv', route_prefix,
user_uuid[:2], user_uuid[2:],
'{}.csv'.format(route_prefix))
return self.file_response(path)
def results_xlsx(self):
"""
Download current list results as XLSX.
"""
results = self.get_effective_data()
# start thread to actually do work / generate progress data
route_prefix = self.get_route_prefix()
key = '{}.results_xlsx'.format(route_prefix)
progress = self.make_progress(key)
thread = Thread(target=self.results_xlsx_thread,
args=(results, self.request.user.uuid, progress))
thread.start()
# send user to progress page
return self.render_progress(progress, {
'cancel_url': self.get_index_url(),
'cancel_msg': "XLSX download was canceled.",
})
def results_xlsx_session(self):
return self.make_isolated_session()
def results_write_xlsx(self, path, fields, results, session, progress=None):
writer = ExcelWriter(path, fields, sheet_title=self.get_model_title_plural())
writer.write_header()
rows = []
def write(obj, i):
data = self.get_xlsx_row(obj, fields)
row = [data[field] for field in fields]
rows.append(row)
self.progress_loop(write, results, progress,
message="Collecting data for Excel")
def finalize(x, i):
writer.write_rows(rows)
writer.auto_freeze()
writer.auto_filter()
writer.auto_resize()
writer.save()
self.progress_loop(finalize, [1], progress,
message="Writing Excel file to disk")
def results_xlsx_thread(self, results, user_uuid, progress):
"""
Thread target, responsible for actually generating the Excel file which
is to be presented for download.
"""
route_prefix = self.get_route_prefix()
session = self.results_xlsx_session()
try:
# create folder(s) for output; make sure file doesn't exist
path = os.path.join(self.rattail_config.datadir(), 'downloads',
'results-xlsx', route_prefix,
user_uuid[:2], user_uuid[2:])
if not os.path.exists(path):
os.makedirs(path)
path = os.path.join(path, '{}.xlsx'.format(route_prefix))
if os.path.exists(path):
os.remove(path)
results = results.with_session(session).all()
fields = self.get_xlsx_fields()
# write output file
self.results_write_xlsx(path, fields, results, session, progress=progress)
except Exception as error:
msg = "generating XLSX file for download failed!"
log.warning(msg, exc_info=True)
session.rollback()
session.close()
if progress:
progress.session.load()
progress.session['error'] = True
progress.session['error_msg'] = "{}: {}".format(
msg, simple_error(error))
progress.session.save()
return
session.commit()
session.close()
if progress:
progress.session.load()
progress.session['complete'] = True
progress.session['success_url'] = self.get_index_url()
progress.session['extra_session_bits'] = {
'{}.results_xlsx.generated'.format(route_prefix): True,
}
progress.session.save()
def results_xlsx_download(self):
route_prefix = self.get_route_prefix()
user_uuid = self.request.user.uuid
path = os.path.join(self.rattail_config.datadir(), 'downloads',
'results-xlsx', route_prefix,
user_uuid[:2], user_uuid[2:],
'{}.xlsx'.format(route_prefix))
return self.file_response(path)
def get_xlsx_fields(self):
"""
Return the list of fields to be written to XLSX download.
"""
fields = []
mapper = orm.class_mapper(self.model_class)
for prop in mapper.iterate_properties:
if isinstance(prop, orm.ColumnProperty):
fields.append(prop.key)
return fields
def get_xlsx_row(self, obj, fields):
"""
Return a dict for use when writing the row's data to CSV download.
"""
row = {}
for field in fields:
row[field] = getattr(obj, field, None)
return row
def download_results_rows_supported_formats(self):
# TODO: default formats should be configurable?
return OrderedDict([
('xlsx', "Excel (XLSX)"),
('csv', "CSV"),
])
def download_results_rows_default_format(self):
# TODO: default format should be configurable
return 'xlsx'
def download_results_rows(self):
"""
View for saving *rows* of current (filtered) data results into a file,
and downloading that file.
"""
route_prefix = self.get_route_prefix()
user_uuid = self.request.user.uuid
# POST means generate a new results file for download
if self.request.method == 'POST':
# make sure a valid format was requested
supported = self.download_results_rows_supported_formats()
if not supported:
self.request.session.flash("There are no supported download formats!",
'error')
return self.redirect(self.get_index_url())
fmt = self.request.POST.get('fmt')
if not fmt:
fmt = self.download_results_rows_default_format() or list(supported)[0]
if fmt not in supported:
self.request.session.flash("Unsupported download format: {}".format(fmt),
'error')
return self.redirect(self.get_index_url())
# parse field list if one was given
fields = self.request.POST.get('fields')
if fields:
fields = fields.split(',')
if not fields:
if fmt == 'csv':
fields = self.get_row_csv_fields()
elif fmt == 'xlsx':
fields = self.get_row_xlsx_fields()
else:
self.request.session.flash("No fields were specified", 'error')
return self.redirect(self.get_index_url())
# start thread to actually do work / report progress
key = '{}.download_results_rows'.format(route_prefix)
progress = self.make_progress(key)
results = self.get_effective_data()
thread = Thread(target=self.download_results_rows_thread,
args=(results, fmt, fields, user_uuid, progress))
thread.start()
# show user the progress page
return self.render_progress(progress, {
'cancel_url': self.get_index_url(),
'cancel_msg': "Download was canceled.",
})
# not POST, so just download a file (if specified)
filename = self.request.GET.get('filename')
if not filename:
return self.redirect(self.get_index_url())
path = self.download_results_rows_path(user_uuid, filename)
return self.file_response(path)
def download_results_rows_filename(self, fmt):
"""
Must return an appropriate "download results" filename for the given
format. E.g. ``'products.csv'``
"""
route_prefix = self.get_route_prefix()
if fmt == 'csv':
return '{}.rows.csv'.format(route_prefix)
if fmt == 'xlsx':
return '{}.rows.xlsx'.format(route_prefix)
def download_results_rows_path(self, user_uuid, filename=None,
typ='results', makedirs=False):
"""
Returns an absolute path for the "results" data file, specific to the
given user UUID.
"""
route_prefix = self.get_route_prefix()
path = os.path.join(self.rattail_config.datadir(), 'downloads',
typ, route_prefix,
user_uuid[:2], user_uuid[2:])
if makedirs and not os.path.exists(path):
os.makedirs(path)
if filename:
path = os.path.join(path, filename)
return path
def download_results_rows_fields_available(self, **kwargs):
"""
Return the list of fields which are *available* to be written to
download file. Default field list will be constructed from the
underlying table columns.
"""
fields = []
mapper = orm.class_mapper(self.model_class)
for prop in mapper.iterate_properties:
if isinstance(prop, orm.ColumnProperty):
fields.append(prop.key)
return fields
def download_results_rows_fields_default(self, fields, **kwargs):
"""
Return the default list of fields to be written to download file.
Unless you override, all "available" fields will be included by
default.
"""
return fields
def download_results_rows_thread(self, results, fmt, fields, user_uuid, progress):
"""
Thread target, which invokes :meth:`download_results_generate()` to
officially generate the data file which is then to be downloaded.
"""
route_prefix = self.get_route_prefix()
session = self.make_isolated_session()
try:
# create folder(s) for output; make sure file doesn't exist
filename = self.download_results_rows_filename(fmt)
path = self.download_results_rows_path(user_uuid, filename, makedirs=True)
if os.path.exists(path):
os.remove(path)
# generate file for download
results = results.with_session(session).all()
self.download_results_rows_setup(fields, progress=progress)
self.download_results_rows_generate(session, results, path, fmt, fields,
progress=progress)
session.commit()
except Exception as error:
msg = "failed to generate results file for download!"
log.warning(msg, exc_info=True)
session.rollback()
if progress:
progress.session.load()
progress.session['error'] = True
progress.session['error_msg'] = "{}: {}".format(
msg, simple_error(error))
progress.session.save()
return
finally:
session.close()
if progress:
progress.session.load()
progress.session['complete'] = True
progress.session['success_url'] = self.get_index_url()
progress.session['extra_session_bits'] = {
'{}.results_rows.generated'.format(route_prefix): path,
}
progress.session.save()
def download_results_rows_setup(self, fields, progress=None):
"""
Perform any up-front caching or other setup required, just prior to
generating a new results data file for download.
"""
def download_results_rows_generate(self, session, results, path, fmt, fields, progress=None):
"""
This method is responsible for actually generating the data file for a
"download rows for results" operation, according to the given params.
"""
# we really are concerned with "rows of results" here, so let's just
# replace the 'results' list with a list of rows
original_results = results
results = []
def collect(obj, i):
results.extend(self.get_row_data(obj).all())
self.progress_loop(collect, original_results, progress,
message="Collecting data for {}".format(self.get_row_model_title_plural()))
if fmt == 'csv':
csv_file = open(path, 'wt', encoding='utf_8')
writer = csv.DictWriter(csv_file, fields)
writer.writeheader()
def write(obj, i):
data = self.download_results_rows_normalize(obj, fields, fmt=fmt)
csvrow = self.download_results_rows_coerce_csv(data, fields)
writer.writerow(csvrow)
self.progress_loop(write, results, progress,
message="Writing data to CSV file")
csv_file.close()
elif fmt == 'xlsx':
writer = ExcelWriter(path, fields,
sheet_title=self.get_row_model_title_plural())
writer.write_header()
xlrows = []
def write(obj, i):
data = self.download_results_rows_normalize(obj, fields, fmt=fmt)
row = self.download_results_rows_coerce_xlsx(data, fields)
xlrow = [row[field] for field in fields]
xlrows.append(xlrow)
self.progress_loop(write, results, progress,
message="Collecting data for Excel")
def finalize(x, i):
writer.write_rows(xlrows)
writer.auto_freeze()
writer.auto_filter()
writer.auto_resize()
writer.save()
self.progress_loop(finalize, [1], progress,
message="Writing Excel file to disk")
def download_results_rows_normalize(self, row, fields, **kwargs):
"""
Normalize the given row object into a data dict, for use when writing
to the results file for download.
"""
data = {}
for field in fields:
value = getattr(row, field, None)
# make timestamps zone-aware
if isinstance(value, datetime.datetime):
value = localtime(self.rattail_config, value,
from_utc=not self.has_local_times)
data[field] = value
return data
def download_results_rows_coerce_csv(self, data, fields, **kwargs):
"""
Coerce the given data dict record, to a "row" dict suitable for use
when writing directly to CSV file. Each value in the dict should be a
string type.
"""
csvrow = dict(data)
for field in fields:
value = csvrow.get(field)
if value is None:
value = ''
else:
value = str(value)
csvrow[field] = value
return csvrow
def download_results_rows_coerce_xlsx(self, data, fields, **kwargs):
"""
Coerce the given data dict record, to a "row" dict suitable for use
when writing directly to XLSX file.
"""
data = dict(data)
for key in data:
value = data[key]
# convert GPC to pretty string
if isinstance(value, GPC):
value = value.pretty()
# make timestamps local, "zone-naive"
elif isinstance(value, datetime.datetime):
value = localtime(self.rattail_config, value, tzinfo=False)
data[key] = value
return data
def row_results_xlsx(self):
"""
Download current *row* results as XLSX.
"""
obj = self.get_instance()
results = self.get_effective_row_data(sort=True)
fields = self.get_row_xlsx_fields()
path = temp_path(suffix='.xlsx')
writer = ExcelWriter(path, fields, sheet_title=self.get_row_model_title_plural())
writer.write_header()
rows = []
for row_obj in results:
data = self.get_row_xlsx_row(row_obj, fields)
row = [data[field] for field in fields]
rows.append(row)
writer.write_rows(rows)
writer.auto_freeze()
writer.auto_filter()
writer.auto_resize()
writer.save()
response = self.request.response
with open(path, 'rb') as f:
response.body = f.read()
os.remove(path)
response.content_length = len(response.body)
response.content_type = str('application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
filename = self.get_row_results_xlsx_filename(obj)
response.content_disposition = str('attachment; filename={}'.format(filename))
return response
def get_row_xlsx_fields(self):
"""
Return the list of row fields to be written to XLSX download.
"""
# TODO: should this be shared at all? in a better way?
return self.get_row_csv_fields()
def get_row_xlsx_row(self, row, fields):
"""
Return a dict for use when writing the row's data to XLSX download.
"""
xlrow = {}
for field in fields:
value = getattr(row, field, None)
if isinstance(value, GPC):
value = str(value)
elif isinstance(value, datetime.datetime):
# datetime values we provide to Excel must *not* have time zone info,
# but we should make sure they're in "local" time zone effectively.
# note however, this assumes a "naive" time value is in UTC zone!
if value.tzinfo:
value = localtime(self.rattail_config, value, tzinfo=False)
else:
value = localtime(self.rattail_config, value, from_utc=True, tzinfo=False)
xlrow[field] = value
return xlrow
def get_row_results_xlsx_filename(self, obj):
return '{}.xlsx'.format(self.get_row_grid_key())
def row_results_csv(self):
"""
Download current row results data for an object, as CSV
"""
obj = self.get_instance()
fields = self.get_row_csv_fields()
data = io.StringIO()
writer = UnicodeDictWriter(data, fields)
writer.writeheader()
for row in self.get_effective_row_data(sort=True):
writer.writerow(self.get_row_csv_row(row, fields))
response = self.request.response
filename = self.get_row_results_csv_filename(obj)
response.text = data.getvalue()
response.content_type = 'text/csv'
response.content_disposition = 'attachment; filename={}'.format(filename)
data.close()
response.content_length = len(response.body)
return response
def get_row_results_csv_filename(self, instance):
return '{}.csv'.format(self.get_row_grid_key())
def get_csv_fields(self):
"""
Return the list of fields to be written to CSV download. Default field
list will be constructed from the underlying table columns.
"""
fields = []
mapper = orm.class_mapper(self.model_class)
for prop in mapper.iterate_properties:
if isinstance(prop, orm.ColumnProperty):
fields.append(prop.key)
return fields
def get_row_csv_fields(self):
"""
Return the list of row fields to be written to CSV download.
"""
try:
mapper = orm.class_mapper(self.model_row_class)
except:
fields = self.get_row_form_fields()
if not fields:
fields = self.get_row_grid_columns()
else:
fields = []
for prop in mapper.iterate_properties:
if isinstance(prop, orm.ColumnProperty):
fields.append(prop.key)
return fields
def get_csv_row(self, obj, fields):
"""
Return a dict for use when writing the row's data to CSV download.
"""
csvrow = {}
for field in fields:
value = getattr(obj, field, None)
if isinstance(value, datetime.datetime):
# TODO: this assumes value is *always* naive UTC
value = localtime(self.rattail_config, value, from_utc=True)
csvrow[field] = '' if value is None else str(value)
return csvrow
def get_row_csv_row(self, row, fields):
"""
Return a dict for use when writing the row's data to CSV download.
"""
csvrow = {}
for field in fields:
value = getattr(row, field, None)
if isinstance(value, datetime.datetime):
# TODO: this assumes value is *always* naive UTC
value = localtime(self.rattail_config, value, from_utc=True)
csvrow[field] = '' if value is None else str(value)
return csvrow
##############################
# CRUD Stuff
##############################
def get_instance(self):
"""
Fetch the current model instance by inspecting the route kwargs and
doing a database lookup. If the instance cannot be found, raises 404.
"""
model_keys = self.get_model_key(as_tuple=True)
query = self.Session.query(self.get_model_class())
def filtr(query, model_key):
key = self.request.matchdict[model_key]
if self.key_is_integer(model_key):
key = int(key)
query = query.filter(getattr(self.model_class, model_key) == key)
return query
# filter query by composite key. we use filter() instead of a simple
# get() here in case view uses a "pseudo-PK"
for i, model_key in enumerate(model_keys):
query = filtr(query, model_key)
try:
obj = query.one()
except orm.exc.NoResultFound:
raise self.notfound()
# pretend global object doesn't exist, unless access allowed
if self.secure_global_objects:
if not obj.local_only:
if not self.has_perm('view_global'):
raise self.notfound()
return obj
def key_is_integer(self, model_key):
# inspect model class to determine if model_key is numeric
cls = self.get_model_class(error=False)
if cls:
attr = getattr(cls, model_key)
if isinstance(attr.type, sa.Integer):
return True
# do not assume integer by default
return False
def get_instance_title(self, instance):
"""
Return a "pretty" title for the instance, to be used in the page title etc.
"""
return str(instance)
@classmethod
def get_form_factory(cls):
"""
Returns the grid factory or class which is to be used when creating new
grid instances.
"""
return getattr(cls, 'form_factory', forms.Form)
@classmethod
def get_row_form_factory(cls):
"""
Returns the factory or class which is to be used when creating new row
forms.
"""
return getattr(cls, 'row_form_factory', forms.Form)
def download_path(self, obj, filename):
"""
Should return absolute path on disk, for the given object and filename.
Result will be used to return a file response to client.
"""
raise NotImplementedError
def render_downloadable_file(self, obj, field):
if hasattr(obj, field):
filename = getattr(obj, field)
else:
filename = obj[field]
if not filename:
return ""
path = self.download_path(obj, filename)
url = self.get_action_url('download', obj, _query={'filename': filename})
return self.render_file_field(path, url)
def render_file_field(self, path, url=None, filename=None):
"""
Convenience for rendering a file with optional download link
"""
if not filename:
filename = os.path.basename(path)
content = "{} ({})".format(filename, self.readable_size(path))
if url:
return tags.link_to(content, url)
return content
def readable_size(self, path, size=None):
# TODO: this was shamelessly copied from FormAlchemy ...
if size is None:
size = self.get_size(path)
if size == 0:
return '0 KB'
if size <= 1024:
return '1 KB'
if size > 1048576:
return '%0.02f MB' % (size / 1048576.0)
return '%0.02f KB' % (size / 1024.0)
def get_size(self, path):
try:
return os.path.getsize(path)
except os.error:
return 0
def make_form(self, instance=None, factory=None, fields=None, schema=None, make_kwargs=None, configure=None, **kwargs):
"""
Creates a new form for the given model class/instance
"""
if factory is None:
factory = self.get_form_factory()
if fields is None:
fields = self.get_form_fields()
if schema is None:
schema = self.make_form_schema()
if make_kwargs is None:
make_kwargs = self.make_form_kwargs
if configure is None:
configure = self.configure_form
# TODO: SQLAlchemy class instance is assumed *unless* we get a dict
# (seems like we should be smarter about this somehow)
# if not self.creating and not isinstance(instance, dict):
if not self.creating:
kwargs['model_instance'] = instance
kwargs = make_kwargs(**kwargs)
form = factory(fields, schema, **kwargs)
configure(form)
return form
def get_form_fields(self):
if hasattr(self, 'form_fields'):
return self.form_fields
# TODO
# raise NotImplementedError
def make_form_schema(self):
if not self.model_class:
# TODO
raise NotImplementedError
def make_form_kwargs(self, **kwargs):
"""
Return a dictionary of kwargs to be passed to the factory when creating
new form instances.
"""
route_prefix = self.get_route_prefix()
defaults = {
'request': self.request,
'readonly': self.viewing,
'model_class': getattr(self, 'model_class', None),
'action_url': self.request.current_route_url(_query=None),
'assume_local_times': self.has_local_times,
'route_prefix': route_prefix,
'can_edit_help': self.can_edit_help(),
}
if defaults['can_edit_help']:
defaults['edit_help_url'] = self.request.route_url(
'{}.edit_field_help'.format(route_prefix))
if self.creating:
kwargs.setdefault('cancel_url', self.get_index_url())
else:
instance = kwargs['model_instance']
kwargs.setdefault('cancel_url', self.get_action_url('view', instance))
defaults.update(kwargs)
return defaults
def iter_view_supplements(self):
"""
Iterate over all registered supplements for this master view.
"""
supplements = self.request.registry.settings.get('tailbone_view_supplements', [])
route_prefix = self.get_route_prefix()
if supplements and route_prefix in supplements:
for cls in supplements[route_prefix]:
supp = cls(self)
yield supp
def configure_form(self, form):
"""
Configure the main "desktop" form for the view's data model.
"""
self.configure_common_form(form)
self.configure_field_customer_key(form)
self.configure_field_member_key(form)
self.configure_field_product_key(form)
for supp in self.iter_view_supplements():
supp.configure_form(form)
def validate_form(self, form):
if form.validate():
self.form_deserialized = form.validated
return True
return False
def objectify(self, form, data=None):
"""
Create and/or update the model instance from the given form, and return
this object.
.. todo::
This needs a better explanation. And probably tests.
"""
if data is None:
data = form.validated
obj = form.schema.objectify(data, context=form.model_instance)
if self.is_contact:
obj = self.objectify_contact(obj, data)
# force "local only" flag unless global access granted
if self.secure_global_objects:
if not self.has_perm('view_global'):
obj.local_only = True
return obj
def objectify_contact(self, contact, data):
app = self.get_rattail_app()
if 'default_email' in data:
address = data['default_email']
if contact.emails:
if address:
email = contact.emails[0]
email.address = address
else:
contact.emails.pop(0)
elif address:
contact.add_email_address(address)
if 'default_phone' in data:
number = app.format_phone_number(data['default_phone'])
if contact.phones:
if number:
phone = contact.phones[0]
phone.number = number
else:
contact.phones.pop(0)
elif number:
contact.add_phone_number(number)
address_fields = ('address_street',
'address_street2',
'address_city',
'address_state',
'address_zipcode')
addr = dict([(field, data[field])
for field in address_fields
if field in data])
if any(addr.values()):
# we strip 'address_' prefix from fields
addr = dict([(field[8:], value)
for field, value in addr.items()])
if contact.addresses:
address = contact.addresses[0]
for field, value in addr.items():
setattr(address, field, value)
else:
contact.add_mailing_address(**addr)
elif any([field in data for field in address_fields]) and contact.addresses:
contact.addresses.pop()
return contact
def save_form(self, form):
form.save()
def before_create(self, form):
"""
Event hook, called just after the form to create a new instance has
been validated, but prior to the form itself being saved.
"""
def after_create(self, instance):
"""
Event hook, called just after a new instance is saved.
"""
def editable_instance(self, obj):
"""
Returns boolean indicating whether or not the given object
should be considered "editable". Returns ``True`` by default;
override as necessary.
:param obj: A top-level record/object, of the type normally
handled by this master view.
:returns: ``True`` if object is editable, else ``False``.
"""
return True
def after_edit(self, instance):
"""
Event hook, called just after an existing instance is saved.
"""
def deletable_instance(self, instance):
"""
Returns boolean indicating whether or not the given instance can be
considered "deletable". Returns ``True`` by default; override as
necessary.
"""
return True
def before_delete(self, instance):
"""
Event hook, called just before deletion is attempted.
"""
def delete_instance(self, instance):
"""
Delete the instance, or mark it as deleted, or whatever you need to do.
"""
# note, we don't use self.Session here, in case we're being called from
# a separate (bulk-delete) thread
session = orm.object_session(instance)
session.delete(instance)
# Flush immediately to force any pending integrity errors etc.; that
# way we don't set flash message until we know we have success.
session.flush()
def get_after_delete_url(self, instance):
"""
Returns the URL to which the user should be redirected after
successfully "deleting" the given instance.
"""
if hasattr(self, 'after_delete_url'):
if callable(self.after_delete_url):
return self.after_delete_url(instance)
return self.after_delete_url
return self.get_index_url()
##############################
# Autocomplete Stuff
##############################
def autocomplete(self):
"""
View which accepts a single ``term`` param, and returns a list
of autocomplete results to match.
"""
app = self.get_rattail_app()
key = self.get_autocompleter_key()
# url may include key, for more specific autocompleter
if 'key' in self.request.matchdict:
key = '{}.{}'.format(key, self.request.matchdict['key'])
autocompleter = app.get_autocompleter(key)
term = self.request.params.get('term', '')
return autocompleter.autocomplete(self.Session(), term)
def get_autocompleter_key(self):
"""
Must return the "key" to be used when locating the
Autocompleter object, for use with autocomplete view.
"""
if hasattr(self, 'autocompleter_key'):
if self.autocompleter_key:
return self.autocompleter_key
return self.get_route_prefix()
##############################
# Associated Rows Stuff
##############################
def create_row(self):
"""
View for creating a new row record.
"""
self.creating = True
parent = self.get_instance()
index_url = self.get_action_url('view', parent)
form = self.make_row_form(self.model_row_class, cancel_url=index_url)
if self.request.method == 'POST':
if self.validate_row_form(form):
self.before_create_row(form)
obj = self.save_create_row_form(form)
self.after_create_row(obj)
return self.redirect_after_create_row(obj)
return self.render_to_response('create_row', {
'index_url': index_url,
'index_title': '{} {}'.format(
self.get_model_title(),
self.get_instance_title(parent)),
'form': form})
# TODO: still need to verify this logic
def save_create_row_form(self, form):
# self.before_create(form)
# with self.Session().no_autoflush:
# obj = self.objectify(form, self.form_deserialized)
# self.before_create_flush(obj, form)
obj = self.objectify(form, self.form_deserialized)
self.Session.add(obj)
self.Session.flush()
return obj
# def save_create_row_form(self, form):
# self.save_row_form(form)
def before_create_row(self, form):
pass
def after_create_row(self, row_object):
pass
def redirect_after_create_row(self, row, **kwargs):
return self.redirect(self.get_row_action_url('view', row))
def save_quick_row_form(self, form):
raise NotImplementedError("You must define `{}:{}.save_quick_row_form()` "
"in order to process quick row forms".format(
self.__class__.__module__,
self.__class__.__name__))
def redirect_after_quick_row(self, row, **kwargs):
return self.redirect(self.get_row_action_url('edit', row))
def view_row(self):
"""
View for viewing details of a single data row.
"""
self.viewing = True
row = self.get_row_instance()
form = self.make_row_form(row)
parent = self.get_parent(row)
return self.render_to_response('view_row', {
'instance': row,
'instance_title': self.get_instance_title(parent),
'row_title': self.get_row_instance_title(row),
'instance_url': self.get_action_url('view', parent),
'instance_editable': self.row_editable(row),
'instance_deletable': self.row_deletable(row),
'rows_creatable': self.rows_creatable and self.rows_creatable_for(parent),
'model_title': self.get_row_model_title(),
'model_title_plural': self.get_row_model_title_plural(),
'parent_instance': parent,
'parent_model_title': self.get_model_title(),
'action_url': self.get_row_action_url,
'form': form})
def rows_creatable_for(self, instance):
"""
Returns boolean indicating whether or not the given instance should
allow new rows to be added to it.
"""
return True
def rows_quickable_for(self, instance):
"""
Must return boolean indicating whether the "quick row" feature should
be allowed for the given instance. Returns ``True`` by default.
"""
return True
def row_editable(self, row):
"""
Returns boolean indicating whether or not the given row can be
considered "editable". Returns ``True`` by default; override as
necessary.
"""
return True
def edit_row(self):
"""
View for editing an existing model record.
"""
self.editing = True
row = self.get_row_instance()
if not self.row_editable(row):
raise self.redirect(self.get_row_action_url('view', row))
form = self.make_row_form(row)
if self.request.method == 'POST':
if self.validate_row_form(form):
self.save_edit_row_form(form)
return self.redirect_after_edit_row(row)
parent = self.get_parent(row)
return self.render_to_response('edit_row', {
'instance': row,
'row_parent': parent,
'parent_model_title': self.get_model_title(),
'parent_title': self.get_instance_title(parent),
'parent_url': self.get_action_url('view', parent),
'parent_instance': parent,
'instance_title': self.get_row_instance_title(row),
'instance_deletable': self.row_deletable(row),
'form': form,
'dform': form.make_deform_form(),
})
def save_edit_row_form(self, form):
obj = self.objectify(form, self.form_deserialized)
self.after_edit_row(obj)
self.Session.flush()
return obj
# def save_row_form(self, form):
# form.save()
def after_edit_row(self, row):
"""
Event hook, called just after an existing row object is saved.
"""
def redirect_after_edit_row(self, row, **kwargs):
return self.redirect(self.get_row_action_url('view', row))
def row_deletable(self, row):
"""
Returns boolean indicating whether or not the given row can be
considered "deletable". Returns ``True`` by default; override as
necessary.
"""
if not self.rows_deletable:
return False
return True
def delete_row_object(self, row):
"""
Perform the actual deletion of given row object.
"""
self.Session.delete(row)
def delete_row(self):
"""
Desktop view which can "delete" a sub-row from the parent.
"""
row = self.get_row_instance()
if not self.row_deletable(row):
raise self.redirect(self.get_row_action_url('view', row))
self.delete_row_object(row)
return self.redirect(self.get_action_url('view', self.get_parent(row)))
def bulk_delete_rows(self):
"""
Delete all row objects matching the current row grid query.
"""
obj = self.get_instance()
rows = self.get_effective_row_data(sort=False).all()
# TODO: this should use a separate thread with progress
self.delete_row_objects(rows)
self.Session.refresh(obj)
return self.redirect(self.get_action_url('view', obj))
def delete_row_objects(self, rows):
"""
Perform the actual deletion of given row objects.
"""
deleted = 0
for row in rows:
if self.row_deletable(row):
self.delete_row_object(row)
deleted += 1
return deleted
def get_parent(self, row):
raise NotImplementedError
def get_row_instance_title(self, instance):
return self.get_row_model_title()
def get_row_instance(self):
# TODO: is this right..?
# key = self.request.matchdict[self.get_model_key()]
key = self.request.matchdict['row_uuid']
instance = self.Session.get(self.model_row_class, key)
if not instance:
raise self.notfound()
return instance
def make_row_form(self, instance=None, factory=None, fields=None, schema=None, **kwargs):
"""
Creates a new row form for the given model class/instance.
"""
if factory is None:
factory = self.get_row_form_factory()
if fields is None:
fields = self.get_row_form_fields()
if schema is None:
schema = self.make_row_form_schema()
if not self.creating:
kwargs['model_instance'] = instance
kwargs = self.make_row_form_kwargs(**kwargs)
form = factory(fields, schema, **kwargs)
self.configure_row_form(form)
return form
def get_row_form_fields(self):
if hasattr(self, 'row_form_fields'):
return self.row_form_fields
# TODO
# raise NotImplementedError
def make_row_form_schema(self):
if not self.model_row_class:
# TODO
raise NotImplementedError
def make_row_form_kwargs(self, **kwargs):
"""
Return a dictionary of kwargs to be passed to the factory when creating
new row forms.
"""
defaults = {
'request': self.request,
'readonly': self.viewing,
'model_class': getattr(self, 'model_row_class', None),
'action_url': self.request.current_route_url(_query=None),
}
if self.creating:
kwargs.setdefault('cancel_url', self.request.get_referrer())
else:
instance = kwargs['model_instance']
if 'cancel_url' not in kwargs:
kwargs['cancel_url'] = self.get_row_action_url('view', instance)
defaults.update(kwargs)
return defaults
def configure_row_form(self, form):
"""
Configure a row form.
"""
# TODO: is any of this stuff from configure_form() needed?
# if self.editing:
# model_class = self.get_model_class(error=False)
# if model_class:
# mapper = orm.class_mapper(model_class)
# for key in mapper.primary_key:
# for field in form.fields:
# if field == key.name:
# form.set_readonly(field)
# break
# form.remove_field('uuid')
self.set_row_labels(form)
self.configure_field_customer_key(form)
self.configure_field_member_key(form)
self.configure_field_product_key(form)
def validate_row_form(self, form):
if form.validate():
self.form_deserialized = form.validated
return True
return False
def get_customer_key_field(self):
app = self.get_rattail_app()
key = app.get_customer_key_field()
return self.customer_key_fields.get(key, key)
def get_customer_key_label(self):
app = self.get_rattail_app()
field = self.get_customer_key_field()
return app.get_customer_key_label(field=field)
def configure_column_customer_key(self, g):
if '_customer_key_' in g.columns:
field = self.get_customer_key_field()
g.replace('_customer_key_', field)
g.set_label(field, self.get_customer_key_label())
g.set_link(field)
def configure_field_customer_key(self, f):
if '_customer_key_' in f:
field = self.get_customer_key_field()
f.replace('_customer_key_', field)
f.set_label(field, self.get_customer_key_label())
def get_member_key_field(self):
app = self.get_rattail_app()
key = app.get_member_key_field()
return self.member_key_fields.get(key, key)
def get_member_key_label(self):
app = self.get_rattail_app()
field = self.get_member_key_field()
return app.get_member_key_label(field=field)
def configure_column_member_key(self, g):
if '_member_key_' in g.columns:
field = self.get_member_key_field()
g.replace('_member_key_', field)
g.set_label(field, self.get_member_key_label())
g.set_link(field)
def configure_field_member_key(self, f):
if '_member_key_' in f:
field = self.get_member_key_field()
f.replace('_member_key_', field)
f.set_label(field, self.get_member_key_label())
def get_product_key_field(self):
app = self.get_rattail_app()
key = app.get_product_key_field()
return self.product_key_fields.get(key, key)
def get_product_key_label(self):
app = self.get_rattail_app()
field = self.get_product_key_field()
return app.get_product_key_label(field=field)
def configure_column_product_key(self, g):
if '_product_key_' in g.columns:
field = self.get_product_key_field()
g.replace('_product_key_', field)
g.set_label(field, self.get_product_key_label())
g.set_link(field)
if field == 'upc':
g.set_renderer(field, self.render_upc)
def configure_field_product_key(self, f):
if '_product_key_' in f:
field = self.get_product_key_field()
f.replace('_product_key_', field)
f.set_label(field, self.get_product_key_label())
if field == 'upc':
f.set_renderer(field, self.render_upc)
def get_row_action_url(self, action, row, **kwargs):
"""
Generate a URL for the given action on the given row.
"""
route_name = '{}.{}_row'.format(self.get_route_prefix(), action)
return self.request.route_url(route_name, **self.get_row_action_route_kwargs(row))
def get_row_action_route_kwargs(self, row):
"""
Hopefully generic kwarg generator for basic action routes.
"""
# TODO: make this smarter?
parent = self.get_parent(row)
return {
'uuid': parent.uuid,
'row_uuid': row.uuid,
}
def make_diff(self, old_data, new_data, **kwargs):
return diffs.Diff(old_data, new_data, **kwargs)
##############################
# Configuration Views
##############################
def configure(self):
"""
Generic view for configuring some aspect of the software.
"""
if self.request.method == 'POST':
if self.request.POST.get('remove_settings'):
self.configure_remove_settings()
self.request.session.flash("All settings for {} have been "
"removed.".format(self.get_config_title()),
'warning')
return self.redirect(self.request.current_route_url())
else:
data = self.request.POST
# collect any uploaded files
uploads = {}
for key, value in data.items():
if isinstance(value, cgi_FieldStorage):
tempdir = tempfile.mkdtemp()
filename = os.path.basename(value.filename)
filepath = os.path.join(tempdir, filename)
with open(filepath, 'wb') as f:
f.write(value.file.read())
uploads[key] = {
'filedir': tempdir,
'filename': filename,
'filepath': filepath,
}
# process any uploads first
if uploads:
self.configure_process_uploads(uploads, data)
# then gather/save settings
settings = self.configure_gather_settings(data)
self.configure_remove_settings()
self.configure_save_settings(settings)
self.configure_flash_settings_saved()
return self.redirect(self.request.current_route_url())
context = self.configure_get_context()
return self.render_to_response('configure', context)
def template_kwargs_configure(self, **kwargs):
kwargs['system_user'] = getpass.getuser()
return kwargs
def configure_flash_settings_saved(self):
self.request.session.flash("Settings have been saved.")
def configure_process_uploads(self, uploads, data):
if self.has_input_file_templates:
templatesdir = os.path.join(self.rattail_config.datadir(),
'templates', 'input_files',
self.get_route_prefix())
def get_next_filedir(basedir):
nextid = 1
while True:
path = os.path.join(basedir, '{:04d}'.format(nextid))
if not os.path.exists(path):
# this should fail if there happens to be a race
# condition and someone else got to this id first
os.mkdir(path)
return path
nextid += 1
for template in self.normalize_input_file_templates():
key = '{}.upload'.format(template['setting_file'])
if key in uploads:
assert self.request.POST[template['setting_mode']] == 'hosted'
assert not self.request.POST[template['setting_file']]
info = uploads[key]
basedir = os.path.join(templatesdir, template['key'])
if not os.path.exists(basedir):
os.makedirs(basedir)
filedir = get_next_filedir(basedir)
filepath = os.path.join(filedir, info['filename'])
shutil.copyfile(info['filepath'], filepath)
shutil.rmtree(info['filedir'])
numdir = os.path.basename(filedir)
data[template['setting_file']] = os.path.join(numdir,
info['filename'])
def configure_get_simple_settings(self):
"""
If you have some "simple" settings, each of which basically
just needs to be rendered as a separate field, then you can
declare them via this method.
You should return a list of settings; each setting should be
represented as a dict with various pieces of info, e.g.::
{
'section': 'rattail.batch',
'option': 'purchase.allow_cases',
'name': 'rattail.batch.purchase.allow_cases',
'type': bool,
'value': config.getbool('rattail.batch',
'purchase.allow_cases'),
'save_if_empty': False,
}
Note that some of the above is optional, in particular it
works like this:
If you pass ``section`` and ``option`` then you do not need to
pass ``name`` since that can be deduced. Also in this case
you need not pass ``value`` as the normal view logic can fetch
the value automatically. Note that when fetching, it honors
``type`` which, if you do not specify, would be ``str`` by
default.
However if you pass ``name`` then you need not pass
``section`` or ``option``, but you must pass ``value`` since
that cannot be automatically fetched in this case.
:returns: List of simple setting info dicts, as described
above.
"""
def configure_get_name_for_simple_setting(self, simple):
if 'name' in simple:
return simple['name']
return '{}.{}'.format(simple['section'],
simple['option'])
def configure_get_context(self, simple_settings=None,
input_file_templates=True):
"""
Returns the full context dict, for rendering the configure
page template.
Default context will include the "simple" settings, as well as
any "input file template" settings.
You may need to override this method, to add additional
"custom" settings.
:param simple_settings: Optional list of simple settings, if
already initialized.
:returns: Context dict for the page template.
"""
context = {}
if simple_settings is None:
simple_settings = self.configure_get_simple_settings()
if simple_settings:
config = self.rattail_config
settings = {}
for simple in simple_settings:
name = self.configure_get_name_for_simple_setting(simple)
if 'value' in simple:
value = simple['value']
elif simple.get('type') is bool:
value = config.getbool(simple['section'],
simple['option'],
default=simple.get('default', False))
else:
value = config.get(simple['section'],
simple['option'])
settings[name] = value
context['simple_settings'] = settings
# add settings for downloadable input file templates, if any
if input_file_templates and self.has_input_file_templates:
settings = {}
file_options = {}
file_option_dirs = {}
for template in self.normalize_input_file_templates(
include_file_options=True):
settings[template['setting_mode']] = template['mode']
settings[template['setting_file']] = template['file']
settings[template['setting_url']] = template['url']
file_options[template['key']] = template['file_options']
file_option_dirs[template['key']] = template['file_options_dir']
context['input_file_template_settings'] = settings
context['input_file_options'] = file_options
context['input_file_option_dirs'] = file_option_dirs
return context
def configure_gather_settings(self, data, simple_settings=None,
input_file_templates=True):
settings = []
# maybe collect "simple" settings
if simple_settings is None:
simple_settings = self.configure_get_simple_settings()
if simple_settings:
for simple in simple_settings:
name = self.configure_get_name_for_simple_setting(simple)
value = data.get(name)
if simple.get('type') is bool:
value = str(bool(value)).lower()
elif simple.get('type') is int:
value = str(int(value or '0'))
elif value is None:
value = ''
else:
value = str(value)
# only want to save this setting if we received a
# value, or if empty values are okay to save
if value or simple.get('save_if_empty'):
settings.append({'name': name,
'value': value})
# maybe also collect input file template settings
if input_file_templates and self.has_input_file_templates:
for template in self.normalize_input_file_templates():
# mode
settings.append({'name': template['setting_mode'],
'value': data.get(template['setting_mode'])})
# file
value = data.get(template['setting_file'])
if value:
# nb. avoid saving if empty, so can remain "null"
settings.append({'name': template['setting_file'],
'value': value})
# url
settings.append({'name': template['setting_url'],
'value': data.get(template['setting_url'])})
return settings
def configure_remove_settings(self, simple_settings=None,
input_file_templates=True):
app = self.get_rattail_app()
model = self.model
names = []
if simple_settings is None:
simple_settings = self.configure_get_simple_settings()
if simple_settings:
names.extend([self.configure_get_name_for_simple_setting(simple)
for simple in simple_settings])
if input_file_templates and self.has_input_file_templates:
for template in self.normalize_input_file_templates():
names.extend([
template['setting_mode'],
template['setting_file'],
template['setting_url'],
])
if names:
# nb. using thread-local session here; we do not use
# self.Session b/c it may not point to Rattail
session = Session()
for name in names:
app.delete_setting(session, name)
def configure_save_settings(self, settings):
app = self.get_rattail_app()
# nb. using thread-local session here; we do not use
# self.Session b/c it may not point to Rattail
session = Session()
for setting in settings:
app.save_setting(session, setting['name'], setting['value'],
force_create=True)
##############################
# Pyramid View Config
##############################
@classmethod
def defaults(cls, config):
"""
Provide default configuration for a master view.
"""
cls._defaults(config)
@classmethod
def get_instance_url_prefix(cls):
"""
Generate the URL prefix specific to an instance for this model view.
Winds up looking something like:
* ``/products/{uuid}``
* ``/params/{foo}|{bar}|{baz}``
"""
url_prefix = cls.get_url_prefix()
model_keys = cls.get_model_key(as_tuple=True)
prefix = '{}/'.format(url_prefix)
for i, key in enumerate(model_keys):
if i:
prefix += '|'
prefix += '{{{}}}'.format(key)
return prefix
@classmethod
def _defaults(cls, config):
"""
Provide default configuration for a master view.
"""
rattail_config = config.registry.settings.get('rattail_config')
route_prefix = cls.get_route_prefix()
url_prefix = cls.get_url_prefix()
instance_url_prefix = cls.get_instance_url_prefix()
permission_prefix = cls.get_permission_prefix()
model_key = cls.get_model_key()
model_title = cls.get_model_title()
model_title_plural = cls.get_model_title_plural()
config_title = cls.get_config_title()
if cls.has_rows:
row_model_title = cls.get_row_model_title()
row_model_title_plural = cls.get_row_model_title_plural()
config.add_tailbone_permission_group(permission_prefix, model_title_plural, overwrite=False)
# on windows/chrome we are seeing some caching when e.g. user
# applies some filters, then views a record, then clicks back
# button, filters no longer are applied. so by default we
# instruct browser to never cache certain pages which contain
# a grid. at this point only /index and /view
# cf. https://docs.pylonsproject.org/projects/pyramid/en/latest/narr/viewconfig.html#non-predicate-arguments
prevent_cache = rattail_config.getbool('tailbone',
'prevent_cache_for_index_views',
default=True)
# edit help info
cls._defaults_edit_help(config)
# list/search
if cls.listable:
# master views which represent a typical model class, and
# allow for an index view, are registered specially so the
# admin may browse the full list of such views
modclass = cls.get_model_class(error=False)
if modclass:
config.add_tailbone_model_view(modclass.__name__,
model_title_plural,
route_prefix,
permission_prefix)
# but regardless we register the index view, for similar reasons
config.add_tailbone_index_page(route_prefix, model_title_plural,
'{}.list'.format(permission_prefix))
# index view
config.add_tailbone_permission(permission_prefix, '{}.list'.format(permission_prefix),
"List / search {}".format(model_title_plural))
config.add_route(route_prefix, '{}/'.format(url_prefix))
kwargs = {'http_cache': 0} if prevent_cache else {}
config.add_view(cls, attr='index', route_name=route_prefix,
permission='{}.list'.format(permission_prefix),
**kwargs)
# download results
# this is the "new" more flexible approach, but we only want to
# enable it if the class declares it, *and* does *not* declare the
# older style(s). that way each class must explicitly choose
# *only* the new style in order to use it
if cls.results_downloadable and not (
cls.results_downloadable_csv or cls.results_downloadable_xlsx):
config.add_tailbone_permission(permission_prefix, '{}.download_results'.format(permission_prefix),
"Download search results for {}".format(model_title_plural))
config.add_route('{}.download_results'.format(route_prefix), '{}/download-results'.format(url_prefix))
config.add_view(cls, attr='download_results', route_name='{}.download_results'.format(route_prefix),
permission='{}.download_results'.format(permission_prefix))
# download results as CSV (deprecated)
if cls.results_downloadable_csv:
config.add_tailbone_permission(permission_prefix, '{}.results_csv'.format(permission_prefix),
"Download {} as CSV".format(model_title_plural))
config.add_route('{}.results_csv'.format(route_prefix), '{}/csv'.format(url_prefix))
config.add_view(cls, attr='results_csv', route_name='{}.results_csv'.format(route_prefix),
permission='{}.results_csv'.format(permission_prefix))
config.add_route('{}.results_csv_download'.format(route_prefix), '{}/csv/download'.format(url_prefix))
config.add_view(cls, attr='results_csv_download', route_name='{}.results_csv_download'.format(route_prefix),
permission='{}.results_csv'.format(permission_prefix))
# download results as XLSX (deprecated)
if cls.results_downloadable_xlsx:
config.add_tailbone_permission(permission_prefix, '{}.results_xlsx'.format(permission_prefix),
"Download {} as XLSX".format(model_title_plural))
config.add_route('{}.results_xlsx'.format(route_prefix), '{}/xlsx'.format(url_prefix))
config.add_view(cls, attr='results_xlsx', route_name='{}.results_xlsx'.format(route_prefix),
permission='{}.results_xlsx'.format(permission_prefix))
config.add_route('{}.results_xlsx_download'.format(route_prefix), '{}/xlsx/download'.format(url_prefix))
config.add_view(cls, attr='results_xlsx_download', route_name='{}.results_xlsx_download'.format(route_prefix),
permission='{}.results_xlsx'.format(permission_prefix))
# download rows for results
if cls.has_rows and cls.results_rows_downloadable:
config.add_tailbone_permission(permission_prefix, '{}.download_results_rows'.format(permission_prefix),
"Download *rows* for {} search results".format(model_title))
config.add_route('{}.download_results_rows'.format(route_prefix), '{}/download-rows-for-results'.format(url_prefix))
config.add_view(cls, attr='download_results_rows', route_name='{}.download_results_rows'.format(route_prefix),
permission='{}.download_results_rows'.format(permission_prefix))
# configure
if cls.configurable:
config.add_tailbone_permission(permission_prefix,
'{}.configure'.format(permission_prefix),
label="Configure {}".format(config_title))
config.add_route('{}.configure'.format(route_prefix),
cls.get_config_url())
config.add_view(cls, attr='configure',
route_name='{}.configure'.format(route_prefix),
permission='{}.configure'.format(permission_prefix))
config.add_tailbone_config_page('{}.configure'.format(route_prefix),
config_title,
'{}.configure'.format(permission_prefix))
# quickie (search)
if cls.supports_quickie_search:
config.add_tailbone_permission(permission_prefix, '{}.quickie'.format(permission_prefix),
"Do a \"quickie search\" for {}".format(model_title_plural))
config.add_route('{}.quickie'.format(route_prefix), '{}/quickie'.format(route_prefix),
request_method='GET')
config.add_view(cls, attr='quickie', route_name='{}.quickie'.format(route_prefix),
permission='{}.quickie'.format(permission_prefix))
# autocomplete
if cls.supports_autocomplete:
# default
config.add_route('{}.autocomplete'.format(route_prefix),
'{}/autocomplete'.format(url_prefix))
config.add_view(cls, attr='autocomplete',
route_name='{}.autocomplete'.format(route_prefix),
renderer='json',
permission='{}.list'.format(permission_prefix))
# special
config.add_route('{}.autocomplete_special'.format(route_prefix),
'{}/autocomplete/{{key}}'.format(url_prefix))
config.add_view(cls, attr='autocomplete',
route_name='{}.autocomplete_special'.format(route_prefix),
renderer='json',
permission='{}.list'.format(permission_prefix))
# create
if cls.creatable:
config.add_tailbone_permission(permission_prefix, '{}.create'.format(permission_prefix),
"Create new {}".format(model_title))
config.add_route('{}.create'.format(route_prefix), '{}/new'.format(url_prefix))
config.add_view(cls, attr='create', route_name='{}.create'.format(route_prefix),
permission='{}.create'.format(permission_prefix))
# populate new object
if cls.populatable:
config.add_route('{}.populate'.format(route_prefix), '{}/{{uuid}}/populate'.format(url_prefix))
config.add_view(cls, attr='populate', route_name='{}.populate'.format(route_prefix),
permission='{}.create'.format(permission_prefix))
# enable/disable set
if cls.supports_set_enabled_toggle:
config.add_tailbone_permission(permission_prefix, '{}.enable_disable_set'.format(permission_prefix),
"Enable / disable set (selection) of {}".format(model_title_plural))
config.add_route('{}.enable_set'.format(route_prefix), '{}/enable-set'.format(url_prefix),
request_method='POST')
config.add_view(cls, attr='enable_set', route_name='{}.enable_set'.format(route_prefix),
permission='{}.enable_disable_set'.format(permission_prefix))
config.add_route('{}.disable_set'.format(route_prefix), '{}/disable-set'.format(url_prefix),
request_method='POST')
config.add_view(cls, attr='disable_set', route_name='{}.disable_set'.format(route_prefix),
permission='{}.enable_disable_set'.format(permission_prefix))
# delete set
if cls.set_deletable:
config.add_tailbone_permission(permission_prefix, '{}.delete_set'.format(permission_prefix),
"Delete set (selection) of {}".format(model_title_plural))
config.add_route('{}.delete_set'.format(route_prefix), '{}/delete-set'.format(url_prefix),
request_method='POST')
config.add_view(cls, attr='delete_set', route_name='{}.delete_set'.format(route_prefix),
permission='{}.delete_set'.format(permission_prefix))
# bulk delete
if cls.bulk_deletable:
config.add_route('{}.bulk_delete'.format(route_prefix), '{}/bulk-delete'.format(url_prefix),
request_method='POST')
config.add_view(cls, attr='bulk_delete', route_name='{}.bulk_delete'.format(route_prefix),
permission='{}.bulk_delete'.format(permission_prefix))
config.add_tailbone_permission(permission_prefix, '{}.bulk_delete'.format(permission_prefix),
"Bulk delete {}".format(model_title_plural))
# merge
if cls.mergeable:
config.add_route('{}.merge'.format(route_prefix), '{}/merge'.format(url_prefix))
config.add_view(cls, attr='merge', route_name='{}.merge'.format(route_prefix),
permission='{}.merge'.format(permission_prefix))
config.add_tailbone_permission(permission_prefix, '{}.merge'.format(permission_prefix),
"Merge 2 {}".format(model_title_plural))
# download input file template
if cls.has_input_file_templates and cls.creatable:
config.add_route('{}.download_input_file_template'.format(route_prefix),
'{}/download-input-file-template'.format(url_prefix))
config.add_view(cls, attr='download_input_file_template',
route_name='{}.download_input_file_template'.format(route_prefix),
permission='{}.create'.format(permission_prefix))
# view
if cls.viewable:
cls._defaults_view(config)
# image
if cls.has_image:
config.add_route('{}.image'.format(route_prefix), '{}/image'.format(instance_url_prefix))
config.add_view(cls, attr='image', route_name='{}.image'.format(route_prefix),
permission='{}.view'.format(permission_prefix))
# thumbnail
if cls.has_thumbnail:
config.add_route('{}.thumbnail'.format(route_prefix), '{}/thumbnail'.format(instance_url_prefix))
config.add_view(cls, attr='thumbnail', route_name='{}.thumbnail'.format(route_prefix),
permission='{}.view'.format(permission_prefix))
# clone
if cls.cloneable:
config.add_tailbone_permission(permission_prefix, '{}.clone'.format(permission_prefix),
"Clone an existing {0} as a new {0}".format(model_title))
config.add_route('{}.clone'.format(route_prefix), '{}/clone'.format(instance_url_prefix))
config.add_view(cls, attr='clone', route_name='{}.clone'.format(route_prefix),
permission='{}.clone'.format(permission_prefix))
# touch
if cls.touchable:
config.add_tailbone_permission(permission_prefix, '{}.touch'.format(permission_prefix),
"\"Touch\" a {} to trigger datasync for it".format(model_title))
config.add_route('{}.touch'.format(route_prefix),
'{}/touch'.format(instance_url_prefix),
# TODO: should add this restriction after the old
# jquery theme is no longer in use
#request_method='POST'
)
config.add_view(cls, attr='touch', route_name='{}.touch'.format(route_prefix),
permission='{}.touch'.format(permission_prefix))
# download
if cls.downloadable:
config.add_route('{}.download'.format(route_prefix), '{}/download'.format(instance_url_prefix))
config.add_view(cls, attr='download', route_name='{}.download'.format(route_prefix),
permission='{}.download'.format(permission_prefix))
config.add_tailbone_permission(permission_prefix, '{}.download'.format(permission_prefix),
"Download associated data for {}".format(model_title))
# edit
if cls.editable:
config.add_tailbone_permission(permission_prefix, '{}.edit'.format(permission_prefix),
"Edit {}".format(model_title))
config.add_route('{}.edit'.format(route_prefix), '{}/edit'.format(instance_url_prefix))
config.add_view(cls, attr='edit', route_name='{}.edit'.format(route_prefix),
permission='{}.edit'.format(permission_prefix))
# execute
if cls.executable:
config.add_tailbone_permission(permission_prefix, '{}.execute'.format(permission_prefix),
"Execute {}".format(model_title))
config.add_route('{}.execute'.format(route_prefix),
'{}/execute'.format(instance_url_prefix),
request_method='POST')
config.add_view(cls, attr='execute', route_name='{}.execute'.format(route_prefix),
permission='{}.execute'.format(permission_prefix))
# delete
if cls.deletable:
config.add_route('{0}.delete'.format(route_prefix), '{}/delete'.format(instance_url_prefix))
config.add_view(cls, attr='delete', route_name='{0}.delete'.format(route_prefix),
permission='{0}.delete'.format(permission_prefix))
config.add_tailbone_permission(permission_prefix, '{0}.delete'.format(permission_prefix),
"Delete {0}".format(model_title))
# import batch from file
if cls.supports_import_batch_from_file:
config.add_tailbone_permission(permission_prefix, '{}.import_file'.format(permission_prefix),
"Create a new import batch from data file")
### sub-rows stuff follows
# download row results as CSV
if cls.has_rows and cls.rows_downloadable_csv:
config.add_tailbone_permission(permission_prefix, '{}.row_results_csv'.format(permission_prefix),
"Download {} results as CSV".format(row_model_title))
config.add_route('{}.row_results_csv'.format(route_prefix), '{}/{{uuid}}/rows-csv'.format(url_prefix))
config.add_view(cls, attr='row_results_csv', route_name='{}.row_results_csv'.format(route_prefix),
permission='{}.row_results_csv'.format(permission_prefix))
# download row results as Excel
if cls.has_rows and cls.rows_downloadable_xlsx:
config.add_tailbone_permission(permission_prefix, '{}.row_results_xlsx'.format(permission_prefix),
"Download {} results as XLSX".format(row_model_title))
config.add_route('{}.row_results_xlsx'.format(route_prefix), '{}/rows-xlsx'.format(instance_url_prefix))
config.add_view(cls, attr='row_results_xlsx', route_name='{}.row_results_xlsx'.format(route_prefix),
permission='{}.row_results_xlsx'.format(permission_prefix))
# create row
if cls.has_rows:
if cls.rows_creatable:
config.add_tailbone_permission(permission_prefix, '{}.create_row'.format(permission_prefix),
"Create new {} rows".format(model_title))
config.add_route('{}.create_row'.format(route_prefix), '{}/new-row'.format(instance_url_prefix))
config.add_view(cls, attr='create_row', route_name='{}.create_row'.format(route_prefix),
permission='{}.create_row'.format(permission_prefix))
# bulk-delete rows
# nb. must be defined before view_row b/c of url similarity
if cls.rows_bulk_deletable:
config.add_tailbone_permission(permission_prefix,
'{}.delete_rows'.format(permission_prefix),
"Bulk-delete {} from {}".format(
row_model_title_plural, model_title))
config.add_route('{}.delete_rows'.format(route_prefix),
'{}/rows/delete'.format(instance_url_prefix),
# TODO: should enforce this
# request_method='POST'
)
config.add_view(cls, attr='bulk_delete_rows',
route_name='{}.delete_rows'.format(route_prefix),
permission='{}.delete_rows'.format(permission_prefix))
# view row
if cls.has_rows:
if cls.rows_viewable:
config.add_route('{}.view_row'.format(route_prefix),
'{}/rows/{{row_uuid}}'.format(instance_url_prefix))
config.add_view(cls, attr='view_row', route_name='{}.view_row'.format(route_prefix),
permission='{}.view'.format(permission_prefix))
# edit row
if cls.has_rows:
if cls.rows_editable or cls.rows_editable_but_not_directly:
config.add_tailbone_permission(permission_prefix, '{}.edit_row'.format(permission_prefix),
"Edit individual {}".format(row_model_title_plural))
if cls.rows_editable:
config.add_route('{}.edit_row'.format(route_prefix),
'{}/rows/{{row_uuid}}/edit'.format(instance_url_prefix))
config.add_view(cls, attr='edit_row', route_name='{}.edit_row'.format(route_prefix),
permission='{}.edit_row'.format(permission_prefix))
# delete row
if cls.has_rows:
if cls.rows_deletable:
config.add_tailbone_permission(permission_prefix, '{}.delete_row'.format(permission_prefix),
"Delete individual {}".format(row_model_title_plural))
config.add_route('{}.delete_row'.format(route_prefix),
'{}/rows/{{row_uuid}}/delete'.format(instance_url_prefix))
config.add_view(cls, attr='delete_row', route_name='{}.delete_row'.format(route_prefix),
permission='{}.delete_row'.format(permission_prefix))
@classmethod
def _defaults_view(cls, config, **kwargs):
"""
Provide default "view" configuration, i.e. for "viewable" things.
"""
rattail_config = config.registry.settings.get('rattail_config')
route_prefix = cls.get_route_prefix()
url_prefix = cls.get_url_prefix()
instance_url_prefix = cls.get_instance_url_prefix()
model_title = cls.get_model_title()
model_title_plural = cls.get_model_title_plural()
# on windows/chrome we are seeing some caching when e.g. user
# applies some filters, then views a record, then clicks back
# button, filters no longer are applied. so by default we
# instruct browser to never cache certain pages which contain
# a grid. at this point only /index and /view
# cf. https://docs.pylonsproject.org/projects/pyramid/en/latest/narr/viewconfig.html#non-predicate-arguments
prevent_cache = rattail_config.getbool('tailbone',
'prevent_cache_for_index_views',
default=True)
# nb. if caller specifies permission prefix, it's assumed they
# have registered it elsewhere
if 'permission_prefix' in kwargs:
permission_prefix = kwargs['permission_prefix']
else:
permission_prefix = cls.get_permission_prefix()
config.add_tailbone_permission(permission_prefix,
'{}.view'.format(permission_prefix),
"View details for {}".format(model_title))
if cls.has_pk_fields:
config.add_tailbone_permission(permission_prefix,
'{}.view_pk_fields'.format(permission_prefix),
"View all PK-type fields for {}".format(model_title_plural))
if cls.secure_global_objects:
config.add_tailbone_permission(permission_prefix,
'{}.view_global'.format(permission_prefix),
"View *global* {}".format(model_title_plural))
# view by grid index
config.add_route('{}.view_index'.format(route_prefix),
'{}/view'.format(url_prefix))
config.add_view(cls, attr='view_index',
route_name='{}.view_index'.format(route_prefix),
permission='{}.view'.format(permission_prefix))
# view by record key
config.add_route('{}.view'.format(route_prefix),
instance_url_prefix)
kwargs = {'http_cache': 0} if prevent_cache and cls.has_rows else {}
config.add_view(cls, attr='view', route_name='{}.view'.format(route_prefix),
permission='{}.view'.format(permission_prefix),
**kwargs)
# version history
if cls.has_versions and rattail_config and rattail_config.versioning_enabled():
config.add_tailbone_permission(permission_prefix,
'{}.versions'.format(permission_prefix),
"View version history for {}".format(model_title))
config.add_route('{}.versions'.format(route_prefix),
'{}/versions/'.format(instance_url_prefix))
config.add_view(cls, attr='versions',
route_name='{}.versions'.format(route_prefix),
permission='{}.versions'.format(permission_prefix))
config.add_route('{}.version'.format(route_prefix),
'{}/versions/{{txnid}}'.format(instance_url_prefix))
config.add_view(cls, attr='view_version',
route_name='{}.version'.format(route_prefix),
permission='{}.versions'.format(permission_prefix))
@classmethod
def _defaults_edit_help(cls, config, **kwargs):
route_prefix = cls.get_route_prefix()
url_prefix = cls.get_url_prefix()
model_title_plural = cls.get_model_title_plural()
# nb. if caller specifies permission prefix, it's assumed they
# have registered it elsewhere
if 'permission_prefix' in kwargs:
permission_prefix = kwargs['permission_prefix']
else:
permission_prefix = cls.get_permission_prefix()
config.add_tailbone_permission(permission_prefix,
'{}.edit_help'.format(permission_prefix),
"Edit help info for {}".format(model_title_plural))
# edit page help
config.add_route('{}.edit_help'.format(route_prefix),
'{}/edit-help'.format(url_prefix),
request_method='POST')
config.add_view(cls, attr='edit_help',
route_name='{}.edit_help'.format(route_prefix),
renderer='json')
# edit field help
config.add_route('{}.edit_field_help'.format(route_prefix),
'{}/edit-field-help'.format(url_prefix),
request_method='POST')
config.add_view(cls, attr='edit_field_help',
route_name='{}.edit_field_help'.format(route_prefix),
renderer='json')
class ViewSupplement(object):
"""
Base class for view "supplements" - which are sort of like plugins
which can "supplement" certain aspects of the view.
Instead of subclassing a master view and "supplementing" it via
method overrides etc., packages can instead define one or more
``ViewSupplement`` classes. All such supplements are registered
so they can be located; their logic is then merged into the
appropriate master view at runtime.
The primary use case for this is within integration packages, such
as tailbone-corepos and the like. A truly custom app might want
supplemental logic from multiple integration packages, in which
case the "subclassing" approach sort of falls apart.
:attribute:: labels
This can be a dict of extra field labels to be used by the
master view. Same meaning as for
:attr:`tailbone.views.master.MasterView.labels`.
"""
labels = {}
def __init__(self, master):
self.master = master
self.request = master.request
self.model = master.model
self.rattail_config = master.rattail_config
self.Session = master.Session
def get_grid_query(self, query):
"""
Return the "base" query for the grid. This is invoked from
within :meth:`tailbone.views.master.MasterView.query()`.
A typical grid query is
essentially:
.. code-block:: sql
SELECT * FROM mytable
But when a schema extension is in "primary" use, meaning for
instance one of the main grid columns displays extension data,
it may be helpful for the base query to join the extension
table, as opposed to doing a "just in time" join based on
sorting and/or filters:
.. code-block:: sql
SELECT * FROM mytable m
LEFT OUTER JOIN myextension e ON e.uuid = m.uuid
This is accomplished by subjecting the current base query to a
join, e.g. something like::
model = self.model
query = query.outerjoin(model.MyExtension)
return query
"""
return query
def configure_grid(self, g):
"""
Configure the grid as needed, e.g. add columns, and set
renderers etc. for them.
"""
def configure_form(self, f):
"""
Configure the form as needed, e.g. add fields, and set
renderers, default values etc. for them.
"""
def get_xref_buttons(self, obj):
return []
def get_xref_links(self, obj):
return []
def get_version_child_classes(self):
"""
Return a list of additional "version child classes" which are
to be taken into account when displaying version history for a
given record.
See also
:meth:`tailbone.views.master.MasterView.get_version_child_classes()`.
"""
return []
@classmethod
def defaults(cls, config):
cls._defaults(config)
@classmethod
def _defaults(cls, config):
config.add_tailbone_view_supplement(cls.route_prefix, cls)
|