Merge pull request #15 from Amsterdam/feature/127644-dso-api

Implement search vve by bag id

.env

@@ -22,7 +22,9 @@ OIDC_OP_TOKEN_ENDPOINT=https://login.microsoftonline.com/72fca1b1-2c2e-4376-a445
 OIDC_OP_USER_ENDPOINT=https://graph.microsoft.com/oidc/userinfo
 OIDC_OP_JWKS_ENDPOINT=https://login.microsoftonline.com/72fca1b1-2c2e-4376-a445-294d80196804/discovery/v2.0/keys
 UWSGI_HTTP=0.0.0.0:8000
-
-# CORS
 CORS_ALLOWED_ORIGINS=http://0.0.0.0:5173,http://localhost:5173,https://zwd-frontend.localhost
 LOCAL_DEVELOPMENT_AUTHENTICATION=False
+DSO_CLIENT_ID=test
+DSO_CLIENT_SECRET=test
+DSO_AUTH_URL=https://
+DSO_API_URL=https://

.gitignore

@@ -160,3 +160,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+.local.env

README.md

@@ -26,6 +26,18 @@ docker-compose -f docker-compose.local.yml up
 
 Visit the Admin at http://localhost:8081/admin/
 
+
+If you want to make use of the DSO api the following vars need to be set in the .local.env file:
+
+```bash
+DSO_API_URL=<url>
+DSO_CLIENT_SECRET=<key>
+DSO_AUTH_URL=<key>
+DSO_API_URL=<key>
+```
+
+
+
 ## Swagger
 
 http://localhost:8081/api/schema/swagger/
@@ -55,7 +67,6 @@ bash bin/install_pre_commit.sh
 ## Running tests
 
 Containers should be running to run tests via docker.
