✨(web-infrastructure) set offers.source_id as a foreign key (#651)

* ✨(web-infrastructure) set offers.source_id as a foreign key

* ♻️(web-infrastructure) offer.source_id is a UUID

* ✅(web-infrastructure) review PR comments

* ♻️(web-infrastructure) drop index on source_id

Author: AntoineAugusti

src/web/application/ingestion/interfaces/archive_offer_by_reference_input.py

@@ -1,7 +1,8 @@
 from dataclasses import dataclass
+from uuid import UUID
 
 
 @dataclass
 class ArchiveOfferByReferenceInput:
     reference: str
-    source_id: str
+    source_id: UUID

src/web/domain/entities/offer.py

@@ -28,7 +28,7 @@ class Offer(IEntity):
     publication_date: datetime
     beginning_date: Optional[LimitDate]
     family_code: Optional[str] = None
-    source_id: Optional[str] = None
+    source_id: UUID = field(default_factory=uuid4)
     processing: bool = False
     processed_at: Optional[datetime] = None
     archived_at: Optional[datetime] = None

src/web/domain/repositories/offers_repository_interface.py

@@ -29,7 +29,7 @@ def get_by_ids(self, offer_ids: List[UUID]) -> List[Offer]: ...
     def get_by_external_id(self, external_id: str) -> Offer: ...
 
     def get_by_reference_and_source_id(
-        self, reference: str, source_id: str
+        self, reference: str, source_id: UUID
     ) -> Offer: ...
 
     def get_by_external_ids(self, external_ids: List[str]) -> List[Offer]: ...

src/web/infrastructure/di/ingestion/ingestion_container.py

@@ -102,6 +102,7 @@ class IngestionContainer(containers.DeclarativeContainer):
             concours_repository=concours_repository,
             offers_repository=offers_repository,
             metiers_repository=metiers_repository,
+            source_repository=source_repository,
         )
     )
 

src/web/infrastructure/django_apps/shared/migrations/0030_alter_offermodel_source_id_not_null.py

@@ -0,0 +1,51 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+def hydrate_source(apps, schema_editor):
+    OfferModel = apps.get_model("shared", "OfferModel")
+    SourceModel = apps.get_model("ingestion", "SourceModel")
+    source = SourceModel.objects.first()
+    if source:
+        OfferModel.objects.update(source=source)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("ingestion", "0024_alter_sourcemodel_id_alter_sourcemodel_source_id"),
+        ("shared", "0029_populate_offer_source_id"),
+    ]
+
+    operations = [
+        migrations.RemoveIndex(
+            model_name="offermodel",
+            name="offers_source_id_idx",
+        ),
+        migrations.RemoveField(
+            model_name="offermodel",
+            name="source_id",
+        ),
+        migrations.AddField(
+            model_name="offermodel",
+            name="source",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="offers",
+                to="ingestion.sourcemodel",
+                to_field="source_id",
+            ),
+        ),
+        migrations.RunPython(hydrate_source, reverse_code=migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name="offermodel",
+            name="source",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="offers",
+                to="ingestion.sourcemodel",
+                to_field="source_id",
+            ),
+        ),
+    ]

src/web/infrastructure/django_apps/shared/models/offer.py

@@ -11,6 +11,7 @@
 from domain.value_objects.localisation import Localisation
 from domain.value_objects.region import Region
 from domain.value_objects.verse import Verse
+from infrastructure.django_apps.ingestion.models.source import SourceModel
 
 
 class OfferModel(models.Model):
@@ -42,7 +43,12 @@ class OfferModel(models.Model):
     organization = models.CharField(max_length=500)
     offer_url = models.URLField(null=True, blank=True)
     code_emploi_csp = models.CharField(max_length=50, null=True, blank=True)
-    source_id = models.UUIDField(null=True, blank=True)
+    source = models.ForeignKey(
+        SourceModel,
+        to_field="source_id",
+        on_delete=models.PROTECT,
+        related_name="offers",
+    )
 
     # Localisation fields stored separately
     area = models.CharField(max_length=2, null=True, blank=True)
@@ -66,7 +72,6 @@ class Meta:
         verbose_name_plural = "Offers"
         indexes = [
             models.Index(fields=["external_id"]),
-            models.Index(fields=["source_id"], name="offers_source_id_idx"),
         ]
 
     def to_entity(self) -> Offer:
@@ -109,7 +114,7 @@ def to_entity(self) -> Offer:
             processed_at=self.processed_at,
             archived_at=self.archived_at,
             family_code=self.code_emploi_csp,
-            source_id=str(self.source_id) if self.source_id else None,
+            source_id=self.source_id,
         )
 
     @classmethod

src/web/infrastructure/gateways/ingestion/document_cleaner.py

@@ -12,6 +12,7 @@
 from domain.repositories.corps_repository_interface import ICorpsRepository
 from domain.repositories.metier_repository_interface import IMetierRepository
 from domain.repositories.offers_repository_interface import IOffersRepository
