[16.0] [REF] shopinvader_api_cart, shopinvader_api_address: Cart and Address hooks

shopinvader_address/models/res_partner.py

@@ -6,7 +6,6 @@
 
 
 class ResPartner(models.Model):
-
     _inherit = "res.partner"
 
     def _ensure_shopinvader_invoicing_address_not_used(self) -> None:
@@ -83,11 +82,9 @@ def _create_shopinvader_invoicing_address(self, vals: dict) -> "ResPartner":
         raise UserError(_("Creation of invoicing addresses is not supported"))
 
     def _update_shopinvader_invoicing_address(
-        self, vals: dict, address_id: int
+        self, vals: dict, address: "ResPartner"
     ) -> "ResPartner":
         self.ensure_one()
-        address = self._get_shopinvader_invoicing_address(address_id)
-
         # if invoicing address is already used, it is not possible to modify it
         # an error will be raised
         address._ensure_shopinvader_invoicing_address_not_used()
@@ -125,20 +122,18 @@ def _create_shopinvader_delivery_address(self, vals: dict) -> "ResPartner":
         return self.env["res.partner"].create(vals)
 
     def _update_shopinvader_delivery_address(
-        self, vals: dict, address_id: int
+        self, vals: dict, address: "ResPartner"
     ) -> "ResPartner":
-
         if any(key in vals for key in ("parent_id", "type")):
             raise UserError(
                 _(
                     "parent_id and type cannot be modified on"
                     " shopinvader delivery address, id: %(address_id)d",
-                    address_id=address_id,
+                    address_id=address.id,
                 )
             )
 
         self.ensure_one()
-        address = self._get_shopinvader_delivery_address(address_id)
 
         # if delivery address is already used, it is not possible to modify it
         address._ensure_shopinvader_delivery_address_not_used()
