def one_request(port):
"""
Listen for one http request on port, then close and return request query
args:
port (int): the port to listen for the request
returns:
str: the request
"""
logger.info("listening for a request on port {}...".format(port))
class RequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(landing_page.encode('utf-8'))
self.server.path = self.path
httpd = HTTPServer(('', port), RequestHandler)
httpd.handle_request()
httpd.server_close()
parsed = urlparse(httpd.path)
logger.info("received a request {}".format(httpd.path))
return parse_qs(parsed.query)
python类BaseHTTPRequestHandler()的实例源码
def _create_healthcheck(self, port):
class HealthcheckHTTPRequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
health = {'status': 'ok'}
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(bytes(json.dumps(health), 'UTF-8'))
def log_message(self, format, *args):
logger.debug("Healthcheck from %s %s" % (self.address_string(), format % args))
self.healthcheck_http_server = HTTPServer(('', port), HealthcheckHTTPRequestHandler)
healthcheck_thread = Thread(target=self._run_healthcheck_thread,
name='healthcheck', args=(), daemon=True)
healthcheck_thread.start()
def _make_standalone_handler(preamble):
"""Class factory used so that preamble can be passed to :py:class:`_StandaloneHandler`
without use of static members"""
class _StandaloneHandler(BaseHTTPRequestHandler, object):
"""HTTP Handler for standalone mode"""
def do_GET(self):
self.send_response(200)
self.send_header('Content-type', 'application/json; charset=utf-8')
self.send_header('Content-length', len(preamble))
self.end_headers()
self.wfile.write(preamble.encode('utf-8'))
def log_message(self, format, *args):
# suppress logging on requests
return
return _StandaloneHandler
def combined_handler(path, path_action, file_id, stream_action, path_input=None, path_output=None, stream_input=None, stream_output=None):
class _(BaseHTTPRequestHandler):
def do_POST(self):
request_path = urlparse(self.path).path
paths_path = alluxio.client._paths_url_path(path, path_action)
streams_path = alluxio.client._streams_url_path(
file_id, stream_action)
close_path = alluxio.client._streams_url_path(file_id, 'close')
if request_path == paths_path:
handle_paths_request(
self, path, path_action, input=path_input, output=path_output)
elif request_path == streams_path:
handle_streams_request(
self, file_id, stream_action, input=stream_input, output=stream_output)
elif request_path == close_path:
self.send_response(200)
return _
def __init__(self, addr, handler, poll_interval=0.5,
log=False, sslctx=None):
class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler):
def __getattr__(self, name, default=None):
if name.startswith('do_'):
return self.process_request
raise AttributeError(name)
def process_request(self):
self.server._handler(self)
def log_message(self, format, *args):
if log:
super(DelegatingHTTPRequestHandler,
self).log_message(format, *args)
HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)
ControlMixin.__init__(self, handler, poll_interval)
self.sslctx = sslctx
def __init__(self, addr, handler, poll_interval=0.5,
log=False, sslctx=None):
class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler):
def __getattr__(self, name, default=None):
if name.startswith('do_'):
return self.process_request
raise AttributeError(name)
def process_request(self):
self.server._handler(self)
def log_message(self, format, *args):
if log:
super(DelegatingHTTPRequestHandler,
self).log_message(format, *args)
HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)
ControlMixin.__init__(self, handler, poll_interval)
self.sslctx = sslctx
def receive_code(self, port, final_redirect=None):
class RequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
try:
query = urlparse(self.path).query
query = parse_qs(query)
self.code = query['code'][0]
except Exception as e:
self.send_response(500)
self.code = None
else:
if final_redirect:
self.send_response(302)
self.send_header("Location", final_redirect)
else:
self.send_response(200)
finally:
self.end_headers()
address = ('localhost', port)
server = HTTPServer(address, RequestHandler)
request, client_address = server.get_request()
code = RequestHandler(request, client_address, server).code
return code
def __init__(self, addr, handler, poll_interval=0.5,
log=False, sslctx=None):
class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler):
def __getattr__(self, name, default=None):
if name.startswith('do_'):
return self.process_request
raise AttributeError(name)
def process_request(self):
self.server._handler(self)
def log_message(self, format, *args):
if log:
super(DelegatingHTTPRequestHandler,
self).log_message(format, *args)
HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)
ControlMixin.__init__(self, handler, poll_interval)
self.sslctx = sslctx
def run(server_class=HTTPServer, handler_class=BaseHTTPRequestHandler):
server_address = ('', 8000)
httpd = server_class(server_address, handler_class)
httpd.serve_forever()
# RELAY.on() activate relay
# Import JSON config file as a dictionary for future reference by ID and IP.
# Thread: Listen for signal from thermostat. When received, use the function
# below and pass it the ID and IP.
# Function: send ID, IP, and delay to the server.
# Thread: Listen for activate command from server. When received, open relay.
def start_server(*resp):
"""HTTP server replying with the given responses to the expected
requests."""
def url(port, path):
return 'http://%s:%s%s' % (socket.gethostname(), port, path)
responses = list(reversed(resp))
class MyHandler(BaseHTTPServer.BaseHTTPRequestHandler):
def do_HEAD(self):
response = responses.pop()
assert response.path == self.path
self.send_response(response.code)
for header, value in list(response.headers.items()):
self.send_header(header, value)
self.end_headers()
httpd = SocketServer.TCPServer(("", 0), MyHandler)
t = threading.Thread(target=httpd.serve_forever)
t.setDaemon(True)
t.start()
port = httpd.server_address[1]
yield functools.partial(url, port)
httpd.shutdown()
def __init__(self, ctx, DocumentRoot, *args):
self._ctx = ctx
self.DocumentRoot = DocumentRoot
BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args)
def __init__(self, ctx, DocumentRoot, *args):
self._ctx = ctx
self.DocumentRoot = DocumentRoot
BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args)
def setUp(self):
super(TestRequestHandler, self).setUp()
client_address = 'localhost'
server = '/tmp/server.lock'
req = mock.MagicMock()
with mock.patch.object(BaseHTTPRequestHandler, '__init__') as m_http:
m_http.return_value = None
self._req_handler = m_pool.RequestHandler(req, client_address,
server)
self._req_handler.rfile = mock.Mock()
self._req_handler.wfile = mock.Mock()
def log_message(self, format, *args):
"""Log an arbitrary message.
This is used by all other logging functions.
It overrides ``BaseHTTPRequestHandler.log_message``, which logs to ``sys.stderr``.
The first argument, FORMAT, is a format string for the message to be logged. If the format
string contains any % escapes requiring parameters, they should be specified as subsequent
arguments (it's just like printf!).
The client ip is prefixed to every message.
"""
self.logger.debug("%s - - %s" % (self.address_string(), format % args))
def mockup_http_static_server(content: bytes = b'Simple file content.', content_type: str = None, **kwargs):
class StaticMockupHandler(BaseHTTPRequestHandler): # pragma: no cover
def serve_text(self):
self.send_header('Content-Type', "text/plain")
self.send_header('Content-Length', str(len(content)))
self.send_header('Last-Modified', self.date_time_string())
self.end_headers()
self.wfile.write(content)
def serve_static_file(self, filename: AnyStr):
self.send_header('Content-Type', guess_type(filename))
with open(filename, 'rb') as f:
self.serve_stream(f)
def serve_stream(self, stream: FileLike):
buffer = io.BytesIO()
self.send_header('Content-Length', str(copy_stream(stream, buffer)))
self.end_headers()
buffer.seek(0)
try:
copy_stream(buffer, self.wfile)
except ConnectionResetError:
pass
# noinspection PyPep8Naming
def do_GET(self):
self.send_response(HTTPStatus.OK)
if isinstance(content, bytes):
self.serve_text()
elif isinstance(content, str):
self.serve_static_file(content)
else:
self.send_header('Content-Type', content_type)
# noinspection PyTypeChecker
self.serve_stream(content)
return simple_http_server(StaticMockupHandler, **kwargs)
def get_token_to_login(self, port=8000):
# WARNING: This method is NOT thread-safe
# FIXME: find another way to store token and ensure method works properly
class AllegroRequestHandler(BaseHTTPRequestHandler):
user_token = None
def do_GET(self):
token = self.path.rsplit('?code=', 1)[-1]
AllegroRequestHandler.user_token = token
server = HTTPServer(('0.0.0.0', port), AllegroRequestHandler)
server.handle_request()
user_token = AllegroRequestHandler.user_token
AllegroRequestHandler.user_token = None
return user_token
def paths_handler(path, action, params=None, input=None, output=None):
class _(BaseHTTPRequestHandler):
def do_POST(self):
handle_paths_request(self, path, action,
params=params, input=input, output=output)
return _
def streams_handler(file_id, action, input=None, output=None):
class _(BaseHTTPRequestHandler):
def do_POST(self):
handle_streams_request(self, file_id, action,
input=input, output=output)
return _
def run(
server_class=HTTPServer,
handler_class=BaseHTTPRequestHandler, port=8098):
server_address = ('127.0.0.1', port)
httpd = server_class(server_address, handler_class)
try:
httpd.serve_forever()
except KeyboardInterrupt:
httpd.socket.close()
def get_handler(callback_on_token):
class HTTPHandler(BaseHTTPRequestHandler):
def do_GET(self):
"""Respond to a GET request."""
logger.debug('Received GET request %s', self.path)
parts = parse_qs(urlparse(self.path).query)
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.write("<html><head><title>Spotify oAuth</title></head>")
self.write("<body><p>This is a token response from Spotify.</p>")
self.write('<p>%r</p>' % parts)
self.write("<p>You accessed path: %s</p>" % self.path)
self.write(
"<p>Check spoppy, if everything went well you're logged in "
"now!</p>"
)
self.write("</body></html>")
if parts:
logger.debug('Got parts %s, calling callback', parts)
callback_on_token(parts)
def write(self, line):
self.wfile.write(line.encode('utf-8'))
return HTTPHandler
def log_message(self, fmt, *args):
"""Override logging settings of base class
This method reformats standard request logging provided by the base
class BaseHTTPRequestHandler and sends it to logger/formatter
configured by the user during logging module initialization
Args:
fmt, args: just like base class
"""
logging.info("REQ: {0} {1}".format(self.address_string(), fmt % args))
def handle_one_request(self):
try:
http_server.BaseHTTPRequestHandler.handle_one_request(self)
except:
pass
# Gracefully handle disconnects
def init_server(server_class=HTTPServer, handler_class=BaseHTTPRequestHandler):
server_address = ('127.0.0.1', PORT)
httpd = server_class(server_address, CvimServer)
httpd.serve_forever()
def __init__(self, app):
self.app = app
self.format = '%s - - [%s] "%s %s %s" - %s'
f = BytesIO()
class FakeSocket:
def makefile(self, *a):
return f
# take log_date_time_string method from BaseHTTPRequestHandler
self.log_date_time_string = BaseHTTPRequestHandler(FakeSocket(), None, None).log_date_time_string
def __init__(self,
processor,
server_address,
iprot_factory,
server_class=http_server.HTTPServer):
"""Set up protocol factories and HTTP server.
See http.server for server_address.
See TServer for protocol factories.
"""
TServer.__init__(self, processor, trans=None,
itrans_factory=None, iprot_factory=iprot_factory,
otrans_factory=None, oprot_factory=None)
thttpserver = self
class RequestHander(http_server.BaseHTTPRequestHandler):
# Don't care about the request path.
def do_POST(self):
# Don't care about the request path.
itrans = TFileObjectTransport(self.rfile)
otrans = TFileObjectTransport(self.wfile)
itrans = TBufferedTransport(
itrans, int(self.headers['Content-Length']))
otrans = TMemoryBuffer()
iprot = thttpserver.iprot_factory.get_protocol(itrans)
oprot = thttpserver.oprot_factory.get_protocol(otrans)
try:
thttpserver.processor.process(iprot, oprot)
except ResponseException as exn:
exn.handler(self)
else:
self.send_response(200)
self.send_header("content-type", "application/x-thrift")
self.end_headers()
self.wfile.write(otrans.getvalue())
self.httpd = server_class(server_address, RequestHander)
def log_message(self, format, *args):
"""Log an arbitrary message.
This is used by all other logging functions.
It overrides ``BaseHTTPRequestHandler.log_message``, which logs to ``sys.stderr``.
The first argument, FORMAT, is a format string for the message to be logged. If the format
string contains any % escapes requiring parameters, they should be specified as subsequent
arguments (it's just like printf!).
The client ip is prefixed to every message.
"""
self.logger.debug("%s - - %s" % (self.address_string(), format % args))
def serve_file(filename, address="", port=45114):
class FileHandler(BaseHTTPRequestHandler):
def do_GET(self): # noqa
try:
file = open(filename, "rb")
stat = os.fstat(file.fileno())
length = stat.st_size
self.send_response(200)
self.send_header("Content-type", "video/mp4")
self.send_header("Content-Length", length)
self.send_header("Accept-Ranges", "bytes")
self.send_header(
"Last-Modified",
time.strftime(
"%a %d %b %Y %H:%M:%S GMT",
time.localtime(os.path.getmtime(filename))
)
)
self.end_headers()
while True:
data = file.read(100 * 1024)
if not data:
break
self.wfile.write(data)
except: # noqa
traceback.print_exc()
file.close()
httpd = SocketServer.TCPServer((address, port), FileHandler)
httpd.serve_forever()
httpd.server_close()
def http_static_server(content: bytes = b'Simple file content.', content_type: str = None, **kwargs):
class StaticMockupHandler(BaseHTTPRequestHandler): # pragma: no cover
def serve_text(self):
self.send_header('Content-Type', "text/plain")
self.send_header('Content-Length', str(len(content)))
self.send_header('Last-Modified', self.date_time_string())
self.end_headers()
self.wfile.write(content)
def serve_static_file(self, filename):
self.send_header('Content-Type', guess_type(filename))
with open(filename, 'rb') as f:
self.serve_stream(f)
def serve_stream(self, stream):
buffer = io.BytesIO()
self.send_header('Content-Length', str(copy_stream(stream, buffer)))
self.end_headers()
buffer.seek(0)
try:
copy_stream(buffer, self.wfile)
except ConnectionResetError:
pass
# noinspection PyPep8Naming
def do_GET(self):
self.send_response(HTTPStatus.OK)
if isinstance(content, bytes):
self.serve_text()
elif isinstance(content, str):
self.serve_static_file(content)
else:
self.send_header('Content-Type', content_type)
# noinspection PyTypeChecker
self.serve_stream(content)
return simple_http_server(None, handler_class=StaticMockupHandler, server_class=HTTPServer, **kwargs)
def HandlerFactory(self):
config = self.config
class Handler(BaseHTTPRequestHandler):
server_version = 'Sauna/' + __version__
def do_GET(self):
data = self.generate_response()
self.wfile.write(data)
def do_HEAD(self):
self.generate_response()
def generate_response(self):
try:
content = self.get_content_from_path()
code = 200
except NotFoundError:
content = {'error': 'Resource not found'}
code = 404
self.send_response(code)
if config['data_type'] == 'json':
self.send_header('Content-Type', 'application/json')
data = json.dumps(content).encode()
elif config['data_type'] == 'html':
self.send_header('Content-Type', 'text/html')
from .html import get_html
data = get_html()
else:
data = 'data type not found'.encode()
self.send_header('Content-Length', len(data))
self.end_headers()
return data
def get_content_from_path(self):
if self.path == '/':
status, code = HTTPServerConsumer.get_current_status()
return {
'status': status,
'code': code,
'checks': HTTPServerConsumer.get_checks_as_dict()
}
else:
raise NotFoundError()
def log_message(self, format, *args):
logger.debug(
'{} {}'.format(self.address_string(), format % args))
return Handler
def _create_handler_class(rules, always_rules):
class _Handler(BaseHTTPRequestHandler):
known_methods = set()
@classmethod
def add_method(cls, method):
'''
Adds a handler function for HTTP method provided
'''
if method in cls.known_methods:
return
func = lambda self: cls._handle(self, method)
setattr(cls, 'do_' + method, func)
cls.known_methods.add(method)
def _read_body(self):
if 'content-length' in self.headers:
length = int(self.headers['content-length'])
return self.rfile.read(length) if length > 0 else None
return None
def _respond(self, response):
self.send_response(response.code)
for key, value in response.headers.items():
self.send_header(key, value)
self.end_headers()
if response.bytes:
self.wfile.write(response.bytes)
def _handle(self, method):
body = self._read_body()
rule = self._respond_with_rules(method, body, rules)
if rule:
rules.remove(rule)
return
always_rule = self._respond_with_rules(method, body, always_rules)
if always_rule:
return
return self.send_error(
500, 'No matching rule found for ' + self.requestline + ' body ' + str(body))
def _respond_with_rules(self, method, body, rules):
matching_rules = [r for r in rules if r.matches(method, self.path, dict(self.headers), body)]
if matching_rules:
rule = matching_rules[0]
self._respond(rule.response)
return rule
return None
for rule in rules:
_Handler.add_method(rule.method)
return _Handler