+from domain.repositories.source_repository_interface import ISourceRepository
 from domain.services.document_cleaner_interface import CleaningResult, IDocumentCleaner
 from domain.services.logger_interface import ILogger
 from infrastructure.gateways.ingestion.concours_cleaner import ConcoursCleaner
@@ -28,11 +29,14 @@ def __init__(
         concours_repository: IConcoursRepository,
         offers_repository: IOffersRepository,
         metiers_repository: IMetierRepository,
+        source_repository: ISourceRepository,
     ):
         self._cleaners = {
             DocumentType.CORPS: CorpsCleaner(logger, corps_repository),
             DocumentType.CONCOURS: ConcoursCleaner(logger, concours_repository),
-            DocumentType.OFFERS: OffersCleaner(logger, offers_repository),
+            DocumentType.OFFERS: OffersCleaner(
+                logger, offers_repository, source_repository
+            ),
             DocumentType.METIERS: MetierCleaner(logger, metiers_repository),
         }
 

src/web/infrastructure/gateways/ingestion/offers_cleaner.py

@@ -1,5 +1,6 @@
 from datetime import datetime
 from typing import List, Optional
+from uuid import UUID
 
 from pydantic import HttpUrl, ValidationError
 
@@ -8,6 +9,7 @@
 from domain.exceptions.document_error import InvalidDocumentTypeError
 from domain.exceptions.offer_errors import OfferDoesNotExist
 from domain.repositories.offers_repository_interface import IOffersRepository
+from domain.repositories.source_repository_interface import ISourceRepository
 from domain.services.document_cleaner_interface import CleaningResult, IDocumentCleaner
 from domain.services.logger_interface import ILogger
 from domain.value_objects.area import GeographicalArea
@@ -34,15 +36,27 @@
 
 
 class OffersCleaner(IDocumentCleaner[Offer]):
-    def __init__(self, logger: ILogger, offers_repository: IOffersRepository):
+    def __init__(
+        self,
+        logger: ILogger,
+        offers_repository: IOffersRepository,
+        source_repository: ISourceRepository,
+    ):
         self.logger = logger
         self.offers_repository = offers_repository
+        self.source_repository = source_repository
 
     def clean(self, raw_documents: List[Document]) -> CleaningResult[Offer]:
         for document in raw_documents:
             if document.type != DocumentType.OFFERS:
                 raise InvalidDocumentTypeError(document.type.value)  # todo: test
 
+        sources = self.source_repository.get_all()
+        if not sources:
+            raise ValueError("No source found in repository")
+        # TODO: replace when we will have multiple instances of Talentsoft
+        source_id = sources[0].source_id
+
         validated_offers = []
         cleaning_errors = []
 
@@ -63,7 +77,7 @@ def clean(self, raw_documents: List[Document]) -> CleaningResult[Offer]:
         for talentsoft_offer in validated_offers:
             self.logger.info(f"Processing offer {talentsoft_offer.reference}")
             try:
-                offer = self._map_talentsoft_to_offer(talentsoft_offer)
+                offer = self._map_talentsoft_to_offer(talentsoft_offer, source_id)
                 offers_list.append(offer)
                 self.logger.debug(
                     f"Successfully processed offer {talentsoft_offer.reference}"
@@ -88,7 +102,7 @@ def clean(self, raw_documents: List[Document]) -> CleaningResult[Offer]:
         return CleaningResult(entities=offers_list, cleaning_errors=cleaning_errors)
 
     def _map_talentsoft_to_offer(
-        self, talentsoft_offer: TalentsoftDetailOffer
+        self, talentsoft_offer: TalentsoftDetailOffer, source_id: UUID
     ) -> Offer:
         # Extract verse from salaryRange if available
         ts_verse = (
@@ -148,6 +162,7 @@ def _map_talentsoft_to_offer(
             publication_date=publication_date,
             beginning_date=beginning_date,
             family_code=family_code_value,
+            source_id=source_id,
         )
         try:
             existing_offer = self.offers_repository.get_by_external_id(

src/web/infrastructure/repositories/shared/postgres_offers_repository.py

@@ -116,7 +116,7 @@ def get_by_external_id(self, external_id: str) -> Offer:
         except OfferModel.DoesNotExist as e:
             raise OfferDoesNotExist(external_id) from e
 
-    def get_by_reference_and_source_id(self, reference: str, source_id: str) -> Offer:
+    def get_by_reference_and_source_id(self, reference: str, source_id: UUID) -> Offer:
         try:
             offer_model = OfferModel.objects.get(
                 external_id__endswith=f"-{reference}", source_id=source_id

src/web/presentation/ingestion/mappers.py

@@ -1,4 +1,5 @@
 from typing import Optional
+from uuid import UUID
 
 from domain.ddd.mapper_interface import IToDomainMapper
 from domain.entities.offer import Offer
@@ -57,5 +58,5 @@ def to_domain(self, data: Optional[dict]) -> Optional[Offer]:
             localisation=self._localisation_mapper.to_domain(raw_localisation),
             beginning_date=LimitDate(debut_contrat) if debut_contrat else None,
             family_code=data["profession"]["metier"],
-            source_id=data["identification"]["source"],
+            source_id=UUID(data["identification"]["source"]),
         )