Add load_upl command

src/open_producten/products/tests/api/test_product.py

@@ -29,9 +29,9 @@ def product_to_dict(product):
         product.product_type,
         exclude=("categories", "conditions", "tags", "related_product_types"),
     )
-    product_dict["product_type"]["uniform_product_name"] = model_to_dict_with_id(
-        product.product_type.uniform_product_name
-    )
+    product_dict["product_type"][
+        "uniform_product_name"
+    ] = product.product_type.uniform_product_name.uri
 
     product_dict["product_type"]["created_on"] = str(
         product.product_type.created_on.astimezone().isoformat()

src/open_producten/producttypes/admin/condition.py

@@ -18,4 +18,4 @@ def display_product_types(self, obj):
         return ", ".join(p.name for p in obj.product_types.all())
 
     def get_queryset(self, request):
-        return super().get_queryset(request).prefetch_related("product_type")
+        return super().get_queryset(request).prefetch_related("product_types")

src/open_producten/producttypes/admin/upn.py

@@ -5,4 +5,15 @@
 
 @admin.register(UniformProductName)
 class UniformProductNameAdmin(admin.ModelAdmin):
-    list_display = ("name", "url")
+    list_display = ("name", "uri", "is_deleted")
+    list_filter = ("is_deleted",)
+    search_fields = ("name", "uri")
+
+    def has_change_permission(self, request, obj=None):
+        return False
+
+    def has_add_permission(self, request):
+        return False
+
+    def has_delete_permission(self, request, obj=None):
+        return False

src/open_producten/producttypes/management/commands/load_upl.py

@@ -0,0 +1,9 @@
+from ..parsers import ParserCommand
+from ..utils import load_upn
+
+
+class Command(ParserCommand):
+    plural_object_name = "upn"
+
+    def handle(self, **options):
+        super().handle(load_upn, **options)

src/open_producten/producttypes/management/parsers.py

@@ -0,0 +1,103 @@
+import csv
+import logging
+import os
+from tempfile import NamedTemporaryFile
+from typing import Any, Dict, List
+
+from django.core.management import BaseCommand, CommandError
+
+import requests
+from lxml import etree
+
+logger = logging.getLogger(__name__)
+
+
+class ParserException(Exception):
+    pass
+
+
+class OwmsParser:
+    available_parsers = {"csv", "xml"}
+
+    def __init__(self, xml_column_names=None):
+        self.xml_column_names = xml_column_names
+
+    def parse(self, filename: str):
+        """Call parsing function based on filename extension."""
+
+        _, extension = os.path.splitext(filename)
+        file_format = extension[1:]
+
+        if not filename.startswith("http"):
+            return self.process_file(filename, file_format)
+
+        with NamedTemporaryFile() as f:
+            f.write(requests.get(filename).content)
+            f.seek(0)
+            return self.process_file(f.name, file_format)
+
+    def process_file(self, filename, file_format):
+        """Process a given file, calling a specific method based on the given format."""
+        if file_format in self.available_parsers:
+            return getattr(self, file_format)(filename)
+        else:
+            raise ParserException("File format does not exist")
+
+    def xml(self, filename: str) -> List[Dict[str, Any]]:
+        if not self.xml_column_names:
+            raise ParserException("Invalid XML column names")
+
+        tree = etree.parse(filename)
+        values = tree.findall("value")
+        return [
+            {
+                column: getattr(value.find(column), "text", None)
+                for column in self.xml_column_names
+            }
+            for value in values
+        ]
+
+    def csv(self, filename: str) -> List[Dict[str, Any]]:
+        with open(filename, encoding="utf-8-sig") as f:
+            data = csv.DictReader(f)
+            return list(data)
+
+
+class ParserCommand(BaseCommand):
+    plural_object_name = "objects"
+    xml_column_names = None
+
+    def __init__(self):
+        self.help = (
+            f"Load {self.plural_object_name} to the database from a given XML/CSV file."
+        )
+        self.parser = OwmsParser(xml_column_names=self.xml_column_names)
+        super().__init__()
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "filename",
+            help="The name of the file to be imported.",
+        )
+
+    def handle(self, handler_function, **options):
+        filename = options.pop("filename")
+
+        self.stdout.write(f"Importing {self.plural_object_name} from {filename}...")
+
+        try:
+            data = self.parse(filename)
+            created_count = handler_function(data)
+        except Exception as e:
+            self.stderr.write(f"Error: {e}")
+            logger.exception(e)
+        else:
+            self.stdout.write(self.style.SUCCESS(f"Done ({created_count} objects)."))
+
+    def parse(self, filename) -> List[Dict[str, Any]]:
+        try:
+            return self.parser.parse(filename)
+        except ParserException:
+            raise CommandError(
+                f"No parser available for that format. Available: {', '.join(self.parser.available_parsers)}"
+            )

