serverless: add copy_objects and create_objects

src/ansys/dynamicreporting/core/exceptions.py

@@ -120,3 +120,9 @@ class MultipleObjectsReturnedError(ADRException):
     """Exception raised if only one object was expected, but multiple were returned."""
 
     detail = "get() returned more than one object."
+
+
+class IntegrityError(ADRException):
+    """Exception raised if there is a constraint violation while saving an object in the database."""
+
+    detail = "A database integrity check failed."

src/ansys/dynamicreporting/core/serverless/adr.py

@@ -1,4 +1,4 @@
-# from collections.abc import Iterable
+from collections.abc import Iterable
 import os
 from pathlib import Path
 import platform
@@ -8,8 +8,6 @@
 
 import django
 from django.core import management
-
-# from django.db import IntegrityError, connection, connections
 from django.http import HttpRequest
 
 from .. import DEFAULT_ANSYS_VERSION
@@ -120,12 +118,29 @@ def _get_install_directory(self, ansys_installation: str) -> Path:
         raise InvalidAnsysPath(f"Unable to detect an installation in: {','.join(dirs_to_check)}")
 
     def _check_dir(self, dir_):
-        dir_path = Path(dir_)
+        dir_path = Path(dir_) if not isinstance(dir_, Path) else dir_
         if not dir_path.is_dir():
             self._logger.error(f"Invalid directory path: {dir_}")
             raise InvalidPath(extra_detail=dir_)
         return dir_path
 
+    def _migrate_db(self, db):
+        try:  # upgrade databases
+            management.call_command("migrate", "--no-input", "--database", db, verbosity=0)
+        except Exception as e:
+            self._logger.error(f"{e}")
+            raise DatabaseMigrationError(extra_detail=str(e))
+        else:
+            from django.contrib.auth.models import Group, Permission, User
+
+            if not User.objects.using(db).filter(is_superuser=True).exists():
+                user = User.objects.using(db).create_superuser("nexus", "", "cei")
+                # include the nexus group (with all permissions)
+                nexus_group, created = Group.objects.using(db).get_or_create(name="nexus")
+                if created:
+                    nexus_group.permissions.set(Permission.objects.using(db).all())
+                nexus_group.user_set.add(user)
+
     def setup(self, collect_static: bool = False) -> None:
         from django.conf import settings
 
@@ -195,22 +210,11 @@ def setup(self, collect_static: bool = False) -> None:
             self._dataset = Dataset.create()
 
         # migrations
-        if self._db_directory is not None:
-            try:  # upgrades all databases
-                management.call_command("migrate", "--no-input", verbosity=0)
-            except Exception as e:
-                self._logger.error(f"{e}")
-                raise DatabaseMigrationError(extra_detail=str(e))
-            else:
-                from django.contrib.auth.models import Group, Permission, User
-
-                if not User.objects.filter(is_superuser=True).exists():
-                    user = User.objects.create_superuser("nexus", "", "cei")
-                    # include the nexus group (with all permissions)
-                    nexus_group, created = Group.objects.get_or_create(name="nexus")
-                    if created:
-                        nexus_group.permissions.set(Permission.objects.all())
-                    nexus_group.user_set.add(user)
+        if self._databases:
+            for db in self._databases:
+                self._migrate_db(db)
+        elif self._db_directory is not None:
+            self._migrate_db("default")
 
         # collectstatic
         if collect_static and self._static_directory is not None:
@@ -311,8 +315,98 @@ def query(
         self,
         query_type: Union[Session, Dataset, Type[Item], Type[Template]],
         query: str = "",
+        **kwargs: Any,
     ) -> ObjectSet:
         if not issubclass(query_type, (Item, Template, Session, Dataset)):
             self._logger.error(f"{query_type} is not valid")
             raise TypeError(f"{query_type} is not valid")
-        return query_type.find(query=query)
+        return query_type.find(query=query, **kwargs)
+
+    @staticmethod
+    def create_objects(
+        objects: Union[list, ObjectSet],
+        **kwargs: Any,
+    ) -> int:
+        if not isinstance(objects, Iterable):
+            raise ADRException("objects must be an iterable")
+        count = 0
+        for obj in objects:
+            obj.save(**kwargs)
+            count += 1
+        return count
+
+    def _is_sqlite(self, database: str) -> bool:
+        return "sqlite" in self._databases[database]["ENGINE"]
+
+    def _get_db_dir(self, database: str) -> str:
+        return self._databases[database]["NAME"]
+
+    def copy_objects(
+        self,
+        object_type: Union[Session, Dataset, Type[Item], Type[Template]],
+        target_database: str,
+        source_database: str = "default",
+        query: str = "",
+        target_media_dir: str = "",
+        test: bool = False,
+    ) -> int:
+        """
+        This copies a selected collection of objects from one database to another.
+
+        GUIDs are preserved and any referenced session and dataset objects are copied as
+        well.
+        """
+        if not issubclass(object_type, (Item, Template, Session, Dataset)):
+            self._logger.error(f"{object_type} is not valid")
+            raise TypeError(f"{object_type} is not valid")
+
+        if target_database not in self._databases or source_database not in self._databases:
+            raise ADRException(
+                f"'{source_database}' and '{target_database}' must be configured first"
+            )
+
+        objects = self.query(object_type, query=query)
+        copy_list = []
+        media_dir = None
+
+        if issubclass(object_type, Item):
+            for item in objects:
+                # check for media dir if item has a physical file
+                if getattr(item, "has_file", False) and not media_dir:
+                    if target_media_dir:
+                        media_dir = target_media_dir
+                    elif self._is_sqlite(target_database):
+                        media_dir = self._check_dir(
+                            Path(self._get_db_dir(target_database)).parent / "media"
+                        )
+                    else:
+                        raise ADRException(
+                            "'target_media_dir' argument must be specified because one of the objects"
+                            " contains media to copy.'"
+                        )
+                # save the sessions, datasets
+                # assign to the item to create the new relation
+                # and then add to the copy list
+                session, _ = Session.get_or_create(**item.session.as_dict(), using=target_database)
+                item.session = session
+                dataset, _ = Dataset.get_or_create(**item.dataset.as_dict(), using=target_database)
+                item.dataset = dataset
+                copy_list.append(item)
+        elif issubclass(object_type, Template):
+            ...
+        else:  # sessions, datasets
+            copy_list = list(objects)
+
+        if test:
+            self._logger.info(f"Copying {len(copy_list)} objects...")
+            return len(copy_list)
+
+        try:
+            count = self.create_objects(copy_list, using=target_database)
+        except Exception as e:
+            raise ADRException(f"Some objects could not be copied: {e}")
+
+        # copy media
+        print(media_dir)
+
+        return count