|
|
# 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 = ["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
@api.model def create(self, vals): if self.env.context and self.env.context.get("bve"): vals["state"] = "base" 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 # of the state field while updating ir.model) q = "UPDATE ir_model SET state = 'manual' WHERE id = %s" self.env.cr.execute(q, (res.id,))
# # update registry if self.env.context.get("bve"): # setup models; this reloads custom models in registry self.pool.setup_models(self._cr)
# signal that registry has changed registry(self.env.cr.dbname).signal_changes()
return res
|
