audrius/reporting-engine
1
0
Fork 1
mirror of https://github.com/OCA/reporting-engine.git synced 2025-02-16 16:30:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

[12.0][MIG] bi_view_editor

Browse Source
This commit is contained in:
Andrea
2019-06-06 13:20:20 +02:00
parent 47e98e495a
commit cd6ba396d5
19 changed files with 791 additions and 636 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
bi_view_editor/models/__init__.py
View File

@@ -2,4 +2,5 @@
from . import models
from . import bve_view
from . import bve_view_line
from . import ir_model

496
bi_view_editor/models/bve_view.py
View File

@@ -1,49 +1,81 @@
# Copyright 2015-2018 Onestein (<http://www.onestein.eu>)
# Copyright 2015-2019 Onestein (<https://www.onestein.eu>)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
import json
from psycopg2.extensions import AsIs
from odoo import api, fields, models, tools
from odoo.exceptions import UserError
from odoo.tools.translate import _
from odoo.addons.base_sparse_field.models.fields import Serialized
from odoo import _, api, fields, models, tools
from odoo.exceptions import UserError, ValidationError
class BveView(models.Model):
_name = 'bve.view'
_description = 'BI View Editor'
@api.depends('group_ids')
@api.multi
@api.depends('group_ids', 'group_ids.users')
def _compute_users(self):
for bve_view in self:
group_ids = bve_view.sudo().group_ids
if group_ids:
bve_view.user_ids = group_ids.mapped('users')
for bve_view in self.sudo():
if bve_view.group_ids:
bve_view.user_ids = bve_view.group_ids.mapped('users')
else:
bve_view.user_ids = self.env['res.users'].sudo().search([])
@api.depends('name')
@api.multi
def _compute_model_name(self):
for bve_view in self:
name = [x for x in bve_view.name.lower() if x.isalnum()]
model_name = ''.join(name).replace('_', '.').replace(' ', '.')
bve_view.model_name = 'x_bve.' + model_name
def _compute_serialized_data(self):
for bve_view in self:
serialized_data = []
for line in bve_view.line_ids.sorted(key=lambda r: r.sequence):
serialized_data_dict = {
'sequence': line.sequence,
'model_id': line.model_id.id,
'id': line.field_id.id,
'name': line.name,
'model_name': line.model_id.name,
'model': line.model_id.model,
'type': line.ttype,
'table_alias': line.table_alias,
'description': line.description,
'row': line.row,
'column': line.column,
'measure': line.measure,
'list': line.in_list,
}
if line.join_node:
serialized_data_dict.update({
'join_node': line.join_node,
'relation': line.relation,
})
serialized_data += [serialized_data_dict]
bve_view.data = json.dumps(serialized_data)
def _inverse_serialized_data(self):
for bve_view in self:
line_ids = self._sync_lines_and_data(bve_view.data)
bve_view.write({'line_ids': line_ids})
name = fields.Char(required=True, copy=False)
model_name = fields.Char(compute='_compute_model_name', store=True)
note = fields.Text(string='Notes')
state = fields.Selection(
[('draft', 'Draft'),
('created', 'Created')],
default='draft',
copy=False)
data = Serialized(
state = fields.Selection([
('draft', 'Draft'),
('created', 'Created')
], default='draft', copy=False)
data = fields.Char(
compute='_compute_serialized_data',
inverse='_inverse_serialized_data',
help="Use the special query builder to define the query "
"to generate your report dataset. "
"NOTE: To be edited, the query should be in 'Draft' status.")
line_ids = fields.One2many(
'bve.view.line',
'bve_view_id',
string='Lines')
action_id = fields.Many2one('ir.actions.act_window', string='Action')
view_id = fields.Many2one('ir.ui.view', string='View')
group_ids = fields.Many2many(
@@ -57,6 +89,7 @@ class BveView(models.Model):
string='Users',
compute='_compute_users',
store=True)
query = fields.Text()
_sql_constraints = [
('name_uniq',
@@ -68,31 +101,20 @@ class BveView(models.Model):
def _create_view_arch(self):
self.ensure_one()
def _get_field_def(name, def_type=''):
if not def_type:
return ''
return """<field name="x_{}" type="{}" />""".format(
name, def_type
)
def _get_field_def(name, def_type):
return """<field name="{}" type="{}" />""".format(name, def_type)
def _get_field_type(field_info):
row = field_info['row'] and 'row'
column = field_info['column'] and 'col'
measure = field_info['measure'] and 'measure'
def _get_field_type(line):
row = line.row and 'row'
column = line.column and 'col'
measure = line.measure and 'measure'
return row or column or measure
def _get_field_list(fields_info):
view_fields = []
for field_info in fields_info:
field_name = field_info['name']
def_type = _get_field_type(field_info)
if def_type:
field_def = _get_field_def(field_name, def_type)
view_fields.append(field_def)
return view_fields
fields_info = json.loads(self.data)
view_fields = _get_field_list(fields_info)
view_fields = []
for line in self.line_ids:
def_type = _get_field_type(line)
if def_type:
view_fields.append(_get_field_def(line.name, def_type))
return view_fields
@api.multi
@@ -100,22 +122,12 @@ class BveView(models.Model):
self.ensure_one()
def _get_field_def(name):
return """<field name="x_{}" />""".format(
name
)
return """<field name="{}" />""".format(name)
def _get_field_list(fields_info):
view_fields = []
for field_info in fields_info:
field_name = field_info['name']
if field_info['list'] and 'join_node' not in field_info:
field_def = _get_field_def(field_name)
view_fields.append(field_def)
return view_fields
fields_info = json.loads(self.data)
view_fields = _get_field_list(fields_info)
view_fields = []
for line in self.line_ids:
if line.in_list and not line.join_node:
view_fields.append(_get_field_def(line.name))
return view_fields
@api.multi
@@ -160,8 +172,7 @@ class BveView(models.Model):
""".format("".join(self._create_view_arch()))
}]
for vals in view_vals:
View.sudo().create(vals)
View.sudo().create(view_vals)
# create Tree view
tree_view = View.sudo().create({
@@ -199,132 +210,105 @@ class BveView(models.Model):
def _build_access_rules(self, model):
self.ensure_one()
def group_ids_with_access(model_name, access_mode):
# pylint: disable=sql-injection
self.env.cr.execute('''SELECT
g.id
FROM
ir_model_access a
JOIN ir_model m ON (a.model_id=m.id)
JOIN res_groups g ON (a.group_id=g.id)
WHERE
m.model=%s AND
a.active = true AND
a.perm_''' + access_mode, (model_name,))
res = self.env.cr.fetchall()
return [x[0] for x in res]
info = json.loads(self.data)
model_names = list(set([f['model'] for f in info]))
read_groups = set.intersection(*[set(
group_ids_with_access(model_name, 'read')
) for model_name in model_names])
if not read_groups and not self.group_ids:
raise UserError(_('Please select at least one group'
' on the security tab.'))
# read access
for group in read_groups:
if not self.group_ids:
self.env['ir.model.access'].sudo().create({
'name': 'read access to ' + self.model_name,
'model_id': model.id,
'group_id': group,
'perm_read': True,
})
else:
# read access only to model
access_vals = []
for group in self.group_ids:
access_vals += [{
'name': 'read access to ' + self.model_name,
'model_id': model.id,
'group_id': group.id,
'perm_read': True
}]
self.env['ir.model.access'].sudo().create(access_vals)
# read and write access
for group in self.group_ids:
self.env['ir.model.access'].sudo().create({
'name': 'read-write access to ' + self.model_name,
'model_id': model.id,
'group_id': group.id,
'perm_read': True,
'perm_write': True,
})
@api.model
@api.multi
def _create_sql_view(self):
self.ensure_one()
def get_fields_info(fields_data):
def get_fields_info(lines):
fields_info = []
for field_data in fields_data:
field = self.env['ir.model.fields'].browse(field_data['id'])
for line in lines:
vals = {
'table': self.env[field.model_id.model]._table,
'table_alias': field_data['table_alias'],
'select_field': field.name,
'as_field': 'x_' + field_data['name'],
'join': False,
'model': field.model_id.model
'table': self.env[line.field_id.model_id.model]._table,
'table_alias': line.table_alias,
'select_field': line.field_id.name,
'as_field': line.name,
'join': line.join_node,
}
if field_data.get('join_node'):
vals.update({'join': field_data['join_node']})
fields_info.append(vals)
return fields_info
def get_join_nodes(info):
join_nodes = [
(f['table_alias'],
f['join'],
f['select_field']) for f in info if f['join'] is not False]
return join_nodes
return [(
f['table_alias'],
f['join'],
f['select_field']
) for f in info if f['join']]
def get_tables(info):
tables = set([(f['table'], f['table_alias']) for f in info])
return tables
return set([(f['table'], f['table_alias']) for f in info])
def get_fields(info):
return [("{}.{}".format(f['table_alias'],
f['select_field']),
f['as_field']) for f in info if 'join_node' not in f]
def get_select_fields(info):
first_field = [(info[0]['table_alias'] + ".id", "id")]
next_fields = [
("{}.{}".format(f['table_alias'], f['select_field']),
f['as_field']) for f in info if 'join_node' not in f
]
return first_field + next_fields
def check_empty_data(data):
if not data or data == '[]':
raise UserError(_('No data to process.'))
if not self.line_ids:
raise UserError(_('No data to process.'))
check_empty_data(self.data)
formatted_data = json.loads(self.data)
info = get_fields_info(formatted_data)
select_fields = get_fields(info)
info = get_fields_info(self.line_ids)
select_fields = get_select_fields(info)
tables = get_tables(info)
join_nodes = get_join_nodes(info)
table_name = self.model_name.replace('.', '_')
view_name = self.model_name.replace('.', '_')
select_str = ', '.join(["{} AS {}".format(f[0], f[1])
for f in select_fields])
from_str = ', '.join(["{} AS {}".format(t[0], t[1])
for t in list(tables)])
where_str = " AND ".join(["{}.{} = {}.id".format(j[0], j[2], j[1])
for j in join_nodes])
# robustness in case something went wrong
# pylint: disable=sql-injection
self._cr.execute('DROP TABLE IF EXISTS "%s"' % table_name)
self._cr.execute('DROP TABLE IF EXISTS %s', (AsIs(view_name), ))
basic_fields = [
("t0.id", "id")
]
# pylint: disable=sql-injection
q = """CREATE or REPLACE VIEW %s as (
self.query = """
SELECT %s
FROM %s
WHERE %s
)""" % (table_name, ','.join(
["{} AS {}".format(f[0], f[1])
for f in basic_fields + select_fields]), ','.join(
["{} AS {}".format(t[0], t[1])
for t in list(tables)]), " AND ".join(
["{}.{} = {}.id".format(j[0], j[2], j[1])
for j in join_nodes] + ["TRUE"]))
self.env.cr.execute(q)
FROM %s
""" % (AsIs(select_str), AsIs(from_str), )
if where_str:
self.query += """
WHERE %s
""" % (AsIs(where_str), )
self.env.cr.execute(
"""CREATE or REPLACE VIEW %s as (
%s
)""", (AsIs(view_name), AsIs(self.query), ))
@api.multi
def action_translations(self):
self.ensure_one()
if self.state != 'created':
return
model = self.env['ir.model'].sudo().search([
('model', '=', self.model_name)
])
translation_obj = self.env['ir.translation'].sudo()
translation_obj.translate_fields('ir.model', model.id)
IrTranslation = self.env['ir.translation'].sudo()
IrTranslation.translate_fields('ir.model', model.id)
for field_id in model.field_id.ids:
translation_obj.translate_fields('ir.model.fields', field_id)
IrTranslation.translate_fields('ir.model.fields', field_id)
return {
'name': 'Translations',
'res_model': 'ir.translation',
@@ -348,53 +332,51 @@ class BveView(models.Model):
def action_create(self):
self.ensure_one()
def _prepare_field(field_data):
if not field_data['custom']:
field = self.env['ir.model.fields'].browse(field_data['id'])
vals = {
'name': 'x_' + field_data['name'],
'complete_name': field.complete_name,
'model': self.model_name,
'relation': field.relation,
'field_description': field_data.get(
'description', field.field_description),
'ttype': field.ttype,
'selection': field.selection,
'size': field.size,
'state': 'manual',
'readonly': True
}
if vals['ttype'] == 'monetary':
vals.update({'ttype': 'float'})
if field.ttype == 'selection' and not field.selection:
model_obj = self.env[field.model_id.model]
selection = model_obj._fields[field.name].selection
if callable(selection):
selection_domain = selection(model_obj)
else:
selection_domain = selection
vals.update({'selection': str(selection_domain)})
return vals
def _prepare_field(line):
field = line.field_id
vals = {
'name': line.name,
'complete_name': field.complete_name,
'model': self.model_name,
'relation': field.relation,
'field_description': line.description,
'ttype': field.ttype,
'selection': field.selection,
'size': field.size,
'state': 'manual',
'readonly': True,
'groups': [(6, 0, field.groups.ids)],
}
if vals['ttype'] == 'monetary':
vals.update({'ttype': 'float'})
if field.ttype == 'selection' and not field.selection:
model_obj = self.env[field.model_id.model]
selection = model_obj._fields[field.name].selection
if callable(selection):
selection_domain = selection(model_obj)
else:
selection_domain = selection
vals.update({'selection': str(selection_domain)})
return vals
# clean dirty view (in case something went wrong)
self.action_reset()
self._check_invalid_lines()
self._check_groups_consistency()
# force removal of dirty views in case something went wrong
self.sudo().action_reset()
# create sql view
self._create_sql_view()
# create model and fields
data = json.loads(self.data)
model_vals = {
fields_data = self.line_ids.filtered(lambda l: not l.join_node)
field_ids = [(0, 0, _prepare_field(f)) for f in fields_data]
model = self.env['ir.model'].sudo().with_context(bve=True).create({
'name': self.name,
'model': self.model_name,
'state': 'manual',
'field_id': [
(0, 0, _prepare_field(field))
for field in data
if 'join_node' not in field]
}
Model = self.env['ir.model'].sudo().with_context(bve=True)
model = Model.create(model_vals)
'field_id': field_ids,
})
# give access rights
self._build_access_rules(model)
@@ -402,9 +384,59 @@ class BveView(models.Model):
# create tree, graph and pivot views
self._create_bve_view()
def _check_groups_consistency(self):
self.ensure_one()
if not self.group_ids:
return
for line_model in self.line_ids.mapped('model_id'):
res_count = self.env['ir.model.access'].sudo().search([
('model_id', '=', line_model.id),
('perm_read', '=', True),
'|',
('group_id', '=', False),
('group_id', 'in', self.group_ids.ids),
], limit=1)
if not res_count:
access_records = self.env['ir.model.access'].sudo().search([
('model_id', '=', line_model.id),
('perm_read', '=', True),
])
group_list = ''
for group in access_records.mapped('group_id'):
group_list += ' * %s\n' % (group.full_name, )
msg_title = _(
'The model "%s" cannot be accessed by users with the '
'selected groups only.' % (line_model.name, ))
msg_details = _(
'At least one of the following groups must be added:')
raise UserError(_(
'%s\n\n%s\n%s' % (msg_title, msg_details, group_list,)
))
def _check_invalid_lines(self):
self.ensure_one()
if any(not line.model_id for line in self.line_ids):
invalid_lines = self.line_ids.filtered(lambda l: not l.model_id)
missing_models = set(invalid_lines.mapped('model_name'))
missing_models = ', '.join(missing_models)
raise UserError(_(
'Following models are missing: %s.\n'
'Probably some modules were uninstalled.' % (missing_models,)
))
if any(not line.field_id for line in self.line_ids):
invalid_lines = self.line_ids.filtered(lambda l: not l.field_id)
missing_fields = set(invalid_lines.mapped('field_name'))
missing_fields = ', '.join(missing_fields)
raise UserError(_(
'Following fields are missing: %s.' % (missing_fields,)
))
@api.multi
def open_view(self):
self.ensure_one()
self._check_invalid_lines()
[action] = self.action_id.read()
action['display_name'] = _('BI View')
return action
@@ -413,7 +445,7 @@ class BveView(models.Model):
def copy(self, default=None):
self.ensure_one()
default = dict(default or {}, name=_("%s (copy)") % self.name)
return super(BveView, self).copy(default=default)
return super().copy(default=default)
@api.multi
def action_reset(self):
@@ -422,23 +454,22 @@ class BveView(models.Model):
has_menus = False
if self.action_id:
action = 'ir.actions.act_window,%d' % (self.action_id.id,)
menus = self.env['ir.ui.menu'].sudo().search([
menus = self.env['ir.ui.menu'].search([
('action', '=', action)
])
has_menus = True if menus else False
menus.unlink()
if self.action_id.view_id:
self.action_id.view_id.sudo().unlink()
self.action_id.sudo().unlink()
self.sudo().action_id.view_id.unlink()
self.sudo().action_id.unlink()
self.env['ir.ui.view'].sudo().search(
[('model', '=', self.model_name)]).unlink()
ir_models = self.env['ir.model'].sudo().search([
('model', '=', self.model_name)
])
for model in ir_models:
model.unlink()
models_to_delete = self.env['ir.model'].sudo().search([
('model', '=', self.model_name)])
if models_to_delete:
models_to_delete.unlink()
table_name = self.model_name.replace('.', '_')
tools.drop_view_if_exists(self.env.cr, table_name)
@@ -450,9 +481,74 @@ class BveView(models.Model):
@api.multi
def unlink(self):
if self.filtered(lambda v: v.state == 'created'):
raise UserError(
_('You cannot delete a created view! '
'Reset the view to draft first.'))
return super().unlink()
@api.model
def _sync_lines_and_data(self, data):
line_ids = [(5, 0, 0)]
fields_info = {}
if data:
fields_info = json.loads(data)
table_model_map = {}
for item in fields_info:
if item.get('join_node', -1) == -1:
table_model_map[item['table_alias']] = item['model_id']
for sequence, field_info in enumerate(fields_info, start=1):
join_model_id = False
join_node = field_info.get('join_node', -1)
if join_node != -1 and table_model_map.get(join_node):
join_model_id = int(table_model_map[join_node])
line_ids += [(0, False, {
'sequence': sequence,
'model_id': field_info['model_id'],
'table_alias': field_info['table_alias'],
'description': field_info['description'],
'field_id': field_info['id'],
'ttype': field_info['type'],
'row': field_info['row'],
'column': field_info['column'],
'measure': field_info['measure'],
'in_list': field_info['list'],
'relation': field_info.get('relation'),
'join_node': field_info.get('join_node'),
'join_model_id': join_model_id,
})]
return line_ids
@api.constrains('line_ids')
def _constraint_line_ids(self):
for view in self:
if view.state == 'created':
raise UserError(
_('You cannot delete a created view! '
'Reset the view to draft first.'))
return super(BveView, self).unlink()
nodes = view.line_ids.filtered(lambda n: n.join_node)
nodes_models = nodes.mapped('table_alias')
nodes_models += nodes.mapped('join_node')
not_nodes = view.line_ids.filtered(lambda n: not n.join_node)
not_nodes_models = not_nodes.mapped('table_alias')
err_msg = _('Inconsistent lines.')
if set(nodes_models) - set(not_nodes_models):
raise ValidationError(err_msg)
if len(set(not_nodes_models) - set(nodes_models)) > 1:
raise ValidationError(err_msg)
@api.model
def get_clean_list(self, data_dict):
serialized_data = json.loads(data_dict)
table_alias_list = set()
for item in serialized_data:
if item.get('join_node', -1) == -1:
table_alias_list.add(item['table_alias'])
for item in serialized_data:
if item.get('join_node', -1) != -1:
if item['table_alias'] not in table_alias_list:
serialized_data.remove(item)
elif item['join_node'] not in table_alias_list:
serialized_data.remove(item)
return json.dumps(serialized_data)

63
bi_view_editor/models/bve_view_line.py Normal file
View File

@@ -0,0 +1,63 @@
# Copyright 2015-2019 Onestein (<https://www.onestein.eu>)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo import _, api, fields, models
from odoo.exceptions import ValidationError
class BveViewLine(models.Model):
_name = 'bve.view.line'
_description = 'BI View Editor Lines'
name = fields.Char(compute='_compute_name')
sequence = fields.Integer(default=1)
bve_view_id = fields.Many2one('bve.view', ondelete='cascade')
model_id = fields.Many2one('ir.model', string='Model')
model_name = fields.Char(compute='_compute_model_name', store=True)
table_alias = fields.Char()
join_model_id = fields.Many2one('ir.model', string='Join Model')
field_id = fields.Many2one('ir.model.fields', string='Field')
field_name = fields.Char(compute='_compute_model_field_name', store=True)
ttype = fields.Char(string='Type')
description = fields.Char(translate=True)
relation = fields.Char()
join_node = fields.Char()
row = fields.Boolean()
column = fields.Boolean()
measure = fields.Boolean()
in_list = fields.Boolean()
@api.constrains('row', 'column', 'measure')
def _constrains_options_check(self):
measure_types = ['float', 'integer', 'monetary']
for line in self:
if line.row or line.column:
if line.join_model_id or line.ttype in measure_types:
err_msg = _('This field cannot be a row or a column.')
raise ValidationError(err_msg)
if line.measure:
if line.join_model_id or line.ttype not in measure_types:
err_msg = _('This field cannot be a measure.')
raise ValidationError(err_msg)
@api.depends('field_id', 'sequence')
def _compute_name(self):
for line in self:
if line.field_id:
field_name = line.field_id.name
line.name = 'x_bve_%s_%s' % (line.sequence, field_name,)
@api.depends('model_id')
def _compute_model_name(self):
for line in self:
if line.model_id:
line.model_name = line.model_id.model
@api.depends('field_id')
def _compute_model_field_name(self):
for line in self:
if line.field_id:
field_name = line.description
model_name = line.model_name
line.field_name = '%s (%s)' % (field_name, model_name, )

213
bi_view_editor/models/ir_model.py
View File

@@ -1,6 +1,8 @@
# Copyright 2015-2018 Onestein (<http://www.onestein.eu>)
# Copyright 2015-2019 Onestein (<https://www.onestein.eu>)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from collections import defaultdict
from odoo import api, models, registry
NO_BI_MODELS = [
@@ -9,14 +11,6 @@ NO_BI_MODELS = [
'fetchmail.server'
]
NO_BI_FIELDS = [
'id',
'create_uid',
'create_date',
'write_uid',
'write_date'
]
NO_BI_TTYPES = [
'many2many',
'one2many',
@@ -73,8 +67,8 @@ class IrModel(models.Model):
return 1
return 0
def _check_unknow(model_name):
if model_name == 'Unknow' or '.' in model_name:
def _check_unknown(model_name):
if model_name == 'Unknown' or '.' in model_name:
return 1
return 0
@@ -84,7 +78,7 @@ class IrModel(models.Model):
count_check += _check_name(model_model)
count_check += _check_startswith(model_model)
count_check += _check_contains(model_model)
count_check += _check_unknow(model_name)
count_check += _check_unknown(model_name)
if not count_check:
return self.env['ir.model.access'].check(
model['model'], 'read', False)
@@ -97,100 +91,72 @@ class IrModel(models.Model):
key=lambda x: x['name'])
return res
@api.model
def _search_fields(self, domain):
Fields = self.env['ir.model.fields']
fields = Fields.sudo().search(domain)
return fields
def get_model_list(self, model_table_map):
if not model_table_map:
return []
domain = [('model_id', 'in', list(model_table_map.keys())),
('store', '=', True),
('ttype', '=', 'many2one')]
fields = self.env['ir.model.fields'].sudo().search(domain)
model_list = []
for field in fields:
for table_alias in model_table_map[field.model_id.id]:
model_list.append(dict(
dict_for_field(field),
table_alias=table_alias,
join_node=-1,
))
return model_list
@api.model
def get_related_fields(self, model_ids):
""" Return list of field dicts for all fields that can be
joined with models in model_ids
"""
def get_model_list(model_ids):
model_list = []
domain = [('model_id', 'in', list(model_ids.values())),
('store', '=', True),
('ttype', 'in', ['many2one'])]
filtered_fields = self._search_fields(domain)
for model in model_ids.items():
for field in filtered_fields:
if model[1] == field.model_id.id:
model_list.append(
dict(dict_for_field(field),
join_node=-1,
table_alias=model[0])
)
return model_list
def get_relation_list(model_ids, model_names):
relation_list = []
domain = [('relation', 'in', list(model_names.values())),
('store', '=', True),
('ttype', 'in', ['many2one'])]
filtered_fields = self._search_fields(domain)
for model in model_ids.items():
for field in filtered_fields:
if model_names[model[1]] == field['relation']:
relation_list.append(
dict(dict_for_field(field),
join_node=model[0],
table_alias=-1)
)
return relation_list
models = self.sudo().browse(model_ids.values())
def get_relation_list(self, model_table_map):
if not model_table_map:
return []
model_names = {}
for model in models:
model_names.update({model.id: model.model})
for model in self.sudo().browse(model_table_map.keys()):
model_names.update({model.model: model.id})
model_list = get_model_list(model_ids)
relation_list = get_relation_list(model_ids, model_names)
return relation_list + model_list
domain = [('relation', 'in', list(model_names.keys())),
('store', '=', True),
('ttype', '=', 'many2one')]
fields = self.env['ir.model.fields'].sudo().search(domain)
relation_list = []
for field in fields:
model_id = model_names[field.relation]
for join_node in model_table_map[model_id]:
relation_list.append(dict(
dict_for_field(field),
join_node=join_node,
table_alias=-1
))
return relation_list
@api.model
def get_related_models(self, model_ids):
def get_related_models(self, model_table_map):
""" Return list of model dicts for all models that can be
joined with the already selected models.
"""
def _get_field(fields, orig, target):
field_list = []
for f in fields:
if f[orig] == -1:
field_list.append(f[target])
return field_list
def _get_list_id(model_ids, fields):
list_model = list(model_ids.values())
list_model += _get_field(fields, 'table_alias', 'model_id')
return list_model
def _get_list_relation(fields):
list_model = _get_field(fields, 'join_node', 'relation')
return list_model
models_list = []
related_fields = self.get_related_fields(model_ids)
list_id = _get_list_id(model_ids, related_fields)
list_model = _get_list_relation(related_fields)
domain = ['|',
('id', 'in', list_id),
('model', 'in', list_model)]
for model in self.sudo().search(domain):
models_list.append(dict_for_model(model))
return self.sort_filter_models(models_list)
domain = [('transient', '=', False)]
if model_table_map:
model_list = self.get_model_list(model_table_map)
relation_list = self.get_relation_list(model_table_map)
model_ids = [f['model_id'] for f in relation_list + model_list]
model_ids += list(model_table_map.keys())
relations = [f['relation'] for f in model_list]
domain += [
'|', ('id', 'in', model_ids), ('model', 'in', relations)]
return self.sudo().search(domain)
@api.model
def get_models(self):
def get_models(self, table_model_map=None):
""" Return list of model dicts for all available models.
"""
self = self.with_context(lang=self.env.user.lang)
model_table_map = defaultdict(list)
for k, v in (table_model_map or {}).items():
model_table_map[v].append(k)
models_list = []
for model in self.search([('transient', '=', False)]):
for model in self.get_related_models(model_table_map):
models_list.append(dict_for_model(model))
return self.sort_filter_models(models_list)
@@ -201,21 +167,23 @@ class IrModel(models.Model):
Return all possible join nodes to add new_field to the query
containing model_ids.
"""
def _get_model_ids(field_data):
model_ids = dict([(field['table_alias'],
field['model_id']) for field in field_data])
return model_ids
def _get_model_table_map(field_data):
table_map = defaultdict(list)
for data in field_data:
table_map[data['model_id']].append(data['table_alias'])
return table_map
def _get_join_nodes_dict(model_ids, new_field):
def _get_join_nodes_dict(model_table_map, new_field):
join_nodes = []
for alias, model_id in model_ids.items():
if model_id == new_field['model_id']:
join_nodes.append({'table_alias': alias})
for field in self.get_related_fields(model_ids):
c = [field['join_node'] == -1, field['table_alias'] == -1]
a = (new_field['model'] == field['relation'])
b = (new_field['model_id'] == field['model_id'])
if (a and c[0]) or (b and c[1]):
for alias in model_table_map[new_field['model_id']]:
join_nodes.append({'table_alias': alias})
for field in self.get_model_list(model_table_map):
if new_field['model'] == field['relation']:
join_nodes.append(field)
for field in self.get_relation_list(model_table_map):
if new_field['model_id'] == field['model_id']:
join_nodes.append(field)
return join_nodes
@@ -229,10 +197,11 @@ class IrModel(models.Model):
nodes_list.append(node)
return nodes_list
model_ids = _get_model_ids(field_data)
self = self.with_context(lang=self.env.user.lang)
model_table_map = _get_model_table_map(field_data)
keys = [(field['table_alias'], field['id'])
for field in field_data if field.get('join_node', -1) != -1]
join_nodes = _get_join_nodes_dict(model_ids, new_field)
join_nodes = _get_join_nodes_dict(model_table_map, new_field)
join_nodes = remove_duplicate_nodes(join_nodes)
return list(filter(
@@ -241,38 +210,34 @@ class IrModel(models.Model):
@api.model
def get_fields(self, model_id):
self = self.with_context(lang=self.env.user.lang)
domain = [
('model_id', '=', model_id),
('store', '=', True),
('name', 'not in', NO_BI_FIELDS),
('name', 'not in', models.MAGIC_COLUMNS),
('ttype', 'not in', NO_BI_TTYPES)
]
fields_dict = []
filtered_fields = self._search_fields(domain)
for field in filtered_fields:
fields_dict.append(
{'id': field.id,
'model_id': model_id,
'name': field.name,
'description': field.field_description,
'type': field.ttype,
'custom': False,
'model': field.model_id.model,
'model_name': field.model_id.name
}
)
sorted_fields = sorted(
for field in self.env['ir.model.fields'].sudo().search(domain):
fields_dict.append({
'id': field.id,
'model_id': model_id,
'name': field.name,
'description': field.field_description,
'type': field.ttype,
'model': field.model,
})
return sorted(
fields_dict,
key=lambda x: x['description'],
reverse=True
)
return sorted_fields
@api.model
def create(self, vals):
if self.env.context and self.env.context.get('bve'):
vals['state'] = 'base'
res = super(IrModel, self).create(vals)
res = super().create(vals)
# this sql update is necessary since a write method here would
# be not working (an orm constraint is restricting the modification

12
bi_view_editor/models/models.py
View File

@@ -1,4 +1,4 @@
# Copyright 2017-2018 Onestein (<http://www.onestein.eu>)
# Copyright 2017-2019 Onestein (<https://www.onestein.eu>)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
import logging
@@ -11,7 +11,7 @@ _logger = logging.getLogger(__name__)
@api.model
def _bi_view(_name):
return _name[0:6] == 'x_bve.'
return _name.startswith('x_bve.')
_auto_init_orig = models.BaseModel._auto_init
@@ -41,23 +41,23 @@ class Base(models.AbstractModel):
@api.model
def _setup_complete(self):
if not _bi_view(self._name):
super(Base, self)._setup_complete()
super()._setup_complete()
else:
self.pool.models[self._name]._log_access = False
@api.model
def _read_group_process_groupby(self, gb, query):
if not _bi_view(self._name):
return super(Base, self)._read_group_process_groupby(gb, query)
return super()._read_group_process_groupby(gb, query)
split = gb.split(':')
if split[0] not in self._fields:
raise UserError(
_('No data to be displayed.'))
return super(Base, self)._read_group_process_groupby(gb, query)
return super()._read_group_process_groupby(gb, query)
@api.model
def _add_magic_fields(self):
if _bi_view(self._name):
self._log_access = False
return super(Base, self)._add_magic_fields()
return super()._add_magic_fields()