MPT-16614 [Monitoring] KeyError - Error synchronizing agreement AGR-2… (#706)

…076-2603-9245

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
Closes [MPT-16614](https://softwareone.atlassian.net/browse/MPT-16614)

- Add PriceManager to centralize price retrieval and missing-price
notification logic for agreement and subscription sync flows.
- Refactor agreement and subscription sync to use PriceManager: build
processable lines, fetch prices via PriceManager, skip lines without
prices, and only update unit price when a price is available.
- Add fallback lookup against MPT pricelist for SKUs missing Airtable
prices and trigger centralized notify_missing_prices for unresolved
SKUs.
- Add unit tests for PriceManager and new missing-price scenarios;
update existing sync tests and fixtures to support the new price
retrieval path.
- Migrate Airtable model queries to structured Field-based formulas
(EQ/Field/BLANK/NE/GT/LTE) and switch model access to meta-based
API/base/table/view patterns; update tests accordingly.
- Short-circuit global-customer deployment flow by returning early when
GC agreement deployment creation fails.
- Bump pyairtable dependency to 3.3.* and add a per-file-ignore for a
lint rule in pyproject.toml.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

[MPT-16614]:
https://softwareone.atlassian.net/browse/MPT-16614?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Author: d3rky

adobe_vipm/airtable/models.py

@@ -8,15 +8,14 @@
 from mpt_extension_sdk.runtime.djapp.conf import get_for_product
 from pyairtable.formulas import (
     AND,
-    EQUAL,
-    FIELD,
-    GREATER,
-    LESS_EQUAL,
+    BLANK,
+    EQ,
+    GT,
     LOWER,
-    NOT_EQUAL,
+    LTE,
+    NE,
     OR,
-    STR_VALUE,
-    to_airtable_value,
+    Field,
 )
 from pyairtable.orm import Model, fields
 from requests import HTTPError
@@ -290,7 +289,7 @@ def get_offer_ids_by_membership_id(product_id: str, membership_id: str) -> list[
         list: List of SKUs that belong to the given membership.
     """
     offer_model = get_offer_model(AirTableBaseInfo.for_migrations(product_id))
-    formula = EQUAL(FIELD("membership_id"), STR_VALUE(membership_id))
+    formula = EQ(Field("membership_id"), membership_id)
 
     return [offer.offer_id for offer in offer_model.all(formula=formula)]
 
@@ -320,8 +319,8 @@ def get_transfers_to_process(product_id: str):
     transfer_model = get_transfer_model(AirTableBaseInfo.for_migrations(product_id))
     return transfer_model.all(
         formula=OR(
-            EQUAL(FIELD("status"), STR_VALUE(STATUS_INIT)),
-            EQUAL(FIELD("status"), STR_VALUE(STATUS_RESCHEDULED)),
+            EQ(Field("status"), STATUS_INIT),
+            EQ(Field("status"), STATUS_RESCHEDULED),
         ),
     )
 
@@ -338,7 +337,7 @@ def get_transfers_to_check(product_id: str):
     """
     transfer_model = get_transfer_model(AirTableBaseInfo.for_migrations(product_id))
     return transfer_model.all(
-        formula=EQUAL(FIELD("status"), STR_VALUE(STATUS_RUNNING)),
+        formula=EQ(Field("status"), STATUS_RUNNING),
     )
 
 
@@ -361,15 +360,15 @@ def get_transfer_by_authorization_membership_or_customer(
     transfer_model = get_transfer_model(AirTableBaseInfo.for_migrations(product_id))
     transfers = transfer_model.all(
         formula=AND(
-            EQUAL(FIELD("authorization_uk"), STR_VALUE(authorization_uk)),
+            EQ(Field("authorization_uk"), authorization_uk),
             OR(
-                EQUAL(
-                    LOWER(FIELD("membership_id")),
-                    LOWER(STR_VALUE(membership_or_customer_id)),
+                EQ(
+                    LOWER(Field("membership_id")),
+                    LOWER(membership_or_customer_id),
                 ),
-                EQUAL(FIELD("customer_id"), STR_VALUE(membership_or_customer_id)),
+                EQ(Field("customer_id"), membership_or_customer_id),
             ),
-            NOT_EQUAL(FIELD("status"), STR_VALUE(STATUS_DUPLICATED)),
+            NE(Field("status"), STATUS_DUPLICATED),
         ),
     )
 
@@ -387,9 +386,9 @@ def get_transfer_link(transfer) -> str:
         str: The link to the transfer record or None in case of an error.
     """
     try:
-        base_id = transfer.Meta.base_id
-        table_id = transfer.get_table().id
-        view_id = transfer.get_table().schema().view("Transfer View").id
+        base_id = transfer.meta.base.id
+        table_id = transfer.meta.table.id
+        view_id = transfer.meta.table.schema().view("Transfer View").id
         record_id = transfer.id
     except HTTPError:
         return None
@@ -441,10 +440,10 @@ def _get_prices_for_skus_from_airtable(
     pricelist_model = get_pricelist_model(AirTableBaseInfo.for_pricing(product_id))
     return pricelist_model.all(
         formula=AND(
-            EQUAL(FIELD("currency"), to_airtable_value(currency)),
-            EQUAL(FIELD("valid_until"), "BLANK()"),
+            EQ(Field("currency"), currency),
+            EQ(Field("valid_until"), BLANK()),
             OR(
-                *[EQUAL(FIELD(column_name), to_airtable_value(sku)) for sku in skus],
+                *[EQ(Field(column_name), sku) for sku in skus],
             ),
         ),
     )
@@ -464,6 +463,8 @@ def get_prices_for_skus(product_id: str, currency: str, skus: list[str]) -> dict
     Returns:
         dict: A dictionary with SKU, purchase price items.
     """
+    if not skus:
+        return {}
     items = _get_prices_for_skus_from_airtable(product_id, currency, skus, "sku")
     return {item.sku: item.unit_pp for item in items}
 
@@ -480,6 +481,8 @@ def get_skus_with_available_prices(product_id: str, currency: str, skus: list[st
     Returns:
         set: A set of SKUs if the price is available.
     """
+    if not skus:
+        return set()
     items = _get_prices_for_skus_from_airtable(product_id, currency, skus, "partial_sku")
     return {item.partial_sku for item in items}
 
@@ -496,6 +499,8 @@ def get_skus_with_available_prices_3yc(
         start_date: The date in which the 3YC started.
         skus: List of SKUs which purchase prices must be retrieved.
     """
+    if not skus:
+        return set()
     items = _get_prices_3yc_for_skus_from_airtable(
         product_id, currency, start_date, skus, "partial_sku"
     )
@@ -508,16 +513,16 @@ def _get_prices_3yc_for_skus_from_airtable(
     pricelist_model = get_pricelist_model(AirTableBaseInfo.for_pricing(product_id))
     return pricelist_model.all(
         formula=AND(
-            EQUAL(FIELD("currency"), to_airtable_value(currency)),
+            EQ(Field("currency"), currency),
             OR(
-                EQUAL(FIELD("valid_until"), "BLANK()"),
+                EQ(Field("valid_until"), BLANK()),
                 AND(
-                    LESS_EQUAL(FIELD("valid_from"), STR_VALUE(to_airtable_value(start_date))),
-                    GREATER(FIELD("valid_until"), STR_VALUE(to_airtable_value(start_date))),
+                    LTE(Field("valid_from"), start_date),
+                    GT(Field("valid_until"), start_date),
                 ),
             ),
             OR(
-                *[EQUAL(FIELD(column_name), to_airtable_value(sku)) for sku in skus],
+                *[EQ(Field(column_name), sku) for sku in skus],
             ),
         ),
         sort=["-valid_until"],
@@ -662,10 +667,10 @@ def get_gc_main_agreement(product_id: str, authorization_uk: str, membership_or_
     )
     gc_main_agreements = gc_main_agreement_model.all(
         formula=AND(
-            EQUAL(FIELD("authorization_uk"), STR_VALUE(authorization_uk)),
+            EQ(Field("authorization_uk"), authorization_uk),
             OR(
-                EQUAL(FIELD("membership_id"), STR_VALUE(membership_or_customer_id)),
-                EQUAL(FIELD("customer_id"), STR_VALUE(membership_or_customer_id)),
+                EQ(Field("membership_id"), membership_or_customer_id),
+                EQ(Field("customer_id"), membership_or_customer_id),
             ),
         ),
     )
@@ -691,7 +696,7 @@ def get_gc_agreement_deployments_by_main_agreement(product_id: str, main_agreeme
     )
     return gc_agreement_deployment_model.all(
         formula=AND(
-            EQUAL(FIELD("main_agreement_id"), STR_VALUE(main_agreement_id)),
+            EQ(Field("main_agreement_id"), main_agreement_id),
         ),
     )
 
@@ -714,8 +719,8 @@ def get_gc_agreement_deployments_to_check(product_id: str):
     )
     return gc_agreement_deployment_model.all(
         formula=OR(
-            EQUAL(FIELD("status"), STR_VALUE(STATUS_GC_PENDING)),
-            EQUAL(FIELD("status"), STR_VALUE(STATUS_GC_ERROR)),
+            EQ(Field("status"), STATUS_GC_PENDING),
+            EQ(Field("status"), STATUS_GC_ERROR),
         ),
     )
 
@@ -734,11 +739,10 @@ def get_agreement_deployment_view_link(product_id: str) -> str | None:
         gc_agreement_deployment_model = get_gc_agreement_deployment_model(
             AirTableBaseInfo.for_migrations(product_id)
         )
-        base_id = gc_agreement_deployment_model.Meta.base_id
-        table_id = gc_agreement_deployment_model.get_table().id
+        base_id = gc_agreement_deployment_model.meta.base.id
+        table_id = gc_agreement_deployment_model.meta.table.id
         view_id = (
-            gc_agreement_deployment_model
-            .get_table()
+            gc_agreement_deployment_model.meta.table
             .schema()
             .view(
                 "Agreement Deployments View",
@@ -787,14 +791,11 @@ def from_short_id(cls, vendor_external_id: str):
             Returns:
                 AdobeProductMapping: entity of the AdobeProductMapping.
             """
-            entity = cls.first(
-                formula=EQUAL(FIELD("vendor_external_id"), STR_VALUE(vendor_external_id))
-            )
+            entity = cls.first(formula=EQ(Field("vendor_external_id"), vendor_external_id))
             if entity is None:
                 raise AdobeProductNotFoundError(
                     f"AdobeProduct with vendor_external_id `{vendor_external_id}` not found."
                 )
-
             return entity
 
         def is_consumable(self) -> bool:

adobe_vipm/flows/global_customer.py

@@ -642,6 +642,7 @@ def process_agreement_deployment(  # noqa: C901
             listing,
             licensee,
         )
+
         if not gc_agreement_id:
             return
 

adobe_vipm/flows/sync/agreement.py

@@ -30,6 +30,7 @@
 from adobe_vipm.flows.mpt import get_agreements_by_3yc_commitment_request_invitation
 from adobe_vipm.flows.sync.asset import AssetSyncer
 from adobe_vipm.flows.sync.helper import check_adobe_subscription_id
+from adobe_vipm.flows.sync.price_manager import PriceManager
 from adobe_vipm.flows.sync.subscription import SubscriptionSyncer
 from adobe_vipm.flows.utils import notify_agreement_unhandled_exception_in_teams
 from adobe_vipm.flows.utils.market_segment import get_market_segment
@@ -520,18 +521,21 @@ def _update_3yc_ordering_params(self, commitment_info: dict, parameters: dict):
     def _update_agreement_line_prices(
         self, agreement: dict, currency: str, product_id: str
     ) -> None:
-        agreement_lines = []
-        for line in agreement["lines"]:
-            if line["item"]["externalIds"]["vendor"] != "adobe-reseller-transfer":
-                actual_sku = models.get_adobe_sku(line["item"]["externalIds"]["vendor"])
-                agreement_lines.append((
-                    line,
-                    flows_utils.get_sku_with_discount_level(actual_sku, self._adobe_customer),
-                ))
+        agreement_lines = self._get_processable_agreement_lines(agreement)
+        price_manager = PriceManager(
+            self._mpt_client,
+            self._adobe_customer,
+            agreement_lines,
+            self.agreement_id,
+            self._agreement["listing"]["priceList"]["id"],
+        )
+        skus = [sku for _, sku in agreement_lines]
+        prices = price_manager.get_sku_prices_for_agreement_lines(skus, product_id, currency)
 
-        skus = [item[1] for item in agreement_lines]
-        prices = models.get_sku_price(self._adobe_customer, skus, product_id, currency)
         for line, actual_sku in agreement_lines:
+            if actual_sku not in prices:
+                continue
+
             current_price = line["price"]["unitPP"]
             line["price"]["unitPP"] = prices[actual_sku]
             logger.info(
@@ -542,6 +546,14 @@ def _update_agreement_line_prices(
                 prices[actual_sku],
             )
 
+    def _get_processable_agreement_lines(self, agreement: dict) -> list[tuple[dict, str]]:
+        agreement_lines = []
+        for line in agreement["lines"]:
+            if line["item"]["externalIds"]["vendor"] != "adobe-reseller-transfer":
+                actual_sku = models.get_adobe_sku(line["item"]["externalIds"]["vendor"])
+                agreement_lines.append((line, actual_sku))
+        return agreement_lines
+
     # REFACTOR: get method must not update subscriptions in mpt or terminate a subscription
     def _get_subscriptions_for_update(self, agreement: dict) -> list[tuple[dict, dict, str]]:  # noqa: C901
         logger.info("Getting subscriptions for update for agreement %s", agreement["id"])

adobe_vipm/flows/sync/price_manager.py

@@ -0,0 +1,61 @@
+from mpt_extension_sdk.mpt_http import mpt
+
+from adobe_vipm.airtable import models
+from adobe_vipm.flows import utils as flows_utils
+from adobe_vipm.utils import get_commitment_start_date
+
+
+class PriceManager:
+    """
+    PriceManager class to manage the prices to update the agreement or subscription.
+
+    Attributes:
+        mpt_client: The MPT client.
+        adobe_customer: The Adobe customer.
+        agreement: The agreement data.
+        agreement_lines: The agreement lines.
+    """
+
+    def __init__(self, mpt_client, adobe_customer, lines, agreement_id, pricelist_id):
+        self._mpt_client = mpt_client
+        self._adobe_customer = adobe_customer
+        self._lines = lines
+        self._agreement_id = agreement_id
+        self._pricelist_id = pricelist_id
+
+    def get_sku_prices_for_agreement_lines(self, skus, product_id, currency):
+        """Get the prices for the given SKUs.
+
+        Args:
+            skus: The SKUs to get the prices for.
+            product_id: The product ID.
+            currency: The currency.
+
+        Returns:
+            A dictionary with the list of SKUs and the prices.
+        """
+        prices = models.get_sku_price(self._adobe_customer, skus, product_id, currency)
+        missing_prices_skus = []
+        for line, actual_sku in self._lines:
+            if actual_sku not in prices:
+                missing_prices_skus.append(actual_sku)
+                mpt_price = mpt.get_item_prices_by_pricelist_id(
+                    self._mpt_client,
+                    self._pricelist_id,
+                    line["item"]["id"],
+                )
+                if mpt_price:
+                    prices[actual_sku] = mpt_price[0]["unitPP"]
+
+        self._notify_missing_prices(missing_prices_skus, product_id, currency)
+        return prices
+
+    def _notify_missing_prices(self, missing_prices_skus, product_id, currency):
+        if missing_prices_skus:
+            flows_utils.notify_missing_prices(
+                self._agreement_id,
+                missing_prices_skus,
+                product_id,
+                currency,
+                get_commitment_start_date(self._adobe_customer),
+            )