feat: Create cli command to transfer applet ownership (M2-7719) (#1575)

src/apps/applets/commands/__init__.py

@@ -1 +1,2 @@
+from apps.applets.commands.applet import app as applet_cli  # noqa: F401
 from apps.applets.commands.applet_ema import app as applet_ema_cli  # noqa: F401

src/apps/applets/commands/applet.py

@@ -0,0 +1,80 @@
+import uuid
+
+import typer
+from rich import print
+
+from apps.applets.service import AppletService
+from apps.transfer_ownership.service import TransferService
+from apps.users import User
+from apps.users.cruds.user import UsersCRUD
+from apps.users.errors import UserIsDeletedError, UserNotFound
+from apps.workspaces.service.check_access import CheckAccessService
+from infrastructure.commands.utils import coro
+from infrastructure.database import atomic, session_manager
+
+app = typer.Typer()
+
+
+def error_msg(msg: str):
+    print(f"[bold red]Error: {msg}[/bold red]")
+
+
+def error(msg: str):
+    error_msg(msg)
+    raise typer.Abort()
+
+
+async def _validate_access(session, user: User, applet_ids: list[uuid.UUID]):
+    for applet_id in applet_ids:
+        try:
+            await AppletService(session, user.id).exist_by_id(applet_id)
+            await CheckAccessService(session, user.id).check_create_transfer_ownership_access(applet_id)
+        except Exception as e:
+            error_msg(f"Applet access error: {applet_id}")
+            error(str(e))
+
+
+@app.command(help="Transfer ownership")
+@coro
+async def transfer_ownership(
+    applet_ids: list[uuid.UUID] = typer.Argument(..., help="A list of Applet IDs for data copying."),
+    source_owner_email: str = typer.Option(
+        ...,
+        "--src-owner-email",
+        "-s",
+        help="Source owner email.",
+    ),
+    target_owner_email: str = typer.Option(
+        ...,
+        "--tgt-owner-email",
+        "-t",
+        help="Target owner email.",
+    ),
+) -> None:
+    source_owner_email = source_owner_email.lower()
+    target_owner_email = target_owner_email.lower()
+    if source_owner_email == target_owner_email:
+        error("Emails are the same.")
+
+    session_maker = session_manager.get_session()
+    async with session_maker() as session:
+        user_repo = UsersCRUD(session)
+        try:
+            source_user = await user_repo.get_by_email(source_owner_email)
+        except (UserNotFound, UserIsDeletedError):
+            error(f"User with email {source_owner_email} not found")
+        await _validate_access(session, source_user, applet_ids)
+
+        try:
+            target_user = await user_repo.get_by_email(target_owner_email)
+        except (UserNotFound, UserIsDeletedError):
+            error(f"User with email {target_owner_email} not found")
+
+        service_from = TransferService(session, source_user)
+        service_to = TransferService(session, target_user)
+        async with atomic(session):
+            for applet_id in applet_ids:
+                print(f"Transfer ownership for applet {applet_id}")
+                transfer = await service_from.save_transfer_request(applet_id, target_owner_email, target_user.id)
+                await service_to.accept_transfer(applet_id, transfer.key)
+                print(f"[green]Transfer ownership for applet {applet_id} finished[/green]")

src/apps/transfer_ownership/service.py

@@ -26,6 +26,18 @@ def __init__(self, session, user: User):
         self._user = user
         self.session = session
 
+    async def save_transfer_request(self, applet_id: uuid.UUID, target_email: str, target_id: uuid.UUID) -> Transfer:
+        transfer = Transfer(
+            email=target_email,
+            applet_id=applet_id,
+            key=uuid.uuid4(),
+            status=TransferOwnershipStatus.PENDING,
+            from_user_id=self._user.id,
+            to_user_id=target_id,
+        )
+        await TransferCRUD(self.session).create(transfer)
+        return transfer
+
     async def initiate_transfer(self, applet_id: uuid.UUID, transfer_request: InitiateTransfer):
         """Initiate a transfer of ownership of an applet."""
         # check if user is owner of applet
@@ -44,15 +56,7 @@ async def initiate_transfer(self, applet_id: uuid.UUID, transfer_request: Initia
             receiver_name = transfer_request.email
             to_user_id = None
 
-        transfer = Transfer(
-            email=transfer_request.email,
-            applet_id=applet_id,
-            key=uuid.uuid4(),
-            status=TransferOwnershipStatus.PENDING,
-            from_user_id=self._user.id,
-            to_user_id=to_user_id,
-        )
-        await TransferCRUD(self.session).create(transfer)
+        transfer = await self.save_transfer_request(applet_id, transfer_request.email, to_user_id)
 
         url = self._generate_transfer_url()
 

src/cli.py

@@ -9,6 +9,7 @@
 
 from apps.activities.commands import activities  # noqa: E402
 from apps.answers.commands import convert_assessments  # noqa: E402
+from apps.applets.commands import applet_cli  # noqa: E402
 from apps.applets.commands import applet_ema_cli  # noqa: E402
 from apps.shared.commands import encryption_cli, patch  # noqa: E402
 from apps.users.commands import token_cli  # noqa: E402
@@ -22,6 +23,7 @@
 cli.add_typer(patch, name="patch")
 cli.add_typer(encryption_cli, name="encryption")
 cli.add_typer(applet_ema_cli, name="applet-ema")
+cli.add_typer(applet_cli, name="applet")
 
 if __name__ == "__main__":
     # with app context?