src/open_producten/producttypes/management/utils.py

@@ -0,0 +1,29 @@
+import logging
+from typing import Any, Dict, List
+
+from open_producten.producttypes.models import UniformProductName
+
+logger = logging.getLogger(__name__)
+
+
+def load_upn(data: List[Dict[str, Any]]) -> int:
+    """
+    Loads UPNs based on a list of dictionaries.
+    """
+    count = 0
+    upn_updated_list = []
+
+    for obj in data:
+        upn, created = UniformProductName.objects.update_or_create(
+            uri=obj.get("URI"),
+            defaults={"name": obj.get("UniformeProductnaam"), "is_deleted": False},
+        )
+        upn_updated_list.append(upn.id)
+
+        if created:
+            count += 1
+
+    UniformProductName.objects.exclude(id__in=upn_updated_list).update(
+        is_deleted=True,
+    )
+    return count

src/open_producten/producttypes/migrations/0006_remove_uniformproductname_url_and_more.py

@@ -0,0 +1,45 @@
+# Generated by Django 4.2.13 on 2024-09-27 13:48
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("producttypes", "0005_producttype_contacts_producttype_locations_and_more"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="uniformproductname",
+            name="url",
+        ),
+        migrations.AddField(
+            model_name="uniformproductname",
+            name="is_deleted",
+            field=models.BooleanField(
+                default=False,
+                help_text="defines if the upn definition is deleted.",
+                verbose_name="Is deleted",
+            ),
+        ),
+        migrations.AddField(
+            model_name="uniformproductname",
+            name="uri",
+            field=models.URLField(
+                help_text="Uri to the upn definition.",
+                unique=True,
+                verbose_name="Uri",
+            ),
+            preserve_default=False,
+        ),
+        migrations.AlterField(
+            model_name="uniformproductname",
+            name="name",
+            field=models.CharField(
+                help_text="Uniform product name",
+                max_length=100,
+                unique=True,
+                verbose_name="Name",
+            ),
+        ),
+    ]

src/open_producten/producttypes/models/upn.py

@@ -6,14 +6,21 @@
 
 class UniformProductName(BaseModel):
     name = models.CharField(
-        verbose_name=_("Name"), max_length=100, help_text=_("Uniform product name")
+        verbose_name=_("Name"),
+        max_length=100,
+        help_text=_("Uniform product name"),
+        unique=True,
     )
 
-    url = models.URLField(
-        verbose_name=_("Url"),
-        blank=True,
-        default="",
-        help_text=_("Url to the upn definition."),
+    uri = models.URLField(
+        verbose_name=_("Uri"),
+        help_text=_("Uri to the upn definition."),
+        unique=True,
+    )
+    is_deleted = models.BooleanField(
+        _("Is deleted"),
+        help_text=_("defines if the upn definition is deleted."),
+        default=False,
     )
 
     class Meta:

src/open_producten/producttypes/serializers/category.py

@@ -3,14 +3,16 @@
 
 from rest_framework import serializers
 
-from open_producten.producttypes.models import Category, ProductType
+from open_producten.producttypes.models import Category, ProductType, UniformProductName
 from open_producten.utils.serializers import build_array_duplicates_error_message
 
-from .children import QuestionSerializer, UniformProductNameSerializer
+from .children import QuestionSerializer
 
 
 class SimpleProductTypeSerializer(serializers.ModelSerializer):
-    uniform_product_name = UniformProductNameSerializer()
+    uniform_product_name = serializers.SlugRelatedField(
+        slug_field="uri", queryset=UniformProductName.objects.all()
+    )
 
     class Meta:
         model = ProductType

src/open_producten/producttypes/serializers/children.py

@@ -14,7 +14,6 @@
     Question,
     Tag,
     TagType,
-    UniformProductName,
 )
 
 
