Skip to content

Commit 66fb1ed

Browse files
kolokkolok
authored
Ajouter un index lors de la résolution de mapping (#1532)
1 parent b9879c2 commit 66fb1ed

File tree

4 files changed

+147
-47
lines changed

4 files changed

+147
-47
lines changed

‎dags/sources/tasks/business_logic/keep_acteur_changed.py‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,18 @@ def retrieve_identifiant_unique_from_existing_acteur(
9191
axis=1,
9292
)
9393

94+
# find the duplicated identifiant in df_acteur_from_db and raise if any because
95+
# we can't resolve simply the mapping between source and db
96+
duplicates = df_acteur_from_db[
97+
df_acteur_from_db.duplicated("identifiant", keep=False)
98+
]
99+
if not duplicates.empty:
100+
logger.warning(
101+
"Duplicated identifiant in df_acteur_from_db"
102+
f" {duplicates["identifiant"].tolist()}"
103+
)
104+
raise ValueError("Duplicated identifiant in df_acteur_from_db")
105+
94106
# Replace identifiant_unique (from source) by identifiant (from db) for acteur
95107
# which doesn't have corelation between source, external_id and identifiant_unique
96108
df_normalized.set_index("identifiant", inplace=True)

‎dags_unit_tests/sources/tasks/business_logic/test_keep_acteur_changed.py‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -991,3 +991,72 @@ def test_keep_acteur_changed_same_acteur_but_different_identifiant_unique(dag_co
991991
pd.testing.assert_frame_equal(df_acteur, df_expected, check_dtype=False)
992992
pd.testing.assert_frame_equal(df_acteur_from_db, df_expected, check_dtype=False)
993993
assert metadata == {}
994+
995+
996+
def test_keep_acteur_changed_same_acteur_but_different_identifiant_unique2(dag_config):
997+
df_normalized = pd.DataFrame(
998+
{
999+
"nom": ["nom 1", "nom 2"],
1000+
"identifiant_unique": ["source1_id1", "source1_id2"],
1001+
"source_code": ["source1", "source1"],
1002+
"identifiant_externe": ["id1", "id2"],
1003+
"label_codes": [[], []],
1004+
"acteur_type_code": [[], []],
1005+
"acteur_service_codes": [[], []],
1006+
"proposition_service_codes": [[], []],
1007+
}
1008+
)
1009+
df_acteur_from_db = pd.DataFrame(
1010+
{
1011+
"nom": ["nom 1", "nom 3"],
1012+
"identifiant_unique": ["source1_id_old", "source1_id3"],
1013+
"source_code": ["source1", "source1"],
1014+
"identifiant_externe": ["id1", "id3"],
1015+
"label_codes": [[], []],
1016+
"acteur_type_code": [[], []],
1017+
"acteur_service_codes": [[], []],
1018+
"proposition_service_codes": [[], []],
1019+
}
1020+
)
1021+
df_expected = pd.DataFrame(
1022+
{
1023+
"nom": ["nom 2"],
1024+
"identifiant_unique": ["source1_id2"],
1025+
"source_code": ["source1"],
1026+
"identifiant_externe": ["id2"],
1027+
"label_codes": [[]],
1028+
"acteur_type_code": [[]],
1029+
"acteur_service_codes": [[]],
1030+
"proposition_service_codes": [[]],
1031+
}
1032+
)
1033+
df_expected_from_db = pd.DataFrame(
1034+
{
1035+
"nom": ["nom 3"],
1036+
"identifiant_unique": ["source1_id3"],
1037+
"source_code": ["source1"],
1038+
"identifiant_externe": ["id3"],
1039+
"label_codes": [[]],
1040+
"acteur_type_code": [[]],
1041+
"acteur_service_codes": [[]],
1042+
"proposition_service_codes": [[]],
1043+
}
1044+
)
1045+
1046+
df_acteur, df_acteur_from_db, metadata = keep_acteur_changed(
1047+
df_normalized=df_normalized,
1048+
df_acteur_from_db=df_acteur_from_db,
1049+
dag_config=dag_config,
1050+
)
1051+
1052+
pd.testing.assert_frame_equal(
1053+
df_acteur.reset_index(drop=True),
1054+
df_expected.reset_index(drop=True),
1055+
check_dtype=False,
1056+
)
1057+
pd.testing.assert_frame_equal(
1058+
df_acteur_from_db.reset_index(drop=True),
1059+
df_expected_from_db.reset_index(drop=True),
1060+
check_dtype=False,
1061+
)
1062+
assert metadata == {}

‎qfdmo/admin/acteur.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,7 @@ class BaseActeurAdmin(admin.GISModelAdmin):
155155
"siret",
156156
"siren",
157157
"identifiant_unique",
158+
"identifiant_externe",
158159
"code_postal",
159160
"ville",
160161
"adresse",
@@ -163,6 +164,7 @@ class BaseActeurAdmin(admin.GISModelAdmin):
163164
search_fields = [
164165
"code_postal",
165166
"identifiant_unique",
167+
"identifiant_externe",
166168
"nom__unaccent",
167169
"siret",
168170
"siren",

‎qfdmo/management/commands/update_external_ids.py‎

Lines changed: 64 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,9 @@
22
import json
33

44
from django.core.management.base import BaseCommand
5+
from django.db import transaction
56

6-
from qfdmo.models.acteur import Acteur, RevisionActeur
7-
8-
CHUNK = 1000
7+
from qfdmo.models.acteur import Acteur, ActeurStatus, RevisionActeur
98

109

1110
class Command(BaseCommand):
@@ -39,61 +38,79 @@ def handle(self, *args, **options):
3938
with open(mapping_file, "r") as f:
4039
mapping = json.load(f)
4140

42-
for old_id, new_id in mapping.items():
43-
# Check if the ids are not empty
44-
if old_id and new_id:
45-
self.stdout.write(
46-
self.style.SUCCESS(f"Updating `{old_id}` to `{new_id}`")
47-
)
48-
else:
49-
self.stdout.write(
50-
self.style.WARNING(
51-
f"Skipping `{old_id}` to `{new_id}` : one of the ids is empty"
41+
with transaction.atomic():
42+
for old_id, new_id in mapping.items():
43+
# Check if the ids are not empty
44+
if old_id and new_id:
45+
self.stdout.write(
46+
self.style.SUCCESS(f"Mapping `{old_id}` to `{new_id}`")
5247
)
53-
)
54-
continue
48+
else:
49+
self.stdout.write(
50+
self.style.WARNING(
51+
f"Skipping `{old_id}` to `{new_id}` : one of the ids is"
52+
" empty"
53+
)
54+
)
55+
continue
56+
57+
# Update acteur if exists
58+
acteur = Acteur.objects.filter(
59+
identifiant_externe=old_id, source__code=source_code
60+
).first()
5561

56-
# Update acteur if exists
57-
acteur = Acteur.objects.filter(
58-
identifiant_externe=old_id, source__code=source_code
59-
).first()
62+
if not acteur:
63+
self.stdout.write(self.style.WARNING(f"Acteur {old_id} not found"))
64+
continue
6065

61-
if not acteur:
62-
self.stdout.write(self.style.WARNING(f"Acteur {old_id} not found"))
63-
continue
66+
# test acteur with this new external id already exists
67+
acteur_from_db = Acteur.objects.filter(
68+
identifiant_externe=new_id, source__code=source_code
69+
).first()
70+
id_index = 0
71+
while acteur_from_db:
72+
self.stdout.write(
73+
self.style.WARNING(
74+
f"Acteur {acteur_from_db.identifiant_externe} already"
75+
" exists, trying to find an unused id"
76+
)
77+
)
78+
id_index += 1
79+
new_id_indexed = f"{new_id}_{id_index}"
80+
acteur_from_db = Acteur.objects.filter(
81+
identifiant_externe=new_id_indexed, source__code=source_code
82+
).first()
83+
84+
statut = ActeurStatus.ACTIF
85+
if id_index:
86+
new_id = f"{new_id}_{id_index}"
87+
statut = ActeurStatus.INACTIF
6488

65-
if not dry_run:
6689
self.stdout.write(
6790
self.style.SUCCESS(
6891
f"Updating acteur {acteur.identifiant_unique} to {new_id}"
6992
)
7093
)
7194
acteur.identifiant_externe = new_id
95+
acteur.statut = statut
7296
acteur.save()
73-
else:
74-
self.stdout.write(
75-
self.style.WARNING(
76-
f"Dry run: would update Acteur {old_id} to {new_id}"
77-
)
78-
)
7997

80-
# Update revision acteur if exists
81-
revision_acteur = RevisionActeur.objects.filter(
82-
identifiant_externe=old_id, source__code=source_code
83-
).first()
98+
# Update revision acteur if exists
99+
revision_acteur = RevisionActeur.objects.filter(
100+
identifiant_externe=old_id, source__code=source_code
101+
).first()
84102

85-
if revision_acteur and not dry_run:
86-
self.stdout.write(
87-
self.style.SUCCESS(
88-
f"Updating revision acteur {revision_acteur.identifiant_unique}"
89-
f" to {new_id}"
103+
if revision_acteur:
104+
self.stdout.write(
105+
self.style.SUCCESS(
106+
"Updating revision acteur "
107+
f"{revision_acteur.identifiant_unique} to {new_id}"
108+
)
90109
)
91-
)
92-
revision_acteur.identifiant_externe = new_id
93-
revision_acteur.save()
94-
if revision_acteur and dry_run:
95-
self.stdout.write(
96-
self.style.WARNING(
97-
f"Dry run: would update RevisionActeur {old_id} to {new_id}"
98-
)
99-
)
110+
revision_acteur.identifiant_externe = new_id
111+
revision_acteur.statut = statut
112+
revision_acteur.save()
113+
114+
if dry_run:
115+
# Rollback the transaction
116+
raise Exception("Rolling back transaction because of dry run")

0 commit comments

Comments
 (0)