[#2931] Add signal for updating company data from KVK API on login

src/open_inwoner/accounts/signals.py

@@ -9,6 +9,8 @@
 from open_inwoner.accounts.models import User
 from open_inwoner.haalcentraal.models import HaalCentraalConfig
 from open_inwoner.haalcentraal.utils import update_brp_data_in_db
+from open_inwoner.kvk.client import KvKClient
+from open_inwoner.kvk.signals import company_branch_selected
 from open_inwoner.openklant.services import OpenKlant2Service, eSuiteKlantenService
 from open_inwoner.utils.logentry import user_action
 
@@ -26,6 +28,35 @@
 }
 
 
+@receiver(company_branch_selected)
+def update_company_from_kvk_api(sender, *args, **kwargs):
+    request = kwargs["request"]
+    user = request.user
+    kvk = user.kvk
+    vestigingsnummer = kwargs["vestigingsnummer"]
+
+    kvk_client = KvKClient()
+
+    vestiging = kvk_client.get_vestiging(kvk=kvk, vestigingsnummer=vestigingsnummer)
+
+    binnenlandsadres = vestiging.get("adres", {}).get("binnenlandsAdres", {})
+
+    fields = {
+        "company_name": vestiging.get("naam"),
+        "city": binnenlandsadres.get("plaats"),
+        "postcode": binnenlandsadres.get("postcode"),
+        "street": binnenlandsadres.get("straatnaam"),
+        "housenumber": binnenlandsadres.get("huisnummer"),
+    }
+
+    for field, new_value in fields.items():
+        current_value = getattr(user, field, None)
+        if new_value and new_value != current_value:
+            setattr(user, field, new_value)
+
+    user.save()
+
+
 @receiver(user_logged_in)
 def update_user_from_klant_on_login(sender, user, request, *args, **kwargs):
     # This additional guard is mainly to facilitate easier testing, where not

src/open_inwoner/kvk/client.py

@@ -1,13 +1,15 @@
 import logging
 from functools import cached_property
-from typing import cast
 from urllib.parse import urlencode
 
 import requests
 from requests.exceptions import JSONDecodeError
 
-from .constants import Bedrijf, CompanyType
+from open_inwoner.utils.decorators import cache
+
+from .constants import CompanyType
 from .models import KvKConfig
+from .types import BedrijfValidator
 
 logger = logging.getLogger(__name__)
 
@@ -123,10 +125,29 @@ def get_all_company_branches(self, kvk: str, **kwargs) -> list[dict | None]:
 
         return branches
 
