xbus_message.py

##############################################################################
#
#    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"])