-
 ```bash
 docker compose -f docker-compose.local.yml -f docker-compose.override.yml up -d
 docker compose exec -T zwd-backend python manage.py test /app/apps

app/apps/address/admin.py

@@ -0,0 +1 @@
+# Register your models here.

app/apps/address/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class AddressConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "apps.address"

app/apps/address/migrations/0001_initial.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.0.8 on 2024-10-09 11:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="Address",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("bag_id", models.CharField(max_length=255, unique=True)),
+            ],
+        ),
+    ]

app/apps/address/models.py

@@ -0,0 +1,5 @@
+from django.db import models
+
+
+class Address(models.Model):
+    bag_id = models.CharField(max_length=255, null=False, unique=True)

app/apps/address/serializers.py

@@ -0,0 +1,8 @@
+from apps.address.models import Address
+from rest_framework import serializers
+
+
+class addressSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Address
+        fields = "__all__"

app/apps/address/tests.py

@@ -0,0 +1 @@
+# Create your tests here.

app/apps/address/views.py

@@ -0,0 +1,12 @@
+from apps import address
+from apps.address.serializers import addressSerializer
+from apps.homeownerassociation.mixins import HomeownerAssociationMixin
+from rest_framework import viewsets
+
+
+class AddressViewset(
+    viewsets.ViewSet,
+    HomeownerAssociationMixin,
+):
+    serializer_class = addressSerializer
+    model = address

app/apps/homeownerassociation/admin.py

@@ -0,0 +1,16 @@
+from django.contrib import admin
+
+from apps.homeownerassociation.models import HomeownerAssociation
+
+
+@admin.register(HomeownerAssociation)
+class HomeownerAssociationAdmin(admin.ModelAdmin):
+    list_display = (
+        "id",
+        "name",
+        "build_year",
+        "number_of_appartments",
+        "created_at",
+        "updated_at",
+    )
+    search_fields = ("id",)

app/apps/homeownerassociation/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class HomeownerassociationConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "apps.homeownerassociation"

app/apps/homeownerassociation/migrations/0001_initial.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.0.8 on 2024-10-09 11:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="HomeownerAssociation",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("name", models.CharField(max_length=255, unique=True)),
+                ("build_year", models.IntegerField()),
+                ("number_of_appartments", models.IntegerField()),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+            ],
+        ),
+    ]

app/apps/homeownerassociation/mixins.py

@@ -0,0 +1,17 @@
+from rest_framework.decorators import action
+from rest_framework.response import Response
+from apps.homeownerassociation.models import HomeownerAssociation
+from apps.homeownerassociation.serializers import HomeownerAssociationSerializer
+
+
+class HomeownerAssociationMixin:
+    @action(
+        detail=True,
+        methods=["get"],
+        url_path="homeowner-association",
+        serializer_class=HomeownerAssociationSerializer,
+    )
+    def get_by_bag_id(self, request, pk=None):
+        model = HomeownerAssociation.get_or_create_hoa_by_bag_id(pk)
+        serializer = HomeownerAssociationSerializer(model)
+        return Response(serializer.data)

app/apps/homeownerassociation/models.py

@@ -0,0 +1,29 @@
+from django.db import models
+from clients.dso_client import DsoClient
+
+
+class HomeownerAssociation(models.Model):
+    name = models.CharField(max_length=255, unique=True)
+    build_year = models.IntegerField()
+    number_of_appartments = models.IntegerField()
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    def get_or_create_hoa_by_bag_id(bag_id):
+        client = DsoClient()
+        hoa_name = client.get_hoa_name_by_bag_id(bag_id)
+        # Check if the HomeownerAssociation already exists in the database
+        existing_hoa = HomeownerAssociation.objects.filter(name=hoa_name).first()
+        if existing_hoa:
+            return existing_hoa
+
+        hoa_response = client.get_hoa_by_name(hoa_name)
+        distinct_hoa_response = list(
+            {hoa["votIdentificatie"]: hoa for hoa in hoa_response}.values()
+        )
+        model = HomeownerAssociation.objects.create(
+            name=hoa_name,
+            build_year=distinct_hoa_response[0].get("pndOorspronkelijkBouwjaar"),
+            number_of_appartments=len(distinct_hoa_response),
+        )
+        return model

app/apps/homeownerassociation/serializers.py

@@ -0,0 +1,13 @@
+from apps.homeownerassociation.models import HomeownerAssociation
+from rest_framework import serializers
+
+
+class HomeownerAssociationSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = HomeownerAssociation
+        fields = [
+            "id",
+            "name",
+            "build_year",
+            "number_of_appartments",
+        ]

app/apps/homeownerassociation/tests/tests_models.py

@@ -0,0 +1,81 @@
+from django.test import TestCase
+from unittest.mock import patch
+from apps.homeownerassociation.models import HomeownerAssociation
+
+
+class HomeownerAssociationModelTest(TestCase):
+
+    @patch("apps.homeownerassociation.models.DsoClient")
+    def test_get_or_create_hoa_by_bag_id_existing_hoa(self, MockDsoClient):
+        # Mock the DsoClient and its methods
+        mock_client = MockDsoClient.return_value
+        mock_client.get_hoa_name_by_bag_id.return_value = "Test HOA"
+
+        # Create an existing HOA
+        existing_hoa = HomeownerAssociation.objects.create(
+            name="Test HOA", build_year=2000, number_of_appartments=10
+        )
+
+        # Call the method
+        result = HomeownerAssociation.get_or_create_hoa_by_bag_id("some_bag_id")
+
+        # Assert the existing HOA is returned
+        self.assertEqual(result, existing_hoa)
+        mock_client.get_hoa_name_by_bag_id.assert_called_once_with("some_bag_id")
+
+    @patch("apps.homeownerassociation.models.DsoClient")
+    def test_get_or_create_hoa_by_bag_id_existing_hoa_no_new_hoa(self, MockDsoClient):
+        # Mock the DsoClient and its methods
+        mock_client = MockDsoClient.return_value
+        mock_client.get_hoa_name_by_bag_id.return_value = "Test HOA"
+
+        # Create an existing HOA
+        existing_hoa = HomeownerAssociation.objects.create(
+            name="Test HOA", build_year=2000, number_of_appartments=10
+        )
+
+        # Call the method
+        result = HomeownerAssociation.get_or_create_hoa_by_bag_id("some_bag_id")
+
+        # Assert the existing HOA is returned
+        self.assertEqual(result, existing_hoa)
+        mock_client.get_hoa_name_by_bag_id.assert_called_once_with("some_bag_id")
+        mock_client.get_hoa_by_name.assert_not_called()
+
+    @patch("apps.homeownerassociation.models.DsoClient")
+    def test_get_or_create_hoa_by_bag_id_new_hoa(self, MockDsoClient):
+        # Mock the DsoClient and its methods
+        mock_client = MockDsoClient.return_value
+        mock_client.get_hoa_name_by_bag_id.return_value = "New HOA"
+        mock_client.get_hoa_by_name.return_value = [
+            {"pndOorspronkelijkBouwjaar": 2010, "votIdentificatie": "123"},
+            {"pndOorspronkelijkBouwjaar": 2010, "votIdentificatie": "123"},
+        ]
+
+        # Call the method
+        result = HomeownerAssociation.get_or_create_hoa_by_bag_id("some_bag_id")
+        # Assert a new HOA is created
+        self.assertIsInstance(result, HomeownerAssociation)
+        self.assertEqual(result.name, "New HOA")
+        self.assertEqual(result.build_year, 2010)
+        self.assertEqual(result.number_of_appartments, 1)
+        mock_client.get_hoa_name_by_bag_id.assert_called_once_with("some_bag_id")
+        mock_client.get_hoa_by_name.assert_called_once_with("New HOA")
+
+    # write a test to verify that duplicate appartements are only counted once
+    @patch("apps.homeownerassociation.models.DsoClient")
+    def test_get_or_create_hoa_by_bag_id_new_hoa_duplicate_appartments(
+        self, MockDsoClient
+    ):
+        mock_client = MockDsoClient.return_value
+        mock_client.get_hoa_name_by_bag_id.return_value = "HOA"
+        mock_client.get_hoa_by_name.return_value = [
+            {"pndOorspronkelijkBouwjaar": 2010, "votIdentificatie": "333"},
+            {"pndOorspronkelijkBouwjaar": 2010, "votIdentificatie": "333"},
+            {"pndOorspronkelijkBouwjaar": 2010, "votIdentificatie": "444"},
+        ]
+        result = HomeownerAssociation.get_or_create_hoa_by_bag_id("unique_id")
+        self.assertIsInstance(result, HomeownerAssociation)
+        self.assertEqual(result.name, "HOA")
+        self.assertEqual(result.build_year, 2010)
+        self.assertEqual(result.number_of_appartments, 2)

app/clients/dso_client.py

@@ -0,0 +1,50 @@
+import requests
+from django.conf import settings
+
+from utils.exceptions import NotFoundException
+
+
+class DsoClient:
+    def __init__(self):
+        self.headers = {"Authorization": f"Bearer {self._get_access_token()}"}
+
+    def get_hoa_name_by_bag_id(self, bag_id):
+        url = f"{settings.DSO_API_URL}?brkVveIsEigendomVve=ja&votIdentificatie={bag_id}"
+        response = requests.get(url, headers=self.headers)
+        response_data = response.json()
+        wonen_verblijfsobject_list = response_data.get("_embedded", {}).get(
+            "wonen_verblijfsobject", []
+        )
+        if wonen_verblijfsobject_list:
+            hoa = wonen_verblijfsobject_list[0]
+            return hoa["brkVveStatutaireNaam"]
+        raise NotFoundException(f"HomeownerAssociation with bag ID {bag_id} not found.")
+
+    def get_hoa_by_name(self, hoa_name):
+        url = f"{settings.DSO_API_URL}?brkVveStatutaireNaam={hoa_name}&_pageSize=300"
+        hoa_json = self._get_paginated_response(url)
+        return hoa_json["_embedded"]["wonen_verblijfsobject"]
+
+    def _get_paginated_response(self, url):
+        response = requests.get(url, headers=self.headers)
+        response_json = response.json()
+        while "_links" in response_json and "next" in response_json["_links"]:
+            next_page = response_json["_links"]["next"]["href"]
+            paged_response = requests.get(next_page, headers=self.headers)
+            paged_json = paged_response.json()
+            response_json["_embedded"]["wonen_verblijfsobject"].extend(
+                paged_json["_embedded"]["wonen_verblijfsobject"]
+            )
+            response_json["_links"] = paged_json["_links"]
+        return response_json
+
+    def _get_access_token(self):
+        url = settings.DSO_AUTH_URL
+        payload = {
+            "client_id": settings.DSO_CLIENT_ID,
+            "grant_type": "client_credentials",
+            "client_secret": settings.DSO_CLIENT_SECRET,
+            "scope": "openid email",
+        }
+        response = requests.post(url, data=payload).json()
+        return response["access_token"]

app/config/settings.py

@@ -45,6 +45,8 @@
     "apps.cases",
     "apps.workflow",
     "apps.events",
+    "apps.homeownerassociation",
+    "apps.address",
     "django_spaghetti",
     "drf_spectacular",
     "django_celery_results",
@@ -71,6 +73,7 @@
         "apps.users.auth.AuthenticationClass",
     ],
     "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
+    "EXCEPTION_HANDLER": "utils.exceptions.custom_exception_handler",
 }
 
 CORS_ALLOWED_ORIGINS = os.environ.get("CORS_ALLOWED_ORIGINS", ("http://default")).split(
@@ -249,3 +252,8 @@ def get_redis_url():
         },
     },
 }
+
+DSO_CLIENT_ID = os.getenv("DSO_CLIENT_ID", "default_client_id")
+DSO_CLIENT_SECRET = os.getenv("DSO_CLIENT_SECRET", "default_client_secret")
+DSO_AUTH_URL = os.getenv("DSO_AUTH_URL", "https://default.auth.url")
+DSO_API_URL = os.getenv("DSO_API_URL", "https://default.api.url")