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
|
import http.server
import logging
import os
import pathlib
import random
import socket
import threading
import urllib.parse
import wsgiref.simple_server
import wsgiref.util
from .util import abspath
from .wsgi import StaticFiles
__all__ = ('resolve_url',)
logger = logging.getLogger(__name__)
def _get_random_port():
while True:
port = random.randint(1023, 65535)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
sock.bind(('localhost', port))
except OSError:
logger.warning('Port %s is in use' % port)
continue
else:
return port
class WSGIRequestHandler11(wsgiref.simple_server.WSGIRequestHandler):
protocol_version = 'HTTP/1.1'
if hasattr(http.server, 'ThreadingHTTPServer'):
# Python 3.7+
class ThreadingWSGIServer(http.server.ThreadingHTTPServer, wsgiref.simple_server.WSGIServer):
pass
else:
# Python 3.6 and earlier
ThreadingWSGIServer = wsgiref.simple_server.WSGIServer
def get_wsgi_server(app):
if hasattr(app, '__webview_url'):
# It's already been spun up and is running
return app.__webview_url
port = _get_random_port()
server = wsgiref.simple_server.make_server(
'localhost', port, app, server_class=ThreadingWSGIServer,
handler_class=WSGIRequestHandler11,
)
t = threading.Thread(target=server.serve_forever)
t.daemon = True
t.start()
app.__webview_url = 'http://localhost:{0}/'.format(port)
logger.debug('HTTP server for {!r} started on {}'.format(app, app.__webview_url))
return app.__webview_url
_path_apps = {}
def resolve_url(url, should_serve):
"""
Given a URL-ish thing and a bool, return a real URL.
* url: A URL, a path-like, a string path, or a wsgi app
* should_serve: Should we start a server
Note that if given a wsgi app, a server will always be started.
"""
if isinstance(url, str):
bits = urllib.parse.urlparse(url)
else:
# To create an empty version of the struct
bits = urllib.parse.urlparse("")
if url is None:
return None
elif bits.scheme and bits.scheme != 'file':
# an http, https, etc URL
return url
elif hasattr(url, '__fspath__') or isinstance(url, str):
# A local path
# 1. Resolve the several options into an actual path
if hasattr(url, '__fspath__'):
path = os.fspath(url)
elif bits.scheme == 'file':
path = os.path.dirname(bits.netloc or bits.path)
else:
path = url
# If it's a relative path, resolve it relative to the app root
path = abspath(path)
# If we have not been asked to serve local paths, bail
if not should_serve:
# using pathlib for this because it turns out file URLs are full of dragons
return pathlib.Path(path).as_uri()
if os.path.isdir(path):
rootdir = path
homepage = None
else:
rootdir, homepage = os.path.split(path)
# Get/Build a WSGI app to serve the path and spin it up
if path not in _path_apps:
_path_apps[path] = StaticFiles(rootdir)
url = get_wsgi_server(_path_apps[path])
if homepage is not None:
url = urllib.parse.urljoin(url, homepage)
return url
elif callable(url):
# A wsgi application
return get_wsgi_server(url)
else:
raise TypeError("Cannot resolve {!r} into a URL".format(url))
|
