Merge branch 'cmd-command' into 'main'

add a management command for topics

See merge request hcdc/django/dasf-broker-django!5

dasf_broker/management/commands/dasf_topic.py

+"""Management command to edit or create a Broker topic"""
+# Disclaimer
+# ----------
+#
+# Copyright (C) 2022 Helmholtz-Zentrum Hereon
+#
+# This file is part of dasf-broker-django and is released under the
+# EUPL-1.2 license.
+# See LICENSE in the root of the repository for full licensing details.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the EUROPEAN UNION PUBLIC LICENCE v. 1.2 or later
+# as published by the European Commission.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# EUPL-1.2 license for more details.
+#
+# You should have received a copy of the EUPL-1.2 license along with this
+# program. If not, see https://www.eupl.eu/.
+
+import argparse
+from typing import Optional
+
+from django.core.management.base import BaseCommand
+from guardian.shortcuts import assign_perm, get_anonymous_user, remove_perm
+
+
+class Command(BaseCommand):
+    """Django command to migrate the database."""
+
+    help = "Create or update a broker topic."
+
+    def add_arguments(self, parser):
+        """Add connection arguments to the parser."""
+
+        parser.add_argument("topic_slug", help="The slug for the topic")
+
+        parser.add_argument(
+            "-n", "--new", help="Create a new topic.", action="store_true"
+        )
+
+        parser.add_argument(
+            "--public",
+            action=argparse.BooleanOptionalAction,
+            help="Make the topic public.",
+            default=None,
+        )
+
+        parser.add_argument(
+            "--anonymous",
+            action="store_true",
+            help=(
+                "Make the topic available for anonymous consumers and "
+                "producers. This option involves '--public' and "
+                "'-c AnonymousUser'."
+            ),
+        )
+
+        parser.add_argument(
+            "-c",
+            "--consumer",
+            help=(
+                "Username of the user account that you want to make a "
+                "consumer of the topic"
+            ),
+        )
+
+        parser.add_argument(
+            "-p",
+            "--producer",
+            help=(
+                "Username of the user account that you want to make a "
+                "producer of the topic"
+            ),
+        )
+
+        parser.add_argument(
+            "--create-consumer",
+            help=(
+                "Create a new user account for the consumer with the given "
+                "username"
+            ),
+            action="store_true",
+        )
+
+        parser.add_argument(
+            "--create-producer",
+            help=(
+                "Create a new user account for the producer with the given "
+                "username"
+            ),
+            action="store_true",
+        )
+
+        parser.add_argument(
+            "-rm",
+            "--remove-permissions",
+            help=(
+                "Remove the permissions for the given consumer and/or "
+                "producer instead of assigning."
+            ),
+            action="store_true",
+        )
+
+        parser.add_argument(
+            "-i",
+            "--show-info",
+            action="store_true",
+            help=(
+                "Show information on the topic (including a list of the user "
+                "names of consumers and producers and whether the topic is "
+                "public or not. The updates to the topic are done before "
+                "printing out the info."
+            ),
+        )
+
+        parser.add_argument(
+            "-db",
+            "--database",
+            help=(
+                "The Django database identifier (see settings.py), "
+                "default: %(default)s"
+            ),
+            default="default",
+        )
+
+    def handle(
+        self,
+        topic_slug: str,
+        *args,
+        new: bool = False,
+        public: Optional[bool] = None,
+        anonymous: bool = False,
+        consumer: Optional[str] = None,
+        producer: Optional[str] = None,
+        create_consumer: bool = False,
+        create_producer: bool = False,
+        remove_permissions: bool = False,
+        show_info: bool = False,
+        database: str = "default",
+        **options,
+    ):
+        """Migrate the database."""
+
+        from django.contrib.auth import get_user_model
+        from django.db import IntegrityError
+
+        from dasf_broker.models import BrokerTopic
+
+        User = get_user_model()  # type: ignore  # noqa: F811
+
+        if new:
+            try:
+                topic = BrokerTopic.objects.using(database).create(
+                    slug=topic_slug
+                )
+            except IntegrityError:
+                raise ValueError(
+                    f"A topic with the slug {topic_slug} already exists."
+                )
+        else:
+            try:
+                topic = BrokerTopic.objects.using(database).get(
+                    slug=topic_slug
+                )
+            except BrokerTopic.DoesNotExist:
+
+                raise ValueError(
+                    f"A topic with the slug {topic_slug} does not exist. "
+                    "If you want to create it, please use "
+                    "'--new' option instead of '--update'."
+                )
+
+        if anonymous:
+            topic.is_public = True
+            assign_perm("can_consume", get_anonymous_user(), topic)
+            topic.save()
+
+        if public is not None:
+            topic.is_public = public
+            topic.save()
+
+        if consumer is not None:
+            if create_consumer:
+                try:
+                    user = User.objects.create(username=consumer)
+                except IntegrityError:
+                    raise ValueError(
+                        f"A user with the username {consumer} already exists!"
+                    )
+            else:
+                try:
+                    user = User.objects.get(username=consumer)
+                except User.DoesNotExist:
+                    raise ValueError(
+                        f"A user with the username {consumer} does not exist. "
+                        "If you want to create it, please add the "
+                        "'--create-consumer' option."
+                    )
+            if remove_permissions:
+                remove_perm("can_consume", user, topic)
+            else:
+                assign_perm("can_consume", user, topic)
+
+        if producer is not None:
+            create_producer = create_producer and not (
+                create_consumer and consumer == producer
+            )
+            if create_producer:
+                try:
+                    user = User.objects.create(username=producer)
+                except IntegrityError:
+                    raise ValueError(
+                        f"A user with the username {producer} already exists!"
+                    )
+            else:
+                try:
+                    user = User.objects.get(username=producer)
+                except User.DoesNotExist:
+                    raise ValueError(
+                        f"A user with the username {producer} does not exist. "
+                        "If you want to create it, please add the "
+                        "'--create-producer' option."
+                    )
+            if remove_permissions:
+                remove_perm("can_produce", user, topic)
+            else:
+                assign_perm("can_produce", user, topic)
+
+        if show_info:
+            print("Consumers:")
+            for user in topic.consumers:
+                print(f"    {user}")
+            print("public:")
+            print(f"    {topic.is_public}")
+            print("Producers:")
+            for user in topic.producers:
+                print(f"    {user}")