############################################################################## # # Xbus Common, for Odoo # Copyright © 2023, 2024 XCG Consulting <https://xcg-consulting.fr> # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # ############################################################################## import base64 import json import pprint from odoo import api, fields, models, tools # type: ignore[import-untyped] class XbusMessage(models.Model): """Represents an Xbus message.""" _name = "xbus.message" _description = "Xbus Message" # No creation or automatic column for this table, it is used with an external system _auto = False _log_access = False _order = "envelope_id DESC, id DESC" _rec_name = "uid" envelope_id = fields.Many2one( string="Envelope", comodel_name="xbus.envelope", required=True ) # When setting the uid, the checksum value needs to be computed too. # As checksum is a bigint, there is no field type matching it so it is not done # there, instead it is expected the UID will never be written and that the field # is only used to display information uid = fields.Char(string="UID", readonly=True) type = fields.Selection(selection="_get_type_selection", required=True) header = fields.Binary(attachment=False) body = fields.Binary(attachment=False, required=True) header_json = fields.Text(compute="_compute_header_json", store=False) body_json = fields.Text(compute="_compute_body_json", store=False) @api.depends("header") def _compute_header_json(self): for message in self: message.header_json = message._convert_binary_to_json( message.with_context(bin_size=False).header ) @api.depends("body") def _compute_body_json(self): for message in self: message.body_json = message._convert_binary_to_json( message.with_context(bin_size=False).body ) def _convert_binary_to_json(self, binary_value): if binary_value: if isinstance( binary_value, bytes ): # Check if it is of type bytes before decoding string_value = base64.b64decode( binary_value ).decode() # utf-8 by default else: string_value = binary_value if string_value.startswith("{") or string_value.startswith("["): try: json_dict = json.loads(string_value) return pprint.pformat(json_dict) except json.JSONDecodeError: return False return False def _get_type_selection(self) -> list[tuple[str, str]]: """Provides a list of message type selection values. When adding elements, remember to clean up the message table when your module is uninstalled. """ return [] link_ids = fields.One2many( comodel_name="xbus.message.link", string="Linked Records", inverse_name="message_id", ) def init(self): super().init() self.env.cr.execute( """CREATE TABLE IF NOT EXISTS xbus_message ( id serial primary key, envelope_id integer not null REFERENCES xbus_envelope(id), uid uuid null, type character varying not null, header bytea null, body bytea not null, checksum bigint not null default 0 )""" ) tools.create_index( self._cr, "xbus_message_envelope_id_index", self._table, ["envelope_id"] ) tools.create_index(self._cr, "xbus_message_uid_index", self._table, ["uid"]) tools.create_index(self._cr, "xbus_message_type_index", self._table, ["type"])