🚧(dashboard) add AnnuaireEntrepriseAPI client for enterprise data ret…

…rieval

Introduced a new API client to communicate with the Annuaire Entreprise API.
The client includes methods for fetching enterprise information using SIREN, with placeholder values for token and context.

.github/workflows/dashboard.yml

@@ -57,6 +57,8 @@ jobs:
       DASHBOARD_PROCONNECT_USER_ENDPOINT: proconnect_endpoint
       DASHBOARD_PROCONNECT_JWKS_ENDPOINT: proconnect_endpoint
       DASHBOARD_PROCONNECT_SESSION_END: proconnect_endpoint
+      DASHBOARD_ANNUAIRE_ENTREPRISE_API_URL: annuaire_api_endpoint
+      DASHBOARD_ANNUAIRE_ENTREPRISE_API_TOKEN: the_secret_token
     steps:
       - uses: actions/checkout@v4
       - name: Install pipenv
@@ -133,3 +135,5 @@ jobs:
           DASHBOARD_PROCONNECT_USER_ENDPOINT: proconnect_endpoint
           DASHBOARD_PROCONNECT_JWKS_ENDPOINT: proconnect_endpoint
           DASHBOARD_PROCONNECT_SESSION_END: proconnect_endpoint
+          DASHBOARD_ANNUAIRE_ENTREPRISE_API_URL: annuaire_api_endpoint
+          DASHBOARD_ANNUAIRE_ENTREPRISE_API_TOKEN: the_secret_token

env.d/dashboard

@@ -48,3 +48,7 @@ DASHBOARD_DEFAULT_FROM_EMAIL=no-reply@qualicharge.beta.gouv.fr
 # Brevo Emailing
 DASHBOARD_BREVO_API_KEY=
 DASHBOARD_CONSENT_VALIDATION_TEMPLATE_ID=3
+
+# Annuaire des Entreprises API
+DASHBOARD_ANNUAIRE_ENTREPRISE_API_URL="https://staging.entreprise.api.gouv.fr/v3/"
+DASHBOARD_ANNUAIRE_ENTREPRISE_API_TOKEN=

src/dashboard/CHANGELOG.md

@@ -22,6 +22,8 @@ and this project adheres to
 - add admin integration for Entity, DeliveryPoint and Consent
 - add mass admin action (make revoked) for consents
 - add validators for SIRET, NAF code and Zip code 
+- add API connector to the "Annuaire des Entreprises" API
+- retrieving company information from a siret and the "Annuaire des Entreprises" API
 - disallow mass action "delete" for consents in admin
 - block the updates of all new data if a consent has the status `REVOKED`
 - block the updates of all new data if a consent has the status `VALIDATED`

src/dashboard/Pipfile

@@ -30,6 +30,7 @@ pytest = "==8.3.4"
 pytest-cov = "==6.0.0"
 pytest-django = "==4.9.0"
 pytest-httpx = "==0.35.0"
+responses = "==0.25.6"
 ruff = "==0.9.5"
 
 [requires]

src/dashboard/apps/consent/views.py

@@ -32,7 +32,8 @@ class IndexView(BreadcrumbContextMixin, TemplateView):
     template_name = "consent/index.html"
     breadcrumb_current = _("Consent")
 
-    def get_context_data(self, **kwargs):  # noqa: D102
+    def get_context_data(self, **kwargs):
+        """Add the user's entities to the context."""
         context = super().get_context_data(**kwargs)
         context["entities"] = self.request.user.get_entities()
         return context
@@ -81,11 +82,7 @@ def form_valid(self, form):
         return super().form_valid(form)
 
     def get_context_data(self, **kwargs):
-        """Add the user's entities to the context.
-
-        Adds to the context the entities that the user has permission to access.
-        If a slug is provided, adds the entity corresponding to the slug.
-        """
+        """Add context data for the view."""
         context = super().get_context_data(**kwargs)
         context["control_authority"] = settings.CONSENT_CONTROL_AUTHORITY
         context["entities"] = self._get_entities()

