Source code for smarter.apps.plugin.management.commands.retrieve_plugin

"""This module retrieves the json representation of a plugin."""

from typing import Optional

from django.core.exceptions import MultipleObjectsReturned

from smarter.apps.account.models import Account, User, UserProfile
from smarter.apps.account.utils import (
    get_cached_user_for_username,
)
from smarter.apps.plugin.manifest.controller import PluginController
from smarter.apps.plugin.models import PluginMeta
from smarter.lib.django.management.base import SmarterCommand


# pylint: disable=E1101
[docs] class Command(SmarterCommand): """Prints the json representation of a plugin to the console."""
[docs] def add_arguments(self, parser): """Add arguments to the command.""" parser.add_argument("--account_number", type=str, required=True, help="Account number that owns the plugin.") parser.add_argument("--username", type=str, required=True, help="The user that owns the plugin.") parser.add_argument("--name", type=str, required=True, help="The name of the plugin to retrieve.")
[docs] def handle(self, *args, **options): """retrieve the plugin.""" self.handle_begin() account_number = options["account_number"] username = options["username"] name = options["name"] account: Optional[Account] = None plugin_meta: Optional[PluginMeta] = None user: Optional[User] = None user_profile: Optional[UserProfile] = None try: user = get_cached_user_for_username(username=username) if user is None: raise User.DoesNotExist(f"User with username {username} does not exist.") except User.DoesNotExist as e: self.handle_completed_failure( e, f"manage.py retrieve_plugin: User {username} does not exist.", ) raise try: account = Account.get_cached_object(account_number=account_number) if account is None: raise Account.DoesNotExist(f"Account with account number {account_number} does not exist.") except Account.DoesNotExist as e: self.handle_completed_failure( e, f"manage.py retrieve_plugin: Account {account_number} does not exist.", ) try: user_profile = UserProfile.get_cached_object(user=user, account=account) # type: ignore except UserProfile.DoesNotExist as e: self.handle_completed_failure( e, f"manage.py retrieve_plugin: UserProfile for {user} and {account} does not exist.", ) self.stdout.write(f"manage.py retrieve_plugin: Retrieving plugin {name} for account {account}") try: plugin_meta = PluginMeta.objects.get(name=name, user_profile__account=account) except MultipleObjectsReturned as e: try: plugin_meta = PluginMeta.objects.get(name=name, user_profile=user_profile) except PluginMeta.DoesNotExist as e2: self.handle_completed_failure( e2, f"manage.py retrieve_plugin: Multiple plugins named {name} exist for account {account_number}, but none for user {user_profile}.", ) except PluginMeta.DoesNotExist as e: self.handle_completed_failure( e, f"manage.py retrieve_plugin: Plugin {name} does not exist.", ) controller = PluginController(user_profile=user_profile, plugin_meta=plugin_meta) # type: ignore plugin = controller.obj if not plugin: self.handle_completed_failure( None, f"Plugin {name} does not exist.", ) raise ValueError(f"Plugin {name} does not exist.") print(plugin.data) self.handle_completed_success()