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
|
"""
Serve command for Pecan.
"""
from __future__ import print_function
import os
import sys
import time
import subprocess
from pecan.commands import BaseCommand
class ServeCommand(BaseCommand):
"""
Serves a Pecan web application.
This command serves a Pecan web application using the provided
configuration file for the server and application.
"""
arguments = BaseCommand.arguments + ({
'name': '--reload',
'help': 'Watch for changes and automatically reload.',
'default': False,
'action': 'store_true'
},)
def run(self, args):
super(ServeCommand, self).run(args)
app = self.load_app()
self.serve(app, app.config)
def create_subprocess(self):
self.server_process = subprocess.Popen(
[arg for arg in sys.argv if arg != '--reload'],
stdout=sys.stdout, stderr=sys.stderr
)
def watch_and_spawn(self, conf):
from watchdog.observers import Observer
from watchdog.events import (
FileSystemEventHandler, FileSystemMovedEvent, FileModifiedEvent,
DirModifiedEvent
)
print('Monitoring for changes...')
self.create_subprocess()
parent = self
class AggressiveEventHandler(FileSystemEventHandler):
def should_reload(self, event):
for t in (
FileSystemMovedEvent, FileModifiedEvent, DirModifiedEvent
):
if isinstance(event, t):
return True
return False
def on_modified(self, event):
if self.should_reload(event):
parent.server_process.kill()
parent.create_subprocess()
# Determine a list of file paths to monitor
paths = self.paths_to_monitor(conf)
event_handler = AggressiveEventHandler()
for path, recurse in paths:
observer = Observer()
observer.schedule(
event_handler,
path=path,
recursive=recurse
)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
def paths_to_monitor(self, conf):
paths = []
for package_name in getattr(conf.app, 'modules', []):
module = __import__(package_name, fromlist=['app'])
if hasattr(module, 'app') and hasattr(module.app, 'setup_app'):
paths.append((
os.path.dirname(module.__file__),
True
))
break
paths.append((os.path.dirname(conf.__file__), False))
return paths
def _serve(self, app, conf):
from wsgiref.simple_server import make_server
host, port = conf.server.host, int(conf.server.port)
srv = make_server(host, port, app)
print('Starting server in PID %s' % os.getpid())
if host == '0.0.0.0':
print(
'serving on 0.0.0.0:%s, view at http://127.0.0.1:%s' %
(port, port)
)
else:
print("serving on http://%s:%s" % (host, port))
try:
srv.serve_forever()
except KeyboardInterrupt:
# allow CTRL+C to shutdown
pass
def serve(self, app, conf):
"""
A very simple approach for a WSGI server.
"""
if self.args.reload:
try:
self.watch_and_spawn(conf)
except ImportError:
print('The `--reload` option requires `watchdog` to be '
'installed.')
print(' $ pip install watchdog')
else:
self._serve(app, conf)
def gunicorn_run():
"""
The ``gunicorn_pecan`` command for launching ``pecan`` applications
"""
try:
from gunicorn.app.wsgiapp import WSGIApplication
except ImportError as exc:
args = exc.args
arg0 = args[0] if args else ''
arg0 += ' (are you sure `gunicorn` is installed?)'
exc.args = (arg0,) + args[1:]
raise
class PecanApplication(WSGIApplication):
def init(self, parser, opts, args):
if len(args) != 1:
parser.error("No configuration file was specified.")
self.cfgfname = os.path.normpath(
os.path.join(os.getcwd(), args[0])
)
self.cfgfname = os.path.abspath(self.cfgfname)
if not os.path.exists(self.cfgfname):
parser.error("Config file not found: %s" % self.cfgfname)
from pecan.configuration import _runtime_conf, set_config
set_config(self.cfgfname, overwrite=True)
# If available, use the host and port from the pecan config file
cfg = {}
if _runtime_conf.get('server'):
server = _runtime_conf['server']
if hasattr(server, 'host') and hasattr(server, 'port'):
cfg['bind'] = '%s:%s' % (
server.host, server.port
)
return cfg
def load(self):
from pecan.deploy import deploy
return deploy(self.cfgfname)
PecanApplication("%(prog)s [OPTIONS] config.py").run()
|