src/dashboard/apps/core/annuaire_entreprise_api.py

@@ -0,0 +1,132 @@
+"""Dashboard core annuaire entreprise API."""
+
+import requests
+from django.conf import settings
+
+from apps.core.validators import validate_siren, validate_siret
+
+
+class CompanyInformationAdapter:
+    """Encapsulation of company information for simplified access to key details.
+
+    Extracts and organizes essential company data from a json structure.
+    Includes functionality to reformat specific codes, such as the NAF code.
+
+    Attributes:
+        company_info (dict): The raw company information data.
+        name (str): The legal company name.
+        legal_form (str): The legal form of the company.
+        siren (str): The unique company identifier (SIREN).
+        siret_siege (str): The SIRET code of the head office.
+    """
+
+    def __init__(self, company_info: dict):
+        """Represents a company information with base attributes."""
+        self.company_info: dict = company_info["data"]
+
+        self.name: str = self.company_info["personne_morale_attributs"][
+            "raison_sociale"
+        ]
+        self.legal_form: str = self.company_info["forme_juridique"]["libelle"]
+        self.siren: str = self.company_info["siren"]
+        self.siret_siege: str = self.company_info["siret_siege_social"]
+
+    @property
+    def naf(self) -> str:
+        """Reformat NAF code.
+
+        The NAF code from the API has the format "12.34A", but we need "1234A" to match.
+        """
+        return self.company_info["activite_principale"]["code"].replace(".", "")
+
+
+class CompanyAddressAdapter:
+    """Encapsulation of company address info for simplified access to key details.
+
+    Extract and format the address details for a company.
+    Includes functionality to reformat address line.
+
+    Attributes:
+        company_address (dict): The raw company address data.
+        address_1 (str): Main address line.
+        address_2 (str): Address complement.
+        city (str): The city of the company.
+        zip_code (str): The zip code of the company.
+    """
+
+    def __init__(self, company_address: dict):
+        """Represents a company address."""
+        self.company_address: dict = company_address["data"]
+
+        self.address_2: str = self.company_address["complement_adresse"]
+        self.city: str = self.company_address["libelle_commune"]
+        self.zip_code: str = self.company_address["code_postal"]
+
+    @property
+    def address_1(self) -> str:
+        """Returns a formatted address line based on the address components."""
+        street_number: str = self.company_address["numero_voie"]
+        street_repetition: str = self.company_address["indice_repetition_voie"]
+        street_type: str = self.company_address["type_voie"]
+        street_label: str = self.company_address["libelle_voie"]
+
+        return (
+            f"{street_number + ' ' if street_number else ''}"
+            f"{street_repetition + ' ' if street_repetition else ''}"
+            f"{street_type + ' ' if street_type else ''}"
+            f"{street_label or ''}"
+        )
+
+
+class AnnuaireEntrepriseAPI:
+    """Annuaire entreprise API client."""
+
+    def __init__(self):
+        """Initialize the API client."""
+        self.api_url = settings.ANNUAIRE_ENTREPRISE_API_URL
+        self.token = settings.ANNUAIRE_ENTREPRISE_API_TOKEN
+        self.context = self._get_context()
+
+    @staticmethod
+    def _get_context():
+        """Get a formatted token string used for specific API-related context.
+
+        Returns:
+            str: The constructed token string in query parameter format.
+        """
+        context = settings.ANNUAIRE_ENTREPRISE_API_CONTEXT["context"]
+        context_object = settings.ANNUAIRE_ENTREPRISE_API_CONTEXT["object"]
+        recipient = settings.ANNUAIRE_ENTREPRISE_API_CONTEXT["recipient"]
+
+        return f"?context={context}&object={context_object}&recipient={recipient}"
+
+    def _make_request(self, endpoint, params=None):
+        """Make a request to the API."""
+        url = f"{self.api_url}/{endpoint}"
+
+        headers = {"Authorization": f"Bearer {self.token}"}
+        response = requests.get(url, headers=headers, params=params, timeout=5000)
+        response.raise_for_status()
+        return response.json()
+
+    def get_company_info(self, siren, json=False):
+        """Get the company information from the API."""
+        validate_siren(siren)
+
+        endpoint = f"insee/sirene/unites_legales/{siren}{self.context}"
+        company_info = self._make_request(endpoint)
+
+        if json:
+            return company_info
+        return CompanyInformationAdapter(company_info)
+
+    def get_company_address(self, siret, json=False):
+        """Get the company information from the API."""
+        validate_siret(siret)
+
+        endpoint = f"insee/sirene/etablissements/{siret}/adresse{self.context}"
+        company_address = self._make_request(endpoint)
+
+        if json:
+            return company_address
+        return CompanyAddressAdapter(company_address)

