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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
# -*- coding: utf-8 -*-
import traceback
import six
class ObjectImportError(Exception):
pass
def import_items(import_directives, style, quiet_load=False):
"""
Import the items in import_directives and return a list of the imported items
Each item in import_directives should be one of the following forms
* a tuple like ('module.submodule', ('classname1', 'classname2')), which indicates a 'from module.submodule import classname1, classname2'
* a tuple like ('module.submodule', 'classname1'), which indicates a 'from module.submodule import classname1'
* a tuple like ('module.submodule', '*'), which indicates a 'from module.submodule import *'
* a simple 'module.submodule' which indicates 'import module.submodule'.
Returns a dict mapping the names to the imported items
"""
imported_objects = {}
for directive in import_directives:
try:
# First try a straight import
if isinstance(directive, six.string_types):
imported_object = __import__(directive)
imported_objects[directive.split('.')[0]] = imported_object
if not quiet_load:
print(style.SQL_COLTYPE("import %s" % directive))
continue
elif isinstance(directive, (list, tuple)) and len(directive) == 2:
if not isinstance(directive[0], six.string_types):
if not quiet_load:
print(style.ERROR("Unable to import %r: module name must be of type string" % directive[0]))
continue
if isinstance(directive[1], (list, tuple)) and all(isinstance(e, six.string_types) for e in directive[1]):
# Try the ('module.submodule', ('classname1', 'classname2')) form
imported_object = __import__(directive[0], {}, {}, directive[1])
imported_names = []
for name in directive[1]:
try:
imported_objects[name] = getattr(imported_object, name)
except AttributeError:
if not quiet_load:
print(style.ERROR("Unable to import %r from %r: %r does not exist" % (name, directive[0], name)))
else:
imported_names.append(name)
if not quiet_load:
print(style.SQL_COLTYPE("from %s import %s" % (directive[0], ', '.join(imported_names))))
elif isinstance(directive[1], six.string_types):
# If it is a tuple, but the second item isn't a list, so we have something like ('module.submodule', 'classname1')
# Check for the special '*' to import all
if directive[1] == '*':
imported_object = __import__(directive[0], {}, {}, directive[1])
for k in dir(imported_object):
imported_objects[k] = getattr(imported_object, k)
if not quiet_load:
print(style.SQL_COLTYPE("from %s import *" % directive[0]))
else:
imported_object = getattr(__import__(directive[0], {}, {}, [directive[1]]), directive[1])
imported_objects[directive[1]] = imported_object
if not quiet_load:
print(style.SQL_COLTYPE("from %s import %s" % (directive[0], directive[1])))
else:
if not quiet_load:
print(style.ERROR("Unable to import %r from %r: names must be of type string" % (directive[1], directive[0])))
else:
if not quiet_load:
print(style.ERROR("Unable to import %r: names must be of type string" % directive))
except ImportError:
try:
if not quiet_load:
print(style.ERROR("Unable to import %r" % directive))
except TypeError:
if not quiet_load:
print(style.ERROR("Unable to import %r from %r" % directive))
return imported_objects
def import_objects(options, style):
from django.apps import apps
from django import setup
if not apps.ready:
setup()
def get_apps_and_models():
for app in apps.get_app_configs():
if app.models_module:
yield app.models_module, app.get_models()
mongoengine = False
try:
from mongoengine.base import _document_registry
mongoengine = True
except:
pass
from django.conf import settings
imported_objects = {}
dont_load_cli = options.get('dont_load') # optparse will set this to [] if it doensnt exists
dont_load_conf = getattr(settings, 'SHELL_PLUS_DONT_LOAD', [])
dont_load = dont_load_cli + dont_load_conf
quiet_load = options.get('quiet_load')
model_aliases = getattr(settings, 'SHELL_PLUS_MODEL_ALIASES', {})
app_prefixes = getattr(settings, 'SHELL_PLUS_APP_PREFIXES', {})
# Perform pre-imports before any other imports
SHELL_PLUS_PRE_IMPORTS = getattr(settings, 'SHELL_PLUS_PRE_IMPORTS', {})
if SHELL_PLUS_PRE_IMPORTS:
if not quiet_load:
print(style.SQL_TABLE("# Shell Plus User Imports"))
imports = import_items(SHELL_PLUS_PRE_IMPORTS, style, quiet_load=quiet_load)
for k, v in six.iteritems(imports):
imported_objects[k] = v
load_models = {}
if mongoengine:
for name, mod in six.iteritems(_document_registry):
name = name.split('.')[-1]
app_name = mod.__module__.split('.')[-2]
if app_name in dont_load or ("%s.%s" % (app_name, name)) in dont_load:
continue
load_models.setdefault(mod.__module__, [])
load_models[mod.__module__].append(name)
for app_mod, app_models in get_apps_and_models():
if not app_models:
continue
app_name = app_mod.__name__.split('.')[-2]
if app_name in dont_load:
continue
app_aliases = model_aliases.get(app_name, {})
for mod in app_models:
if "%s.%s" % (app_name, mod.__name__) in dont_load:
continue
if mod.__module__:
# Only add the module to the dict if `__module__` is not empty.
load_models.setdefault(mod.__module__, [])
load_models[mod.__module__].append(mod.__name__)
if not quiet_load:
print(style.SQL_TABLE("# Shell Plus Model Imports"))
for app_mod, models in sorted(six.iteritems(load_models)):
try:
app_name = app_mod.split('.')[-2]
except IndexError:
# Some weird model naming scheme like in Sentry.
app_name = app_mod
app_aliases = model_aliases.get(app_name, {})
prefix = app_prefixes.get(app_name)
model_labels = []
for model_name in sorted(models):
try:
imported_object = getattr(__import__(app_mod, {}, {}, [model_name]), model_name)
if "%s.%s" % (app_name, model_name) in dont_load:
continue
alias = app_aliases.get(model_name)
if not alias and prefix:
alias = "%s_%s" % (prefix, model_name)
else:
alias = model_name
imported_objects[alias] = imported_object
if model_name == alias:
model_labels.append(model_name)
else:
model_labels.append("%s (as %s)" % (model_name, alias))
except AttributeError as e:
if options.get("traceback"):
traceback.print_exc()
if not quiet_load:
print(style.ERROR("Failed to import '%s' from '%s' reason: %s" % (model_name, app_mod, str(e))))
continue
if not quiet_load:
print(style.SQL_COLTYPE("from %s import %s" % (app_mod, ", ".join(model_labels))))
# Imports often used from Django
if getattr(settings, 'SHELL_PLUS_DJANGO_IMPORTS', True):
if not quiet_load:
print(style.SQL_TABLE("# Shell Plus Django Imports"))
from django import VERSION as DJANGO_VERSION
SHELL_PLUS_DJANGO_IMPORTS = {
'django.core.cache': ['cache'],
'django.conf': ['settings'],
'django.db': ['transaction'],
'django.db.models': [
'Avg', 'Case', 'Count', 'F', 'Max', 'Min', 'Prefetch', 'Q',
'Sum', 'When',
],
'django.utils': ['timezone'],
}
if DJANGO_VERSION < (1, 10):
SHELL_PLUS_DJANGO_IMPORTS.update({
'django.core.urlresolvers': ['reverse'],
})
else:
SHELL_PLUS_DJANGO_IMPORTS.update({
'django.urls': ['reverse'],
})
imports = import_items(SHELL_PLUS_DJANGO_IMPORTS.items(), style, quiet_load=quiet_load)
for k, v in six.iteritems(imports):
imported_objects[k] = v
# Perform post-imports after any other imports
SHELL_PLUS_POST_IMPORTS = getattr(settings, 'SHELL_PLUS_POST_IMPORTS', {})
if SHELL_PLUS_POST_IMPORTS:
if not quiet_load:
print(style.SQL_TABLE("# Shell Plus User Imports"))
imports = import_items(SHELL_PLUS_POST_IMPORTS, style, quiet_load=quiet_load)
for k, v in six.iteritems(imports):
imported_objects[k] = v
return imported_objects
|