-    def get_company_branch(self, vestigingsnummer: str, **kwargs) -> dict:
-        kwargs.update({"vestigingsnummer": vestigingsnummer})
-        vestiging = self.search(**kwargs).get("resultaten", {})
-        return cast(Bedrijf, vestiging)
+    @cache("kvk:{kvk}vestiging:{vestigingsnummer}")
+    def get_vestiging(self, kvk: str, vestigingsnummer: str) -> dict:
+        response = self.search(vestigingsnummer=vestigingsnummer)
+
+        results = response.get("resultaten")
+        if not results:
+            logger.error(
+                "No company branch with vestigingsnummer %s found", vestigingsnummer
+            )
+            return {}
+
+        # the same vestigingsnummer could be associated with different kvk numbers
+        try:
+            vestiging = next(r for r in results if r["kvkNummer"] == kvk)
+        except StopIteration:
+            logger.error(
+                "Company branch with vestigingsnummer %s and kvk number %s not found",
+                vestigingsnummer,
+                kvk,
+            )
+            return {}
+
+        return BedrijfValidator.validate_python(vestiging)
 
     def retrieve_rsin_with_kvk(self, kvk, **kwargs) -> str | None:
         basisprofiel = self._request(

src/open_inwoner/kvk/constants.py

@@ -1,5 +1,3 @@
-from typing import Literal, Optional, TypedDict
-
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 
@@ -8,40 +6,3 @@ class CompanyType(models.TextChoices):
     hoofdvestiging = "hoofdvestiging", _("Hoofdvestiging")
     nevenvestiging = "nevenvestiging", _("Nevenvestiging")
     rechtspersoon = "rechtspersoon", _("Rechtspersoon")
-
-
-AdresType = Literal["bezoekadres", "postadres"]
-BedrijfsType = Literal["hoofdvestiging", "nevenvestiging", "rechtspersoon"]
-
-
-class BedrijfsBinnenlandsAdres(TypedDict, total=False):
-    type: AdresType
-    plaats: Optional[str]
-    postcode: Optional[str]
-    straatnaam: Optional[str]
-    huisnummer: Optional[str]
-    huisletter: Optional[str]
-    postbusnummer: Optional[str]
-
-
-class BedrijfsBuitenlandsAdres(TypedDict, total=False):
-    straatHuisnummer: Optional[str]
-    postcodeWoonplaats: Optional[str]
-    land: str
-
-
-class BedrijfsAdres(TypedDict, total=False):
-    binnenlandsAdres: Optional[BedrijfsBinnenlandsAdres]
-    buitenlandsAdres: Optional[BedrijfsBuitenlandsAdres]
-
-
-class Bedrijf(TypedDict, total=False):
-    kvkNummer: str
-    rsin: Optional[str]
-    vestigingsnummer: Optional[str]
-    naam: str
-    adres: BedrijfsAdres
-    type: BedrijfsType
-    actief: Optional[str]
-    vervallenNaam: Optional[str]
-    _links: Optional[dict]

src/open_inwoner/kvk/signals.py

@@ -1,6 +1,7 @@
 import logging
 
 from django.contrib.auth.signals import user_logged_in
+from django.db.models.signals import Signal
 from django.dispatch import receiver
 from django.utils.translation import gettext as _
 
@@ -34,3 +35,6 @@ def on_kvk_change(sender, user, request, *args, **kwargs):
             user.rsin = rsin
             user.is_prepopulated = True
             user.save()
+
+
+company_branch_selected = Signal()

src/open_inwoner/kvk/tests/mocks.py

@@ -69,6 +69,57 @@
     },
 }
 
+
+nevenvestigingen = {
+    "pagina": 1,
+    "resultatenPerPagina": 10,
+    "totaal": 2,
+    "resultaten": [
+        {
+            "kvkNummer": "68750110",
+            "vestigingsnummer": "000037178601",
+            "naam": "Test BV Donald Nevenvestiging",
+            "adres": {
+                "binnenlandsAdres": {
+                    "type": "bezoekadres",
+                    "straatnaam": "Brinkerinckbaan",
+                    "plaats": "Diepenveen",
+                }
+            },
+            "type": "nevenvestiging",
+            "_links": {
+                "basisprofiel": {
+                    "href": "https://api.kvk.nl/test/api/v1/basisprofielen/68750110"
+                },
+                "vestigingsprofiel": {
+                    "href": "https://api.kvk.nl/test/api/v1/vestigingsprofielen/000037178601"
+                },
+            },
+        },
+        {
+            "kvkNummer": "12345678",
+            "vestigingsnummer": "000037178601",
+            "naam": "Andere Nevenvestiging",
+            "adres": {
+                "binnenlandsAdres": {
+                    "type": "bezoekadres",
+                    "straatnaam": "Brinkerinckbaan",
+                    "plaats": "Fantasieland",
+                }
+            },
+            "type": "nevenvestiging",
+            "_links": {
+                "basisprofiel": {
+                    "href": "https://api.kvk.nl/test/api/v1/basisprofielen/68750110"
+                },
+                "vestigingsprofiel": {
+                    "href": "https://api.kvk.nl/test/api/v1/vestigingsprofielen/000037178601"
+                },
+            },
+        },
+    ],
+}
+
 multiple_branches = {
     "pagina": 1,
     "resultatenPerPagina": 10,

src/open_inwoner/kvk/tests/test_api.py

@@ -1,11 +1,13 @@
+import datetime
 from unittest.mock import patch
 
 from django.test import TestCase
 
 import requests_mock
+from freezegun import freeze_time
 from requests.exceptions import InvalidJSONError, SSLError
 
-from ..client import KvKClient
+from ..client import KvKClient, logger
 from ..models import KvKConfig
 from . import mocks
 from .factories import CLIENT_CERT, CLIENT_CERT_PAIR, SERVER_CERT
@@ -179,6 +181,110 @@ def test_search_all_branches(self, m):
             ],
         )
 
