我们从Python开源项目中，提取了以下32个代码示例，用于说明如何使用http.server.BaseHTTPRequestHandler()。
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)
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 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 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 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 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
def start_listening(bot=None, loop=None, name="", port=8000, certfile=None, webhook_receiver=BaseHTTPRequestHandler, friendly_name="UNKNOWN"): if loop: asyncio.set_event_loop(loop) if bot: webhook_receiver._bot = bot try: httpd = HTTPServer((name, port), webhook_receiver) if certfile: httpd.socket = ssl.wrap_socket( httpd.socket, certfile=certfile, server_side=True) socket = httpd.socket.getsockname() logger.info("%s : %s:%s...", friendly_name, socket[0], socket[1]) httpd.serve_forever() except ssl.SSLError: logger.exception("%s : %s:%s, pem file is invalid/corrupt", friendly_name, name, port) except OSError as err: if err.errno == 2: message = ".pem file is missing/unavailable" elif err.errno == 98: message = "address/port in use" else: message = str(err.strerror) logger.exception("%s : %s:%s, %s", friendly_name, name, port, message) try: httpd.socket.close() except: pass except KeyboardInterrupt: httpd.socket.close()
def get_redirect_response(send_oauth2_request, oauth2_url, redirect_port, handle, available_duration_for_login_in_seconds): def get_redirect_response(): server = start_up_server(SpotifyRedirectURIHandler()) send_oauth2_request(oauth2_url) block_thread_until_spotify_responds(server) shut_down_server(server) def start_up_server(handler): return HTTPServer(('', redirect_port), handler) def SpotifyRedirectURIHandler(): class SpotifyRedirectURIHandler(BaseHTTPRequestHandler): def do_GET(self): self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() handle(self.path) html = open(self.__html_location(), "r").read() self.wfile.write(bytes(html, "utf8")) return def __html_location(self): return os.path.abspath( os.path.join( os.path.dirname(__file__), "../../resources/index.html" ) ) return SpotifyRedirectURIHandler def block_thread_until_spotify_responds(server): circuit_breaker = threading.Timer(available_duration_for_login_in_seconds, send_http_request_to_self) circuit_breaker.start() server.handle_request() # blocking happens here circuit_breaker.cancel() def send_http_request_to_self(): try: urllib.request.urlopen("http://localhost:{}/".format(str(redirect_port))).read() except: pass def shut_down_server(server): server.socket.close() get_redirect_response()
