mypy

models/x_data_loader.py

@@ -19,6 +19,7 @@
 ##############################################################################
 import logging
 from enum import Enum
+from typing import Literal
 
 from odoo import _, api, exceptions, models  # type: ignore[import-untyped]
 
@@ -51,7 +52,7 @@
         actor: str,
         tag: str,
         inputs: list[dict[str, int | str]],
-    ):
+    ) -> Literal[True] | list[dict[str, int | str]]:
         """Process the incoming envelopes for the provided actor type/actor/tag.
 
         :param actor_type: consumer/worker
@@ -61,7 +62,6 @@
         :param inputs: envelope id by input name
         :return: outputs, empty dict if actor_type is consumer, envelope id by output
         name if actor_type is worker
-        :rtype: bool | list[dict[str, int|str]]
         """
         _logger.info("Process inputs for %s (actor %s)", tag, actor)
         # Add a key showing the fact this loading is external.
@@ -87,7 +87,7 @@
         raise exceptions.UserError(_("Invalid Actor Type"))
 
     @api.model
-    def process_consumer(self, tag: str, inputs: dict[str, models.Model]):
+    def process_consumer(self, tag: str, inputs: dict[str, models.BaseModel]):
         """Process consumer envelopes.
         Implement to process those envelopes.
 
@@ -98,10 +98,12 @@
         raise UnprocessedEnvelopeException  # pragma: no cover
 
     @api.model
-    def process_worker(self, tag: str, inputs: dict[str, models.Model]):
+    def process_worker(
+        self, tag: str, inputs: dict[str, models.BaseModel]
+    ) -> dict[str, models.BaseModel]:
         """Process worker envelopes.
         Implement to process those envelopes.
 
         :param tag: tag of the actor.
         :param inputs: envelope by input name
         :return: envelope by output name
@@ -102,9 +104,8 @@
         """Process worker envelopes.
         Implement to process those envelopes.
 
         :param tag: tag of the actor.
         :param inputs: envelope by input name
         :return: envelope by output name
-        :rtype: dict[str, models.Model]
         """
         raise UnprocessedEnvelopeException  # pragma: no cover

tests/models.py

@@ -82,7 +82,9 @@
             self.env["res.partner"].create(partners_to_create)
 
     @api.model
-    def process_worker(self, tag: str, inputs: dict[str, models.Model]):
+    def process_worker(
+        self, tag: str, inputs: dict[str, models.BaseModel]
+    ) -> dict[str, models.BaseModel]:
         # This code expects a single creation message. Otherwise, the output would need
         # to contain all the ids
         messages = defaultdict(list)