+    def test_search_vestiging(self, m):
+        m.return_value.json.return_value = mocks.nevenvestigingen
+
+        # test caching
+        with freeze_time("2024-12-13 12:00") as frozen_time:
+            branch = self.kvk_client.get_vestiging(
+                kvk="68750110", vestigingsnummer="000037178601"
+            )
+            self.kvk_client.get_vestiging(
+                kvk="68750110", vestigingsnummer="000037178601"
+            )
+
+            m.assert_called_once()
+
+            frozen_time.tick(delta=datetime.timedelta(hours=1))
+
+            self.kvk_client.get_vestiging(
+                kvk="68750110", vestigingsnummer="000037178601"
+            )
+
+            with self.assertRaises(AssertionError):
+                m.assert_called_once()
+
+        # test result
+        self.assertEqual(
+            branch,
+            {
+                "kvkNummer": "68750110",
+                "vestigingsnummer": "000037178601",
+                "naam": "Test BV Donald Nevenvestiging",
+                "adres": {
+                    "binnenlandsAdres": {
+                        "type": "bezoekadres",
+                        "straatnaam": "Brinkerinckbaan",
+                        "plaats": "Diepenveen",
+                    },
+                },
+                "type": "nevenvestiging",
+                "_links": {
+                    "basisprofiel": {
+                        "href": "https://api.kvk.nl/test/api/v1/basisprofielen/68750110"
+                    },
+                    "vestigingsprofiel": {
+                        "href": "https://api.kvk.nl/test/api/v1/vestigingsprofielen/000037178601"
+                    },
+                },
+            },
+        )
+
+    def test_search_vestiging_cache_invalidation_with_new_kvk(self, m):
+        m.return_value.json.return_value = mocks.nevenvestigingen
+
+        branch = self.kvk_client.get_vestiging(
+            kvk="68750110", vestigingsnummer="000037178601"
+        )
+        branch = self.kvk_client.get_vestiging(
+            kvk="12345678", vestigingsnummer="000037178601"
+        )
+
+        with self.assertRaises(AssertionError):
+            m.assert_called_once()
+
+        self.assertEqual(
+            branch,
+            {
+                "kvkNummer": "12345678",
+                "vestigingsnummer": "000037178601",
+                "naam": "Andere Nevenvestiging",
+                "adres": {
+                    "binnenlandsAdres": {
+                        "type": "bezoekadres",
+                        "straatnaam": "Brinkerinckbaan",
+                        "plaats": "Fantasieland",
+                    }
+                },
+                "type": "nevenvestiging",
+                "_links": {
+                    "basisprofiel": {
+                        "href": "https://api.kvk.nl/test/api/v1/basisprofielen/68750110"
+                    },
+                    "vestigingsprofiel": {
+                        "href": "https://api.kvk.nl/test/api/v1/vestigingsprofielen/000037178601"
+                    },
+                },
+            },
+        )
+
+    def test_search_vestiging_invalid_combination_kvk_vestigingsnummer(self, m):
+        m.return_value.json.return_value = mocks.nevenvestigingen
+
+        with self.assertLogs(logger.name, level="ERROR") as cm:
+            branch = self.kvk_client.get_vestiging(
+                kvk="00000000", vestigingsnummer="000037178601"
+            )
+
+            self.assertEqual(len(cm.output), 1)
+            self.assertEqual(
+                cm.output[0],
+                "ERROR:open_inwoner.kvk.client:Company branch with vestigingsnummer "
+                "000037178601 and kvk number 00000000 not found",
+            )
+
+        self.assertEqual(branch, {})
+
     def test_no_search_without_config(self, m):
         m.return_value.json.return_value = mocks.multiple_branches