@@ -139,12 +138,6 @@ class Meta:
         fields = "__all__"
 
 
-class UniformProductNameSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = UniformProductName
-        fields = "__all__"
-
-
 class QuestionSerializer(serializers.ModelSerializer):
     class Meta:
         model = Question

src/open_producten/producttypes/serializers/producttype.py

@@ -13,7 +13,6 @@
     PriceSerializer,
     QuestionSerializer,
     TagSerializer,
-    UniformProductNameSerializer,
 )
 
 
@@ -37,14 +36,10 @@ class ProductTypeSerializer(serializers.ModelSerializer):
         many=True, queryset=ProductType.objects.all(), default=[]
     )
 
-    uniform_product_name_id = serializers.PrimaryKeyRelatedField(
-        queryset=UniformProductName.objects.all(),
-        write_only=True,
-        source="uniform_product_name",
+    uniform_product_name = serializers.SlugRelatedField(
+        slug_field="uri", queryset=UniformProductName.objects.all()
     )
 
-    uniform_product_name = UniformProductNameSerializer(read_only=True)
-
     conditions = ConditionSerializer(many=True, read_only=True)
     condition_ids = serializers.PrimaryKeyRelatedField(
         many=True,
@@ -133,7 +128,7 @@ def update(self, instance, validated_data):
 
 
 class ProductTypeCurrentPriceSerializer(serializers.ModelSerializer):
-    upl_uri = serializers.ReadOnlyField(source="uniform_product_name.url")
+    upl_uri = serializers.ReadOnlyField(source="uniform_product_name.uri")
     upl_name = serializers.ReadOnlyField(source="uniform_product_name.name")
     current_price = PriceSerializer(allow_null=True)
 

src/open_producten/producttypes/tests/api/test_product_type_price.py

@@ -354,7 +354,7 @@ def test_get_current_prices_when_product_type_has_no_prices(self):
                     "id": str(self.product_type.id),
                     "name": self.product_type.name,
                     "upl_name": self.product_type.uniform_product_name.name,
-                    "upl_uri": self.product_type.uniform_product_name.url,
+                    "upl_uri": self.product_type.uniform_product_name.uri,
                     "current_price": None,
                 },
             ],
@@ -375,7 +375,7 @@ def test_get_current_prices_when_product_type_only_has_price_in_future(self):
                     "id": str(self.product_type.id),
                     "name": self.product_type.name,
                     "upl_name": self.product_type.uniform_product_name.name,
-                    "upl_uri": self.product_type.uniform_product_name.url,
+                    "upl_uri": self.product_type.uniform_product_name.uri,
                     "current_price": None,
                 },
             ],
@@ -399,7 +399,7 @@ def test_get_current_prices_when_product_type_has_current_price(self):
                     "id": str(self.product_type.id),
                     "name": self.product_type.name,
                     "upl_name": self.product_type.uniform_product_name.name,
-                    "upl_uri": self.product_type.uniform_product_name.url,
+                    "upl_uri": self.product_type.uniform_product_name.uri,
                     "current_price": {
                         "id": str(price.id),
                         "valid_from": "2024-01-01",

src/open_producten/producttypes/tests/api/test_producttype.py

@@ -51,9 +51,7 @@ def product_type_to_dict(product_type):
     product_type_dict["updated_on"] = str(
         product_type.updated_on.astimezone().isoformat()
     )
-    product_type_dict["uniform_product_name"] = model_to_dict_with_id(
-        product_type.uniform_product_name
-    )
+    product_type_dict["uniform_product_name"] = product_type.uniform_product_name.uri
     product_type_dict["icon"] = None
     product_type_dict["image"] = None
     return product_type_dict
@@ -70,7 +68,7 @@ def setUp(self):
             "name": "test-product-type",
             "summary": "test",
             "content": "test test",
-            "uniform_product_name_id": upn.id,
+            "uniform_product_name": upn.uri,
             "category_ids": [category.id],
         }
 

src/open_producten/producttypes/tests/factories.py

@@ -18,7 +18,7 @@
 
 class UniformProductNameFactory(factory.django.DjangoModelFactory):
     name = factory.Sequence(lambda n: f"upn {n}")
-    url = factory.Faker("url")
+    uri = factory.Faker("url")
 
     class Meta:
         model = UniformProductName