标签: tornado

是否有可能构造一个正则表达式来捕获所有出现的由字符序列分隔的文本？换句话说，我正在寻找一种等效的标准.split()方法。

我不能使用split()，因为正则表达式用于指定Tornado Web应用程序的URL路由。例如：

handlers = [
    (r'/posts/([0-9a-zA-Z_\-]+)', PostsHandler),
]

当为在Tornado，Django或任何其他实现Routes模式的Web框架上构建的Web应用程序指定URL路由时，这样的正则表达式会派上用场。特别是将未知长度的URL路径解析为参数列表。

到目前为止，我设法提出了以下正则表达式：

/^\/posts(?:\/([a-zA-Z0-9_\-]+))+/

不幸的是，当表达式匹配时/posts/show/some-slug/15，它仅返回最后一个匹配组（15），而不是['show', 'some-slug', '15']。

我想要实现的是：

• /posts/edit/15/ => ['edit', '15']
• /posts/edit/15 => ['edit', '15']
• /posts/2010/15/11 => ['2010', '15', '11']

最后决定使用Tornado作为WebSocket服务器,但我对如何实现它有疑问.

在完成了有关创建工作服务器的基本教程之后,我最终得到了以下内容:

#!/usr/bin/env python

from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.web import Application
from tornado.websocket import WebSocketHandler

class Handler(WebSocketHandler):
        def open(self):
            print "New connection opened."

        def on_message(self, message):
                print message


        def on_close(self):
                print "Connection closed."

print "Server started."
HTTPServer(Application([("/", Handler)])).listen(1024)
IOLoop.instance().start()

它的伟大工程和所有的,但如果实际需要的其他模块(tornado.httpserver,tornado.ioloop和tornado.web)来运行服务器我想知道.

拥有它们并不是一个大问题,但我只是想确保没有更好的方法来做任何事情(我还没有完全覆盖这些模块).

