(PC-11045) Check if there is no exisiting offer with same idAtProvide…

…rs before updating idAtProviders

When running update_offer_and_stock_id_at_providers for venue 2450, there was an error '(UniqueViolation) duplicate key value violates unique constraint offer_idAtProviders_key' because a stock had already been created with the same 'new' idAtProviders. I dont know if we should delete the stock with the old idAtProviders

src/pcapi/core/offers/api.py

@@ -8,7 +8,6 @@
 import pytz
 from sqlalchemy import exc
 from sqlalchemy.orm import joinedload
-from sqlalchemy.sql.functions import func
 import yaml
 from yaml.scanner import ScannerError
 
@@ -564,37 +563,50 @@ def create_mediation(
 def update_offer_and_stock_id_at_providers(venue: Venue, old_siret: str) -> None:
     current_siret = venue.siret
 
-    offer_ids = (
-        Offer.query.filter(Offer.venueId == venue.id)
-        .filter(Offer.idAtProviders.endswith(old_siret))
-        .with_entities(Offer.id)
-        .all()
+    offers = Offer.query.filter(Offer.venueId == venue.id).filter(Offer.idAtProviders.endswith(old_siret)).all()
+    stocks = (
+        Stock.query.join(Offer).filter(Offer.venueId == venue.id).filter(Stock.idAtProviders.endswith(old_siret)).all()
     )
-    stock_ids = (
-        Stock.query.join(Offer)
-        .filter(Offer.venueId == venue.id)
-        .filter(Stock.idAtProviders.endswith(old_siret))
-        .with_entities(Stock.id)
-        .all()
+
+    offer_ids_already_migrated = []
+    offers_to_update = []
+
+    for offer in offers:
+        new_id_at_providers = offer.idAtProviders.replace(old_siret, current_siret)
+        if db.session.query(
+            Offer.query.filter_by(venueId=venue.id, idAtProviders=new_id_at_providers).exists()
+        ).scalar():
+            offer_ids_already_migrated.append(offer.id)
+            continue
+        offer.idAtProviders = new_id_at_providers
+        offers_to_update.append(offer)
+
+    logger.warning(
+        "The following offers are already migrated from old siret to new siret: %s",
+        offer_ids_already_migrated,
+        extra={"venueId": venue.id, "current_siret": venue.siret, "old_siret": old_siret},
     )
 
-    batch_size = 100
+    repository.save(*offers_to_update)
 
-    for offer_index in range(0, len(offer_ids), batch_size):
-        Offer.query.filter(Offer.id.in_(offer_ids[offer_index : offer_index + batch_size])).update(
-            {Offer.idAtProviders: func.replace(Offer.idAtProviders, old_siret, current_siret)},
-            synchronize_session=False,
-        )
-        db.session.commit()
-        offer_index = offer_index + batch_size
+    stock_ids_already_migrated = []
+    stocks_to_update = []
 
-    for stock_index in range(0, len(stock_ids), batch_size):
-        Stock.query.filter(Stock.id.in_(stock_ids[stock_index : stock_index + batch_size])).update(
-            {Stock.idAtProviders: func.replace(Stock.idAtProviders, old_siret, current_siret)},
-            synchronize_session=False,
-        )
-        db.session.commit()
-        stock_index = stock_index + batch_size
+    for stock in stocks:
+        new_id_at_providers = stock.idAtProviders.replace(old_siret, current_siret)
+        if db.session.query(Stock.query.filter_by(idAtProviders=new_id_at_providers).exists()).scalar():
+            stock_ids_already_migrated.append(stock.id)
+            continue
+        stock.idAtProviders = new_id_at_providers
+        stocks_to_update.append(stock)
+
+    logger.warning(
+        "The following stocks are already migrated from old siret to new siret: [%s]",
+        stock_ids_already_migrated,
+        extra={"venueId": venue.id, "current_siret": venue.siret, "old_siret": old_siret},
+    )
+
+    repository.save(*stocks_to_update)
 
 
 def get_expense_domains(offer: Offer) -> list[ExpenseDomain]:

src/pcapi/core/offers/repository.py

@@ -211,7 +211,7 @@ def get_offers_map_by_venue_reference(id_at_providers: list[str], venue_id: int)
     offers_map = {}
     for offer_id, offer_id_at_provider in (
         db.session.query(Offer.id, Offer.idAtProvider)
-        .filter(and_(Offer.venueId == venue_id, Offer.idAtProvider.in_(id_at_providers)))
+        .filter(Offer.venueId == venue_id, Offer.idAtProvider.in_(id_at_providers))
         .all()
     ):
         offers_map[compute_venue_reference(offer_id_at_provider, venue_id)] = offer_id

tests/core/offers/test_api.py

@@ -1238,6 +1238,8 @@ def test_update_offer_and_stock_id_at_providers(self):
         current_siret = "88888888888888"
         venue = VenueFactory(siret=current_siret)
         offer = OfferFactory(venue=venue, idAtProviders="1111111111111@22222222222222")
+        offer_already_migrated = OfferFactory(venue=venue, idAtProviders="1111111111112@22222222222222")
+        OfferFactory(venue=venue, idAtProviders="1111111111112@88888888888888")
         other_venue_offer = OfferFactory(venue=venue, idAtProviders="3333333333333@12222222222222")
         stock = StockFactory(offer=offer, idAtProviders="1111111111111@22222222222222")
 
@@ -1246,6 +1248,7 @@ def test_update_offer_and_stock_id_at_providers(self):
 
         # Then
         assert offer.idAtProviders == "1111111111111@88888888888888"
+        assert offer_already_migrated.idAtProviders == "1111111111112@22222222222222"
         assert stock.idAtProviders == "1111111111111@88888888888888"
         assert other_venue_offer.idAtProviders == "3333333333333@12222222222222"