@@ -147,17 +142,11 @@ def _update_shopinvader_delivery_address(
         address.write(vals)
         return address
 
-    def _delete_shopinvader_delivery_address(self, address_id: int) -> None:
+    def _delete_shopinvader_delivery_address(self, address: "ResPartner") -> None:
         """
         Delete of shopinvader delivery addresses will result to an archive
         """
-        address = self._get_shopinvader_delivery_address(address_id)
-        if address:
-            address._ensure_shopinvader_delivery_address_not_used()
+        address._ensure_shopinvader_delivery_address_not_used()
 
-            # archive address
-            address.active = False
-        else:
-            raise MissingError(
-                _("No address found, id: %(address_id)d", address_id=address_id)
-            )
+        # archive address
+        address.active = False

shopinvader_api_address/routers/address_service.py

@@ -1,9 +1,14 @@
-from typing import Annotated, List
+from typing import Annotated, List, Literal
 
 from fastapi import APIRouter, Depends
 
+from odoo import api, fields, models
+
 from odoo.addons.base.models.res_partner import Partner as ResPartner
-from odoo.addons.fastapi.dependencies import authenticated_partner
+from odoo.addons.fastapi.dependencies import (
+    authenticated_partner,
+    authenticated_partner_env,
+)
 from odoo.addons.shopinvader_schema_address.schemas import (
     DeliveryAddress,
     InvoicingAddress,
@@ -54,14 +59,17 @@ def get_invoicing_address(
 )
 def create_invoicing_address(
     data: InvoicingAddressCreate,
+    env: Annotated[api.Environment, Depends(authenticated_partner_env)],
     partner: Annotated[ResPartner, Depends(authenticated_partner)],
 ) -> InvoicingAddress:
     """
     Create invoicing address
     Raise error since invoicing address is the authenticated partner
     """
-    vals = data.to_res_partner_vals()
-    address = partner._create_shopinvader_invoicing_address(vals)
+    helper = env["shopinvader_api_address.address_router.helper"].new(
+        {"partner": partner}
+    )
+    address = helper._create_address(data, "invoicing")
     return InvoicingAddress.from_res_partner(address)
 
 
@@ -70,20 +78,18 @@ def create_invoicing_address(
 )
 def update_invoicing_address(
     data: InvoicingAddressUpdate,
+    env: Annotated[api.Environment, Depends(authenticated_partner_env)],
     partner: Annotated[ResPartner, Depends(authenticated_partner)],
     address_id: int,
 ) -> InvoicingAddress:
     """
     Update invoicing address of authenticated user
     invoicing address corresponds to authenticated partner
     """
-    vals = data.to_res_partner_vals()
-    # sudo() is needed because some addons override the write
-    # function of res.partner to do some checks before writing.
-    # These checks need more rights than what we are giving to
-    # the enspoint's user
-    # (e.g. snailmail/models/res_partner.py)
-    address = partner.sudo()._update_shopinvader_invoicing_address(vals, address_id)
+    helper = env["shopinvader_api_address.address_router.helper"].new(
+        {"partner": partner}
+    )
+    address = helper._update_address(data, "invoicing", address_id)
     return InvoicingAddress.from_res_partner(address)
 
 
@@ -120,49 +126,122 @@ def get_delivery_address(
 )
 def create_delivery_address(
     data: DeliveryAddressCreate,
+    env: Annotated[api.Environment, Depends(authenticated_partner_env)],
     partner: Annotated[ResPartner, Depends(authenticated_partner)],
 ) -> DeliveryAddress:
     """
     Create delivery address of authenticated user
     """
-    vals = data.to_res_partner_vals()
-    address = partner._create_shopinvader_delivery_address(vals)
-
+    helper = env["shopinvader_api_address.address_router.helper"].new(
+        {"partner": partner}
+    )
+    address = helper._create_address(data, "delivery")
     return DeliveryAddress.from_res_partner(address)
 
 
 @address_router.post("/addresses/delivery/{address_id}", response_model=DeliveryAddress)
 def update_delivery_address(
     data: DeliveryAddressUpdate,
+    env: Annotated[api.Environment, Depends(authenticated_partner_env)],
     partner: Annotated[ResPartner, Depends(authenticated_partner)],
     address_id: int,
 ) -> DeliveryAddress:
     """
     Update delivery address of authenticated user
     """
-    vals = data.to_res_partner_vals()
-    # sudo() is needed because some addons override the write
-    # function of res.partner to do some checks before writing.
-    # These checks need more rights than what we are giving to
-    # the enspoint's user
-    # (e.g. snailmail/models/res_partner.py)
-    address = partner.sudo()._update_shopinvader_delivery_address(vals, address_id)
+    helper = env["shopinvader_api_address.address_router.helper"].new(
+        {"partner": partner}
+    )
+    address = helper._update_address(data, "delivery", address_id)
     return DeliveryAddress.from_res_partner(address)
 
 
 @address_router.delete("/addresses/delivery/{address_id}")
 def delete_delivery_address(
     address_id: int,
+    env: Annotated[api.Environment, Depends(authenticated_partner_env)],
     partner: Annotated[ResPartner, Depends(authenticated_partner)],
 ) -> None:
     """
     Delete delivery address of authenticated user
     Address will be archived.
     """
-
-    # sudo() is needed because some addons override the write
-    # function of res.partner to do some checks before writing.
-    # These checks need more rights than what we are giving to
-    # the enspoint's user
-    # (e.g. snailmail/models/res_partner.py)
-    partner.sudo()._delete_shopinvader_delivery_address(address_id)
+    helper = env["shopinvader_api_address.address_router.helper"].new(
+        {"partner": partner}
+    )
+    helper._delete_address(address_id, "delivery")
+
+
+AddressType = Literal["invoicing", "delivery"]
+
+
+class ShopinvaderApiAddressRouterHelper(models.AbstractModel):
+    _name = "shopinvader_api_address.address_router.helper"
+    _description = "ShopInvader API Address Router Helper"
+
+    partner = fields.Many2one(
+        comodel_name="res.partner",
+    )
+
+    def _get_address(self, address_id: int, address_type: AddressType):
+        self.ensure_one()
+        return getattr(self.partner, f"_get_shopinvader_{address_type}_address")(
+            address_id
+        )
+
+    def _prepare_create_address_vals(
+        self,
+        data: InvoicingAddressCreate | DeliveryAddressCreate,
+        address_type: AddressType,
+    ):
+        return data.to_res_partner_vals()
+
+    def _prepare_update_address_vals(
+        self,
+        data: InvoicingAddressUpdate | DeliveryAddressUpdate,
+        address_type: AddressType,
+        address: ResPartner,
+    ):
+        return data.to_res_partner_vals()
+
+    def _create_address(
+        self,
+        data: InvoicingAddressCreate | DeliveryAddressCreate,
+        address_type: AddressType,
+    ):
+        vals = self._prepare_create_address_vals(data, address_type)
+        return getattr(self.partner, f"_create_shopinvader_{address_type}_address")(
+            vals
+        )
+
+    def _update_address(
+        self,
+        data: InvoicingAddressUpdate | DeliveryAddressUpdate,
+        address_type: AddressType,
+        address_id: int,
+    ):
+        address = self._get_address(address_id, address_type)
+        vals = self._prepare_update_address_vals(data, address_type, address)
+        # sudo() is needed because some addons override the write
+        # function of res.partner to do some checks before writing.
+        # These checks need more rights than what we are giving to
+        # the enspoint's user
+        # (e.g. snailmail/models/res_partner.py)
+        return getattr(
+            self.partner.sudo(), f"_update_shopinvader_{address_type}_address"
+        )(vals, address)
+
+    def _delete_address(
+        self,
+        address_id: int,
+        address_type: AddressType,
+    ):
+        address = self._get_address(address_id, address_type)
+        # sudo() is needed because some addons override the write
+        # function of res.partner to do some checks before writing.
+        # These checks need more rights than what we are giving to
+        # the enspoint's user
+        # (e.g. snailmail/models/res_partner.py)
+        return getattr(
+            self.partner.sudo(), f"_delete_shopinvader_{address_type}_address"
+        )(address)

shopinvader_api_cart/routers/cart.py

@@ -279,9 +279,15 @@ def _sync_cart(
             self._apply_transactions(cart, transactions)
         return cart
 
+    def _prepare_update_cart_vals(self, data: CartUpdateInput, cart: SaleOrder) -> dict:
+        return data._to_sale_order_vals()
+
     def _update(self, partner, data, uuid):
         cart = self.env["sale.order"]._find_open_cart(partner.id, uuid)
         if not cart:
             cart = self.env["sale.order"]._create_empty_cart(partner.id)
-        cart.write(data.convert_to_sale_write(cart))
+
+        vals = self._prepare_update_cart_vals(data, cart)
+        cart.write(vals)
+
         return cart

shopinvader_api_cart/schemas/cart.py

@@ -31,7 +31,7 @@ class CartUpdateInput(StrictExtendableBaseModel, extra="ignore"):
     invoicing: InvoicingUpdateInfo | None = None
     note: str | None = None
 
-    def convert_to_sale_write(self, cart):
+    def _to_sale_order_vals(self):
         vals = {}
         data = self.model_dump(exclude_unset=True)
         if "client_order_ref" in data:

shopinvader_api_cart_step/__init__.py

@@ -1 +1,2 @@
 from . import schemas
+from . import helper

shopinvader_api_cart_step/helper.py

@@ -0,0 +1,18 @@
+from odoo import models
+
+from odoo.addons.sale.models.sale_order import SaleOrder
+
+from .schemas.cart import CartUpdateInput
+
+
+class ShopinvaderApiCartRouterHelper(models.AbstractModel):
+    _inherit = "shopinvader_api_cart.cart_router.helper"
+
+    def _prepare_update_cart_vals(self, data: CartUpdateInput, cart: SaleOrder):
+        vals = super()._prepare_update_cart_vals(data, cart)
+        if data.current_step or data.next_step:
+            step_data = cart._cart_step_update_vals(
+                current_step=data.current_step, next_step=data.next_step
+            )
+            vals.update(step_data)
+        return vals

shopinvader_api_cart_step/schemas/cart.py

@@ -8,15 +8,5 @@
 
 
 class CartUpdateInput(BaseCartUpdateInput, extends=True):
-
     current_step: str | None = None
     next_step: str | None = None
-
-    def convert_to_sale_write(self, cart):
-        vals = super().convert_to_sale_write(cart)
-        if self.current_step or self.next_step:
-            step_data = cart._cart_step_update_vals(
-                current_step=self.current_step, next_step=self.next_step
-            )
-            vals.update(step_data)
-        return vals