src/dashboard/apps/core/helpers.py

@@ -0,0 +1,58 @@
+"""Dashboard core helpers."""
+
+from typing import Optional
+
+from apps.auth.models import DashboardUser
+from apps.core.annuaire_entreprise_api import (
+    AnnuaireEntrepriseAPI,
+    CompanyAddressAdapter,
+    CompanyInformationAdapter,
+)
+from apps.core.models import Entity
+from apps.core.validators import validate_siret
+
+
+def sync_entity_from_siret(siret: str, user: Optional[DashboardUser] = None) -> Entity:
+    """Retrieve, Update or create and populate entity.
+
+    Retrieve and populate entity data based on SIRET input.
+    The function uses an external API to fetch company information and address using
+    the provided SIRET number.
+    If the entity does not already exist in the database, it will be created.
+    Optionally associates a user with the entity.
+
+    Arguments:
+        siret (str): The SIRET number used to identify and fetch the company's
+        information.
+        user (Optional[User]): A user instance to be associated with the created or
+        updated entity. Defaults to None.
+
+    Returns:
+        Entity: The created or updated instance of the Entity model.
+    """
+    api: AnnuaireEntrepriseAPI = AnnuaireEntrepriseAPI()
+
+    validate_siret(siret)
+    siren: str = siret[:9]
+
+    company_info: CompanyInformationAdapter = api.get_company_info(siren)
+    company_address: CompanyAddressAdapter = api.get_company_address(
+        company_info.siret_siege
+    )
+
+    entity, created = Entity.objects.update_or_create(
+        name=company_info.name,
+        siret=siret,
+        legal_form=company_info.legal_form,
+        naf=company_info.naf,
+        address_1=company_address.address_1,
+        address_2=company_address.address_2,
+        address_city=company_address.city,
+        address_zip_code=company_address.zip_code,
+        defaults={"name": company_info.name, "siret": siret},
+    )
+
+    if user:
+        entity.users.add(user)
+
+    return entity

src/dashboard/apps/core/management/__init__.py

@@ -0,0 +1 @@
+"""Dashboard core app management."""

src/dashboard/apps/core/management/commands/__init__.py

@@ -0,0 +1 @@
+"""Dashboard core app management commands."""

src/dashboard/apps/core/management/commands/populateentity.py

@@ -0,0 +1,31 @@
+"""Core management commands."""
+
+from django.core.management.base import BaseCommand
+
+from apps.core.helpers import sync_entity_from_siret
+
+
+class Command(BaseCommand):
+    """Populate an entity with data from the API `annuaire des entreprises`.
+
+    Note: this command is actually meant to be used in dev only.
+    """
+
+    help = __doc__
+
+    def add_arguments(self, parser):
+        """Add arguments to the command."""
+        parser.add_argument("siret", type=str)
+
+    def handle(self, *args, **options):
+        """Executes the command for populating an entity."""
+        self.stdout.write(
+            self.style.WARNING("this command is actually meant to be used in dev only.")
+        )
+        self.stdout.write(
+            self.style.NOTICE(
+                'Querying "Annuaire des Entreprises" API and populating entity...'
+            )
+        )
+        sync_entity_from_siret(siret=options["siret"])
+        self.stdout.write(self.style.SUCCESS("Done."))