Tornado使用派生自tornado.web.RequestHandler定义post和get方法的类.我已经为路径创建了一个请求处理程序r/action/*

由于存在不同的更具体的路径,action/doThis因此action/requestThat我get使用if-then-else结构检查具体请求URI.

这看起来非常笨拙,有没有更复杂的方法来构建,例如使用方法？或者Tornado方式是否只是定义更多的RequestHandlers？我喜欢使用函数装饰器处理瓶子的方式,但由于它们是Python 3,因此目前不是一个选项.

我正在尝试使用龙卷风编写一个应用程序,使用gunicorn处理工作线程.我创建了下面显示的代码,但是尽管启动了多个工作程序,但它并没有共享请求.一名工人似乎一直处理所有请求(不是间歇性的).

码:

from tornado.web import RequestHandler, asynchronous, Application
from tornado.ioloop import IOLoop

import time
from datetime import timedelta
import os

class MainHandler(RequestHandler):
    def get(self):
        print "GET start"
        print "pid: "+str(os.getpid())
        time.sleep(3)
        self.write("Hello, world.<br>pid: "+str(os.getpid()))
        print "GET finish"

app = Application([
    (r"/", MainHandler)
])

在控制台中输出(我在3秒窗口内轻松刷新了3个浏览器选项卡,但它们仍然使用相同的进程并按顺序运行):

2014-04-12 20:57:52 [30465] [INFO] Starting gunicorn 18.0
2014-04-12 20:57:52 [30465] [INFO] Listening at: http://127.0.0.1:8000 (30465)
2014-04-12 20:57:52 [30465] [INFO] Using worker: tornado
2014-04-12 20:57:52 [30474] [INFO] Booting worker with pid: 30474
2014-04-12 20:57:52 [30475] [INFO] Booting …

我正在尝试做一个非常简单的示例，该示例使用由Python函数使用Tornado处理的ajax（POST）提交表单。

在工作期间，我可以返回响应，我的主要目标是返回一些数据，而无需重新加载新的/相同的页面。只传回一些数据，这些数据应该由jQuery处理。

这是我的Python代码：

class Application(tornado.web.Application):
    def __init__(self):
        handlers = [
            (r"/(happiness)", Happiness),
        ]
        settings = {
            "template_path": Settings.TEMPLATE_PATH,
            "static_path": Settings.STATIC_PATH,
        }

        tornado.web.Application.__init__(self, handlers, **settings)

class Happiness(tornado.web.RequestHandler):
    def get(self, call):
        resp = valid_calls[call]
        if resp:
            template, ctx = resp()
            self.render(template, ctx=ctx)

    def post(self, source):
        text = self.get_argument('message')
        self.write(text)

这是我的jQuery代码：

$(function() {
    var text = ''
    $('.error').hide();
    $(".add_post").click(function() {
        text = $("input#message").val();
        if (text == "") {
            $("label#name_error").show();
            $("input#text").focus();
            return false;
        }

        $.ajax({
            type: "POST",
            data: "message=" + text,
            success: function(data) { …

我正在尝试为我的龙卷风应用实施Google Oauth 2.0登录.他们用3.2.2做了一些改动,他们似乎没有明确的指示.以下是我的代码:

class BaseHandler(tornado.web.RequestHandler):
    def get_current_user(self):
        user = self.get_secure_cookie('trakr')
        if not user: return None
        return True

class ProductsHandler(BaseHandler):
    @tornado.web.authenticated
    def get(self):
        self.render("products.html")
        return

class GAuthLoginHandler(BaseHandler, tornado.auth.GoogleOAuth2Mixin):
    @tornado.gen.coroutine
    def get(self):
        if self.get_current_user():
            self.redirect('/products')
            return

        if self.get_argument('code', False):
            user = yield self.get_authenticated_user(redirect_uri=settings.google_redirect_url,
                code=self.get_argument('code'))
            if not user:
                self.clear_all_cookies() 
                raise tornado.web.HTTPError(500, 'Google authentication failed')

            access_token = str(user['access_token'])
            http_client = self.get_auth_http_client()
            http_client.fetch('https://www.googleapis.com/oauth2/v1/userinfo?access_token='+access_token, self._save_user_profile)
            return

        elif self.get_secure_cookie('trakr'):
            self.redirect('/products')
            return

        else:
            yield self.authorize_redirect(
                redirect_uri=settings.google_redirect_url,
                client_id=self.settings['google_oauth']['key'],
                scope=['email'],
                response_type='code',
                extra_params={'approval_prompt': 'auto'})

    def _save_user_profile(self, response):
        if not response:
            raise …

我有写在一个WebService 瓶,包裹在一个WSGIContainer由担任龙卷风利用其FallbackHandler机制.我在flask webservice中的一个路由运行一个非常长的操作(大约需要5分钟完成),当触发此路由时,阻止对任何路由的每个其他调用,直到操作完成.我该如何解决这个问题？

以下是使用Tornado提供Flask应用程序的方法:

parse_command_line()

    frontend_path = os.path.join(os.path.dirname(__file__),"..","webapp")

    rest_app = WSGIContainer(app)
    tornado_app = Application(
        [
            (r"/api/(.*)", FallbackHandler, dict(fallback=rest_app)),
            (r"/app/(.*)", StaticFileHandler, dict(path=frontend_path))
        ]
    )

如何在if/else里面获取一个函数__init__:

class Foo(object):
    def __init__(self, q, **keywords):
        if a == "":
            print "No empty strings"
        else:
            def on_g(self, response):
                if response.error:
                    print "Check your internet settings"
                else:
                    self.Bar()
            http_client.fetch("http://www.google.com/", self.on_g)

因为该程序不读取on_g()如果我把一个空字符串!

如果我使用on_g()外部并行,__init__()我需要一个声明的变量,例如:

class Foo(object):
    def __init__(self, q, **keywords):
        if a == "":
            print "No empty strings"
        else:
            self.on_g()
   def on_g(self):
       print 'hello there'

将返回 hello there

我有以下代码:

  try:
     responses = yield [httpClient.fetch(url) for url in urls]
  except (HTTPError, IOError, ValueError) as e:
     print("caught")

我不能保证给出的网址是有效的.我希望能够使用该例外来验证网址.如何判断哪个URL在捕获的异常中失败？

此外,如果一次提取失败(比如说第一次),它看起来就像是其他提取中断了吗？有办法防止这种情况吗？或者是否有更好的方法来检查在实际获取之前可以获取URL？对此有更好的模式吗？基本上我想并行获取所有URL并知道哪一个可能失败.

我对龙卷风还比较陌生，如果条件通过，我将尝试停止执行方法，否则应继续执行下一行代码（类似于return）

这是我的代码卡住的部分，

class PostMatchHandler(BaseRequestHandler):
    result1 = "some sql query".execute()
    if not result1:
        response.update({'info': 'Levels Completed', 'status': settings.STATUS_200})
        self.write(response)
        self.finish()

    else:
        result1 = result1[0]
        do_something()
    self.write(response)

BaseRequestHandler是：

class BaseRequestHandler(tornado.web.RequestHandler):
    """
    Base Class for all the future classes
    Activities like logging should be done in this class.
    Allow cors request in this class
    """

    def set_default_headers(self):
        print "setting headers!!!"

    def write_error(self, status_code, **kwargs):
        response = {}
        handle_error_messages..
        self.write(response)

    @staticmethod
    def extract_psycopg2_integrity_error(error):
        return error.message.split("Key")[1].replace("(", "").replace(")", "").split(".")[0].replace("=", " ")

    def prepare(self):
        logging.debug(
            "[info] Class …