我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.escape()。
def set_status(self, status_code, reason=None): """Sets the status code for our response. :arg int status_code: Response status code. If ``reason`` is ``None``, it must be present in `httplib.responses <http.client.responses>`. :arg string reason: Human-readable reason phrase describing the status code. If ``None``, it will be filled in from `httplib.responses <http.client.responses>`. """ self._status_code = status_code if reason is not None: self._reason = escape.native_str(reason) else: try: self._reason = httputil.responses[status_code] except KeyError: raise ValueError("unknown status code %d", status_code)
def post(self): """ Example Handler for an `HTTP POST <http://en.wikipedia.org/wiki/POST_(HTTP)>`_ request. Doesn't actually do anything. """ # If data is POSTed to this handler via an XMLHTTPRequest send() it # will show up like this: #posted_as_a_whole = self.request.body # xhr.send() # If data was POSTed as arguments (i.e. traditional form) it will show # up as individual arguments like this: #posted_as_argument = self.get_argument("arg") # Form elem 'name="arg"' # This is how you can parse JSON: #parsed = tornado.escape.json_decode(posted_as_an_argument) json_output = {'result': 'Success!'} self.write(json_output) # You'd put self.finish() here if post() was wrapped with tornado's # asynchronous decorator. # WebSocket actions (aka commands or "functions that are exposed")
def get_upload_text_data(handler): try: text_data = {} text_data['result'] = 'fail' text_data['con'] = tornado.escape.xhtml_escape(handler.get_argument('con')).replace('&nbsp',' ').replace('<br>','<br>') text_data['title1'] = tornado.escape.xhtml_escape(handler.get_argument('title1')).replace('&nbsp',' ').replace('<br>','<br>') text_data['title2'] = tornado.escape.xhtml_escape(handler.get_argument('title2')).replace('&nbsp',' ').replace('<br>','<br>') text_data['description'] = tornado.escape.xhtml_escape(handler.get_argument('description')).replace('<br>','<br>').replace('&nbsp',' ') text_data['year'] = tornado.escape.xhtml_escape(handler.get_argument('year')) text_data['month'] = tornado.escape.xhtml_escape(handler.get_argument('month')) text_data['background_color'] = tornado.escape.xhtml_escape(handler.get_argument('background_color')) text_data['result'] = 'success' return text_data except Exception as e: print(str(e)) return text_data
def write(self, chunk): """Writes the given chunk to the output buffer. To write the output to the network, use the flush() method below. If the given chunk is a dictionary, we write it as JSON and set the Content-Type of the response to be ``application/json``. (if you want to send JSON as a different ``Content-Type``, call set_header *after* calling write()). Note that lists are not converted to JSON because of a potential cross-site security vulnerability. All JSON output should be wrapped in a dictionary. More details at http://haacked.com/archive/2008/11/20/anatomy-of-a-subtle-json-vulnerability.aspx """ if self._finished: raise RuntimeError("Cannot write() after finish(). May be caused " "by using async operations without the " "@asynchronous decorator.") if isinstance(chunk, dict): chunk = escape.json_encode(chunk) self.set_header("Content-Type", "application/json; charset=UTF-8") chunk = utf8(chunk) self._write_buffer.append(chunk)
def set_status(self, status_code, reason=None): """Sets the status code for our response. :arg int status_code: Response status code. If ``reason`` is ``None``, it must be present in `httplib.responses <http.client.responses>`. :arg string reason: Human-readable reason phrase describing the status code. If ``None``, it will be filled in from `httplib.responses <http.client.responses>`. """ self._status_code = status_code if reason is not None: self._reason = escape.native_str(reason) else: try: self._reason = httputil.responses[status_code] except KeyError: raise ValueError("unknown status code %d" % status_code)
def test_linkify(self): for text, kwargs, html in linkify_tests: linked = tornado.escape.linkify(text, **kwargs) self.assertEqual(linked, html)
def test_escape_return_types(self): # On python2 the escape methods should generally return the same # type as their argument self.assertEqual(type(xhtml_escape("foo")), str) self.assertEqual(type(xhtml_escape(u("foo"))), unicode_type)
def __call__(self, request): data = {} response = [] def start_response(status, response_headers, exc_info=None): data["status"] = status data["headers"] = response_headers return response.append app_response = self.wsgi_application( WSGIContainer.environ(request), start_response) try: response.extend(app_response) body = b"".join(response) finally: if hasattr(app_response, "close"): app_response.close() if not data: raise Exception("WSGI app did not call start_response") status_code, reason = data["status"].split(' ', 1) status_code = int(status_code) headers = data["headers"] header_set = set(k.lower() for (k, v) in headers) body = escape.utf8(body) if status_code != 304: if "content-length" not in header_set: headers.append(("Content-Length", str(len(body)))) if "content-type" not in header_set: headers.append(("Content-Type", "text/html; charset=UTF-8")) if "server" not in header_set: headers.append(("Server", "TornadoServer/%s" % tornado.version)) start_line = httputil.ResponseStartLine("HTTP/1.1", status_code, reason) header_obj = httputil.HTTPHeaders() for key, value in headers: header_obj.add(key, value) request.connection.write_headers(start_line, header_obj, chunk=body) request.connection.finish() self._log(status_code, request)
def environ(request): """Converts a `tornado.httputil.HTTPServerRequest` to a WSGI environment. """ hostport = request.host.split(":") if len(hostport) == 2: host = hostport[0] port = int(hostport[1]) else: host = request.host port = 443 if request.protocol == "https" else 80 environ = { "REQUEST_METHOD": request.method, "SCRIPT_NAME": "", "PATH_INFO": to_wsgi_str(escape.url_unescape( request.path, encoding=None, plus=False)), "QUERY_STRING": request.query, "REMOTE_ADDR": request.remote_ip, "SERVER_NAME": host, "SERVER_PORT": str(port), "SERVER_PROTOCOL": request.version, "wsgi.version": (1, 0), "wsgi.url_scheme": request.protocol, "wsgi.input": BytesIO(escape.utf8(request.body)), "wsgi.errors": sys.stderr, "wsgi.multithread": False, "wsgi.multiprocess": True, "wsgi.run_once": False, } if "Content-Type" in request.headers: environ["CONTENT_TYPE"] = request.headers.pop("Content-Type") if "Content-Length" in request.headers: environ["CONTENT_LENGTH"] = request.headers.pop("Content-Length") for key, value in request.headers.items(): environ["HTTP_" + key.replace("-", "_").upper()] = value return environ
def write(self, chunk): """Writes the given chunk to the output buffer. To write the output to the network, use the flush() method below. If the given chunk is a dictionary, we write it as JSON and set the Content-Type of the response to be ``application/json``. (if you want to send JSON as a different ``Content-Type``, call set_header *after* calling write()). Note that lists are not converted to JSON because of a potential cross-site security vulnerability. All JSON output should be wrapped in a dictionary. More details at http://haacked.com/archive/2009/06/25/json-hijacking.aspx/ and https://github.com/facebook/tornado/issues/1009 """ if self._finished: raise RuntimeError("Cannot write() after finish()") if not isinstance(chunk, (bytes, unicode_type, dict)): message = "write() only accepts bytes, unicode, and dict objects" if isinstance(chunk, list): message += ". Lists not accepted for security reasons; see http://www.tornadoweb.org/en/stable/web.html#tornado.web.RequestHandler.write" raise TypeError(message) if isinstance(chunk, dict): chunk = escape.json_encode(chunk) self.set_header("Content-Type", "application/json; charset=UTF-8") chunk = utf8(chunk) self._write_buffer.append(chunk)
def xsrf_form_html(self): """An HTML ``<input/>`` element to be included with all POST forms. It defines the ``_xsrf`` input value, which we check on all POST requests to prevent cross-site request forgery. If you have set the ``xsrf_cookies`` application setting, you must include this HTML within all of your HTML forms. In a template, this method should be called with ``{% module xsrf_form_html() %}`` See `check_xsrf_cookie()` above for more information. """ return '<input type="hidden" name="_xsrf" value="' + \ escape.xhtml_escape(self.xsrf_token) + '"/>'
def render(self, text, **kwargs): return escape.linkify(text, **kwargs)
def reverse(self, *args): assert self._path is not None, \ "Cannot reverse url regex " + self.regex.pattern assert len(args) == self._group_count, "required number of arguments "\ "not found" if not len(args): return self._path converted_args = [] for a in args: if not isinstance(a, (unicode_type, bytes)): a = str(a) converted_args.append(escape.url_escape(utf8(a), plus=False)) return self._path % tuple(converted_args)
def example_opt_esc_handler(self, message, term=None, multiplex=None): """ Gate One includes a mechanism for plugins to send messages from terminal programs directly to plugins written in Python. It's called the "Special Optional Escape Sequence Handler" or SOESH for short. Here's how it works: Whenever a terminal program emits, "\\x1b]_;" it gets detected by Gate One's :class:`~terminal.Terminal` class (which lives in `terminal.py`) and it will execute whatever callback is registered for SOESH. Inside of Gate One this callback will always be :func:`gateone.TerminalWebSocket.esc_opt_handler`. """ message = {'go:notice': "You just executed the Example plugin's optional escape sequence handler!"} self.write_message(message)
