-
Notifications
You must be signed in to change notification settings - Fork 882
/
paster.py
307 lines (243 loc) · 10.6 KB
/
paster.py
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
import os
import sys
from code import interact
from paste.deploy import loadapp
from paste.script.command import Command
from paste.script.templates import Template
from paste.util.template import paste_script_template_renderer
from pyramid.scripting import get_root
class PyramidTemplate(Template):
def pre(self, command, output_dir, vars): # pragma: no cover
vars['random_string'] = os.urandom(20).encode('hex')
package_logger = vars['package']
if package_logger == 'root':
# Rename the app logger in the rare case a project is named 'root'
package_logger = 'app'
vars['package_logger'] = package_logger
return Template.pre(self, command, output_dir, vars)
class StarterProjectTemplate(PyramidTemplate):
_template_dir = 'paster_templates/starter'
summary = 'pyramid starter project'
template_renderer = staticmethod(paste_script_template_renderer)
class ZODBProjectTemplate(PyramidTemplate):
_template_dir = 'paster_templates/zodb'
summary = 'pyramid ZODB starter project'
template_renderer = staticmethod(paste_script_template_renderer)
class RoutesAlchemyProjectTemplate(PyramidTemplate):
_template_dir = 'paster_templates/routesalchemy'
summary = 'pyramid SQLAlchemy project using url dispatch (no traversal)'
template_renderer = staticmethod(paste_script_template_renderer)
class AlchemyProjectTemplate(PyramidTemplate):
_template_dir = 'paster_templates/alchemy'
summary = 'pyramid SQLAlchemy project using traversal'
template_renderer = staticmethod(paste_script_template_renderer)
def get_app(config_file, name, loadapp=loadapp):
""" Return the WSGI application named ``name`` in the PasteDeploy
config file ``config_file``"""
config_name = 'config:%s' % config_file
here_dir = os.getcwd()
app = loadapp(config_name, name=name, relative_to=here_dir)
return app
_marker = object()
class PCommand(Command):
get_app = staticmethod(get_app) # hook point
get_root = staticmethod(get_root) # hook point
group_name = 'pyramid'
interact = (interact,) # for testing
loadapp = (loadapp,) # for testing
verbose = 3
def __init__(self, *arg, **kw):
# needs to be in constructor to support Jython (used to be at class
# scope as ``usage = '\n' + __doc__``.
self.usage = '\n' + self.__doc__
Command.__init__(self, *arg, **kw)
class PShellCommand(PCommand):
"""Open an interactive shell with a :app:`Pyramid` app loaded.
This command accepts two positional arguments:
``config_file`` -- specifies the PasteDeploy config file to use
for the interactive shell.
``section_name`` -- specifies the section name in the PasteDeploy
config file that represents the application.
Example::
$ paster pshell myapp.ini main
.. note:: You should use a ``section_name`` that refers to the
actual ``app`` section in the config file that points at
your Pyramid app without any middleware wrapping, or this
command will almost certainly fail.
"""
summary = "Open an interactive shell with a pyramid app loaded"
min_args = 2
max_args = 2
parser = Command.standard_parser(simulate=True)
parser.add_option('-d', '--disable-ipython',
action='store_true',
dest='disable_ipython',
help="Don't use IPython even if it is available")
def command(self, IPShell=_marker):
if IPShell is _marker:
try: #pragma no cover
from IPython.Shell import IPShell
except ImportError: #pragma no cover
IPShell = None
cprt =('Type "help" for more information. "root" is the Pyramid app '
'root object, "registry" is the Pyramid registry object.')
banner = "Python %s on %s\n%s" % (sys.version, sys.platform, cprt)
config_file, section_name = self.args
self.logging_file_config(config_file)
app = self.get_app(config_file, section_name, loadapp=self.loadapp[0])
root, closer = self.get_root(app)
shell_globals = {'root':root, 'registry':app.registry}
if IPShell is not None and not self.options.disable_ipython:
try:
shell = IPShell(argv=[], user_ns=shell_globals)
shell.IP.BANNER = shell.IP.BANNER + '\n\n' + banner
shell.mainloop()
finally:
closer()
else:
try:
self.interact[0](banner, local=shell_globals)
finally:
closer()
BFGShellCommand = PShellCommand # b/w compat forever
class PRoutesCommand(PCommand):
"""Print all URL dispatch routes used by a Pyramid application in the
order in which they are evaluated. Each route includes the name of the
route, the pattern of the route, and the view callable which will be
invoked when the route is matched.
This command accepts two positional arguments:
``config_file`` -- specifies the PasteDeploy config file to use
for the interactive shell.
``section_name`` -- specifies the section name in the PasteDeploy
config file that represents the application.
Example::
$ paster proutes myapp.ini main
.. note:: You should use a ``section_name`` that refers to the
actual ``app`` section in the config file that points at
your Pyramid app without any middleware wrapping, or this
command will almost certainly fail.
"""
summary = "Print all URL dispatch routes related to a Pyramid application"
min_args = 2
max_args = 2
stdout = sys.stdout
parser = Command.standard_parser(simulate=True)
def _get_mapper(self, app):
from pyramid.config import Configurator
registry = app.registry
config = Configurator(registry = registry)
return config.get_routes_mapper()
def out(self, msg): # pragma: no cover
print msg
def command(self):
from pyramid.interfaces import IRouteRequest
from pyramid.interfaces import IViewClassifier
from pyramid.interfaces import IView
from zope.interface import Interface
config_file, section_name = self.args
app = self.get_app(config_file, section_name, loadapp=self.loadapp[0])
registry = app.registry
mapper = self._get_mapper(app)
if mapper is not None:
routes = mapper.get_routes()
fmt = '%-15s %-30s %-25s'
if not routes:
return
self.out(fmt % ('Name', 'Pattern', 'View'))
self.out(
fmt % ('-'*len('Name'), '-'*len('Pattern'), '-'*len('View')))
for route in routes:
request_iface = registry.queryUtility(IRouteRequest,
name=route.name)
view_callable = None
if (request_iface is None) or (route.factory is not None):
self.out(fmt % (route.name, route.pattern, '<unknown>'))
else:
view_callable = registry.adapters.lookup(
(IViewClassifier, request_iface, Interface),
IView, name='', default=None)
self.out(fmt % (route.name, route.pattern, view_callable))
# For paste.deploy server instantiation (egg:pyramid#wsgiref)
def wsgiref_server_runner(wsgi_app, global_conf, **kw): # pragma: no cover
from wsgiref.simple_server import make_server
host = kw.get('host', '0.0.0.0')
port = int(kw.get('port', 8080))
server = make_server(host, port, wsgi_app)
print('Starting HTTP server on http://%s:%s' % (host, port))
server.serve_forever()
# For paste.deploy server instantiation (egg:pyramid#cherrypy)
def cherrypy_server_runner(
app, global_conf=None, host='127.0.0.1', port=None,
ssl_pem=None, protocol_version=None, numthreads=None,
server_name=None, max=None, request_queue_size=None,
timeout=None
): # pragma: no cover
"""
Entry point for CherryPy's WSGI server
Serves the specified WSGI app via CherryPyWSGIServer.
``app``
The WSGI 'application callable'; multiple WSGI applications
may be passed as (script_name, callable) pairs.
``host``
This is the ipaddress to bind to (or a hostname if your
nameserver is properly configured). This defaults to
127.0.0.1, which is not a public interface.
``port``
The port to run on, defaults to 8080 for HTTP, or 4443 for
HTTPS. This can be a string or an integer value.
``ssl_pem``
This an optional SSL certificate file (via OpenSSL) You can
generate a self-signed test PEM certificate file as follows:
$ openssl genrsa 1024 > host.key
$ chmod 400 host.key
$ openssl req -new -x509 -nodes -sha1 -days 365 \\
-key host.key > host.cert
$ cat host.cert host.key > host.pem
$ chmod 400 host.pem
``protocol_version``
The protocol used by the server, by default ``HTTP/1.1``.
``numthreads``
The number of worker threads to create.
``server_name``
The string to set for WSGI's SERVER_NAME environ entry.
``max``
The maximum number of queued requests. (defaults to -1 = no
limit).
``request_queue_size``
The 'backlog' argument to socket.listen(); specifies the
maximum number of queued connections.
``timeout``
The timeout in seconds for accepted connections.
"""
is_ssl = False
if ssl_pem:
port = port or 4443
is_ssl = True
if not port:
if ':' in host:
host, port = host.split(':', 1)
else:
port = 8080
bind_addr = (host, int(port))
kwargs = {}
for var_name in ('numthreads', 'max', 'request_queue_size', 'timeout'):
var = locals()[var_name]
if var is not None:
kwargs[var_name] = int(var)
from cherrypy import wsgiserver
server = wsgiserver.CherryPyWSGIServer(bind_addr, app,
server_name=server_name, **kwargs)
server.ssl_certificate = server.ssl_private_key = ssl_pem
if protocol_version:
server.protocol = protocol_version
try:
protocol = is_ssl and 'https' or 'http'
if host == '0.0.0.0':
print('serving on 0.0.0.0:%s view at %s://127.0.0.1:%s' %
(port, protocol, port))
else:
print('serving on %s://%s:%s' % (protocol, host, port))
server.start()
except (KeyboardInterrupt, SystemExit):
server.stop()
return server