✨ [#564] add command to import objecttypes from api

docs/manual/index.rst

@@ -8,3 +8,4 @@ Manual
    :caption: Further reading
 
    scripts
+   migration

docs/manual/migration.rst

@@ -0,0 +1,19 @@
+.. _objecttype_migration:
+
+ObjectTypes API migration
+=========================
+
+In version 4.0.0 the ObjectTypes API will be merged into the Objects API so that only one application is needed. This means that from version 4.0.0 only objecttypes that exist in the database are supported, and no external objecttypes can be used.
+
+Importing objecttype data
+-------------------------
+Before updating to 4.0.0 all objecttypes from the ObjectTypes API instance need to be imported. This can be done with the ``import_objecttypes`` command that can be executed from the Objects container.
+This command will fetch all objecttypes and their versions from an objecttype service based on its identifier/slug (which can be found in the admin interface under ``Configuration > Services``)
+and update existing objecttypes or create new ones if they have not been added to the objecttypes API.
+
+.. code-block:: bash
+
+    src/manage.py import_objecttypes objecttypes-api
+
+Please note that after the update the objecttypes API is still being used in Objects API version <4.0.0, the command only fetches and imports the data.
+From 4.0.0 onwards it will use the imported objecttypes.

src/objects/core/constants.py

@@ -0,0 +1,25 @@
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+
+class ObjectVersionStatus(models.TextChoices):
+    published = "published", _("Published")
+    draft = "draft", _("Draft")
+    deprecated = "deprecated", _("Deprecated")
+
+
+class DataClassificationChoices(models.TextChoices):
+    open = "open", _("Open")
+    intern = "intern", _("Intern")
+    confidential = "confidential", _("Confidential")
+    strictly_confidential = "strictly_confidential", _("Strictly confidential")
+
+
+class UpdateFrequencyChoices(models.TextChoices):
+    real_time = "real_time", _("Real-time")
+    hourly = "hourly", _("Hourly")
+    daily = "daily", _("Daily")
+    weekly = "weekly", _("Weekly")
+    monthly = "monthly", _("Monthly")
+    yearly = "yearly", _("Yearly")
+    unknown = "unknown", _("Unknown")

src/objects/core/management/commands/import_objecttypes.py

@@ -0,0 +1,145 @@
+from typing import Any
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import transaction
+from django.utils.translation import gettext as _
+
+from djangorestframework_camel_case.util import underscoreize
+from packaging.version import Version
+from requests.exceptions import RequestException
+from zgw_consumers.models import Service
+
+from objects.core.models import ObjectType, ObjectTypeVersion
+from objects.utils.client import get_objecttypes_client
+
+MIN_OBJECTTYPES_VERSION = "3.4.0"  # added boolean field linkable_to_zaken to ObjectType
+
+
+class Command(BaseCommand):
+    help = "Import ObjectTypes & ObjectTypeVersions from an Objecttypes API based on the service identifier."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "service_slug",
+            help=_("Identifier/slug of Objecttypes API service"),
+        )
+
+    @transaction.atomic
+    def handle(self, *args, **options):
+        service_slug = options["service_slug"]
+        service = self._get_service(service_slug)
+
+        with get_objecttypes_client(service) as client:
+            try:
+                self._check_objecttypes_api_version(client)
+
+                objecttypes = client.list_objecttypes()
+                data = self._parse_objecttype_data(objecttypes, service)
+                self._bulk_create_or_update_objecttypes(data)
+                self.stdout.write("Successfully imported %s objecttypes" % len(data))
+
+                for objecttype in data:
+                    objecttype_versions = client.list_objecttype_versions(
+                        objecttype.uuid
+                    )
+                    data = self._parse_objectversion_data(
+                        objecttype_versions, objecttype
+                    )
+                    self._bulk_create_or_update_objecttype_versions(data)
+                    self.stdout.write(
+                        "Successfully imported %s versions for type: %s"
+                        % (len(data), objecttype.name)
+                    )
+
+            except RequestException as e:
+                raise CommandError(
+                    _(
+                        "Something went wrong while making requests to Objecttypes API: {}"
+                    ).format(e)
+                )
+
+    def _get_service(self, slug):
+        try:
+            return Service.objects.get(slug=slug)
+        except Service.DoesNotExist:
+            raise CommandError(_("Service '{}' does not exist").format(slug))
+
+    def _check_objecttypes_api_version(self, client):
+        api_version = client.get_objecttypes_api_version()
+        if api_version is None or Version(
+            client.get_objecttypes_api_version()
+        ) < Version(MIN_OBJECTTYPES_VERSION):
+            raise CommandError(
+                _("Object types API version must be {} or higher.").format(
+                    MIN_OBJECTTYPES_VERSION
+                )
+            )
+
+    def _bulk_create_or_update_objecttypes(self, data):
+        ObjectType.objects.bulk_create(
+            data,
+            update_conflicts=True,  # Updates existing Objecttypes based on unique_fields
+            unique_fields=[
+                "uuid",
+                "service",
+            ],  # TODO remove service from unique_fields after objecttype migration since it will no longer be part of the ObjectType model.
+            update_fields=[
+                "is_imported",
+                "name",
+                "name_plural",
+                "description",
+                "data_classification",
+                "maintainer_organization",
+                "maintainer_department",
+                "contact_person",
+                "contact_email",
+                "source",
+                "update_frequency",
+                "provider_organization",
+                "documentation_url",
+                "labels",
+                "created_at",
+                "modified_at",
+                "allow_geometry",
+                "linkable_to_zaken",
+            ],
+        )
+
+    def _bulk_create_or_update_objecttype_versions(self, data):
+        ObjectTypeVersion.objects.bulk_create(
+            data,
+            ignore_conflicts=True,
+            unique_fields=[
+                "object_type",
+                "version",
+            ],
+            update_fields=[
+                "created_at",
+                "modified_at",
+                "published_at",
+                "json_schema",
+                "status",
+            ],
+        )
+
+    def _parse_objecttype_data(
+        self, objecttypes: list[dict[str, Any]], service: Service
+    ) -> list[ObjectType]:
+        data = []
+        for objecttype in objecttypes:
+            objecttype.pop("versions")
+            objecttype.pop("url")
+            objecttype["service"] = service
+            objecttype["is_imported"] = True
+            data.append(ObjectType(**underscoreize(objecttype)))
+        return data
+
+    def _parse_objectversion_data(
+        self, objecttype_versions: list[dict[str, Any]], objecttype
+    ) -> list[ObjectTypeVersion]:
+        data = []
+        for objecttype_version in objecttype_versions:
+            objecttype_version.pop("url")
+            objecttype_version["objectType"] = objecttype
+            data.append(ObjectTypeVersion(**underscoreize(objecttype_version)))
+        return data

