reporting-engine/bi_view_editor/models/ir_model.py

218 lines
7.3 KiB
Python
Raw Normal View History

2021-04-28 15:43:56 +00:00
# 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 flectra import api, models
NO_BI_MODELS = ["fetchmail.server"]
NO_BI_TTYPES = ["many2many", "one2many", "html", "binary", "reference"]
def dict_for_field(field):
return {
"id": field.id,
"name": field.name,
"description": field.field_description,
"type": field.ttype,
"relation": field.relation,
"custom": False,
"model_id": field.model_id.id,
"model": field.model_id.model,
"model_name": field.model_id.name,
}
def dict_for_model(model):
return {"id": model.id, "name": model.name, "model": model.model}
class IrModel(models.Model):
_inherit = "ir.model"
@api.model
def _filter_bi_models(self, model):
def _check_name(model_model):
if model_model in NO_BI_MODELS:
return 1
return 0
def _check_startswith(model_model):
if (
model_model.startswith("workflow")
or model_model.startswith("ir.")
or model_model.startswith("base_")
):
return 1
return 0
def _check_contains(model_model):
if (
"mail" in model_model
or "report" in model_model
or "edi." in model_model
):
return 1
return 0
def _check_unknown(model_name):
if model_name == "Unknown" or "." in model_name:
return 1
return 0
model_model = model["model"]
model_name = model["name"]
count_check = 0
count_check += _check_name(model_model)
count_check += _check_startswith(model_model)
count_check += _check_contains(model_model)
count_check += _check_unknown(model_name)
if not count_check:
return self.env["ir.model.access"].check(model["model"], "read", False)
return False
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
def get_relation_list(self, model_table_map):
if not model_table_map:
return []
model_names = {}
for model in self.sudo().browse(model_table_map.keys()):
model_names.update({model.model: model.id})
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_domain(self, model_table_map):
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 domain
@api.model
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.
"""
domain = self._get_related_models_domain(model_table_map)
return self.sudo().search(domain, order="name asc")
@api.model
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 = self.get_related_models(model_table_map)
# filter out abstract models (they do not have DB tables)
non_abstract_models = self.env.registry.models.keys()
models = models.filtered(lambda m: m.model in non_abstract_models)
return list(map(dict_for_model, models))
@api.model
def get_join_nodes(self, field_data, new_field):
"""Return list of field dicts of join nodes
Return all possible join nodes to add new_field to the query
containing model_ids.
"""
def remove_duplicate_nodes(join_nodes):
seen = set()
nodes_list = []
for node in join_nodes:
node_tuple = tuple(node.items())
if node_tuple not in seen:
seen.add(node_tuple)
nodes_list.append(node)
return nodes_list
self = self.with_context(lang=self.env.user.lang)
keys = []
model_table_map = defaultdict(list)
for field in field_data:
model_table_map[field["model_id"]].append(field["table_alias"])
if field.get("join_node", -1) != -1:
keys.append((field["table_alias"], field["id"]))
# nodes in current model
existing_aliases = model_table_map[new_field["model_id"]]
join_nodes = [{"table_alias": alias} for alias in existing_aliases]
# nodes in past selected models
for field in self.get_model_list(model_table_map):
if new_field["model"] == field["relation"]:
if (field["table_alias"], field["id"]) not in keys:
join_nodes.append(field)
# nodes in new model
for field in self.get_relation_list(model_table_map):
if new_field["model_id"] == field["model_id"]:
if (field["table_alias"], field["id"]) not in keys:
join_nodes.append(field)
return remove_duplicate_nodes(join_nodes)
@api.model
def get_fields(self, model_id):
self = self.with_context(lang=self.env.user.lang)
fields = (
self.env["ir.model.fields"]
.sudo()
.search(
[
("model_id", "=", model_id),
("store", "=", True),
("name", "not in", models.MAGIC_COLUMNS),
("ttype", "not in", NO_BI_TTYPES),
],
order="field_description desc",
)
)
fields_dict = list(map(dict_for_field, fields))
return fields_dict