forked from sbussetti/django-admincommand
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcore.py
More file actions
75 lines (63 loc) · 2.66 KB
/
core.py
File metadata and controls
75 lines (63 loc) · 2.66 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
from StringIO import StringIO
from importlib import import_module
from django.conf import settings
from django.core import management
from django.core.management import get_commands
from django.core.management import load_command_class
from django.core.management.base import BaseCommand
from django.contrib.auth.models import User
from async import schedule
from admincommand.models import AdminCommand
# Cache variable to store runnable commands configuration
_command_configs = {}
def get_admin_commands():
if not _command_configs:
for app_module_path in settings.INSTALLED_APPS:
try:
admin_commands_path = '%s.admincommands' % app_module_path
module = import_module(admin_commands_path)
except ImportError:
pass
else:
configs = dir(module)
for config_name in configs:
AdminCommandClass = getattr(module, config_name)
if (isinstance(AdminCommandClass, type)
and AdminCommandClass is not AdminCommand
and issubclass(AdminCommandClass, AdminCommand)):
command_config = AdminCommandClass()
_command_configs[command_config.url_name()] = command_config
return _command_configs
def get_command(name):
# this is a copy pasted from django.core.management.call_command
app_name = get_commands()[name]
if isinstance(app_name, BaseCommand):
# If the command is already loaded, use it directly.
klass = app_name
else:
klass = load_command_class(app_name, name)
return klass
def call_command(command_name, user_pk, args=None, kwargs=None):
"""Call command and store output"""
user = User.objects.get(pk=user_pk)
kwargs = kwargs if kwargs else {}
args = args if args else []
output = StringIO()
kwargs['stdout'] = output
management.call_command(command_name, *args, **kwargs)
return output.getvalue()
def run_command(command_config, cleaned_data, user):
if hasattr(command_config, 'get_command_arguments'):
args, kwargs = command_config.get_command_arguments(cleaned_data, user)
else:
args, kwargs = list(), dict()
if command_config.asynchronous:
task = schedule(call_command, [command_config.command_name(), user.pk, args, kwargs])
return task
else:
# Change stdout to a StringIO to be able to retrieve output and
# display it to the user
output = StringIO()
kwargs['stdout'] = output
management.call_command(command_config.command_name(), *args, **kwargs)
return output.getvalue()