src/objects/core/migrations/0035_objecttype_allow_geometry_objecttype_contact_email_and_more.py

@@ -0,0 +1,115 @@
+# Generated by Django 5.2.7 on 2025-12-04 11:34
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0034_alter_objectrecord__object_type_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='objecttype',
+            name='allow_geometry',
+            field=models.BooleanField(default=True, help_text="Shows whether the related objects can have geographic coordinates. If the value is 'false' the related objects are not allowed to have coordinates and the creation/update of objects with `geometry` property will raise an error ", verbose_name='allow geometry'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='contact_email',
+            field=models.CharField(blank=True, help_text='Email of the person in the organization who can provide information about the object type', max_length=200, verbose_name='contact email'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='contact_person',
+            field=models.CharField(blank=True, help_text='Name of the person in the organization who can provide information about the object type', max_length=200, verbose_name='contact person'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='created_at',
+            field=models.DateField(blank=True, help_text='Date when the object type was created', null=True, verbose_name='created at'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='data_classification',
+            field=models.CharField(choices=[('open', 'Open'), ('intern', 'Intern'), ('confidential', 'Confidential'), ('strictly_confidential', 'Strictly confidential')], default='open', help_text='Confidential level of the object type', max_length=50, verbose_name='data classification'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='description',
+            field=models.CharField(blank=True, help_text='The description of the object type', max_length=1000, verbose_name='description'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='documentation_url',
+            field=models.URLField(blank=True, help_text='Link to the documentation for the object type', verbose_name='documentation url'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='labels',
+            field=models.JSONField(blank=True, default=dict, help_text='Key-value pairs of keywords related for the object type', verbose_name='labels'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='linkable_to_zaken',
+            field=models.BooleanField(default=False, help_text='Objects of this type can have a link to 1 or more Zaken.\nTrue indicates the lifetime of the object is linked to the lifetime of linked zaken, i.e., when all linked Zaken to an object are archived/destroyed, the object will also be archived/destroyed.', verbose_name='linkable to zaken'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='maintainer_department',
+            field=models.CharField(blank=True, help_text='Business department which is responsible for the object type', max_length=200, verbose_name='maintainer department'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='maintainer_organization',
+            field=models.CharField(blank=True, help_text='Organization which is responsible for the object type', max_length=200, verbose_name='maintainer organization'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='modified_at',
+            field=models.DateField(blank=True, help_text='Last date when the object type was modified', null=True, verbose_name='modified at'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='name',
+            field=models.CharField(blank=True, help_text='Name of the object type', max_length=100, verbose_name='name'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='name_plural',
+            field=models.CharField(blank=True, help_text='Plural name of the object type', max_length=100, verbose_name='name plural'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='provider_organization',
+            field=models.CharField(blank=True, help_text='Organization which is responsible for publication of the object type', max_length=200, verbose_name='provider organization'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='source',
+            field=models.CharField(blank=True, help_text='Name of the system from which the object type originates', max_length=200, verbose_name='source'),
+        ),
+        migrations.AddField(
+            model_name='objecttype',
+            name='update_frequency',
+            field=models.CharField(choices=[('real_time', 'Real-time'), ('hourly', 'Hourly'), ('daily', 'Daily'), ('weekly', 'Weekly'), ('monthly', 'Monthly'), ('yearly', 'Yearly'), ('unknown', 'Unknown')], default='unknown', help_text='Indicates how often the object type is updated', max_length=10, verbose_name='update frequency'),
+        ),
+        migrations.CreateModel(
+            name='ObjectTypeVersion',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('version', models.PositiveSmallIntegerField(help_text='Integer version of the OBJECTTYPE', verbose_name='version')),
+                ('created_at', models.DateField(blank=True, help_text='Date when the version was created', null=True, verbose_name='created at')),
+                ('modified_at', models.DateField(blank=True, help_text='Last date when the version was modified', null=True, verbose_name='modified at')),
+                ('published_at', models.DateField(blank=True, help_text='Date when the version was published', null=True, verbose_name='published_at')),
+                ('json_schema', models.JSONField(default=dict, help_text='JSON schema for Object validation', verbose_name='JSON schema')),
+                ('status', models.CharField(choices=[('published', 'Published'), ('draft', 'Draft'), ('deprecated', 'Deprecated')], default='draft', help_text='Status of the object type version', max_length=20, verbose_name='status')),
+                ('object_type', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='versions', to='core.objecttype')),
+            ],
+            options={
+                'unique_together': {('object_type', 'version')},
+            },
+        ),
+    ]

src/objects/core/migrations/0036_objecttype_is_imported.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.7 on 2025-12-05 13:54
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0035_objecttype_allow_geometry_objecttype_contact_email_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='objecttype',
+            name='is_imported',
+            field=models.BooleanField(default=False, editable=False, verbose_name='Is imported'),
+        ),
+    ]