How to use tornado - 10 common examples

def make_app():
    basedir = "/".join(os.path.abspath(__file__).split("/")[:-1])
    return tornado.web.Application([
        (r"/", MainHandler,),
        (r"/robots.txt", RobotHandler,),
        (r"/page", PageHandler,),
        (r"/page2", PageHandler,),
        (r"/page3", PageHandler,),
        (r"/static/(.*)", tornado.web.StaticFileHandler, {'path': basedir}),
    ])

def test_upload(self):
        image_bits = open(self.png_file_path(), 'rb').read()
        filename = 'a / b % 2.png'  # Test escaping weird chars.
        response = self.fetch_rpc(
            'metaWeblog.newMediaObject',
            (
                1,  # Blog id, always 1.
                tornado_options.user,
                tornado_options.password,
                {
                    'name': filename,
                    'bits': xmlrpclib.Binary(image_bits),
                    'type': 'image/png'}))

        # Make sure we can now fetch the image.
        self.http_client.fetch(response['url'], self.stop)
        response = self.wait()
        self.assertEqual(200, response.code)
        self.assertEqual('image/png', response.headers['Content-Type'])
        self.assertEqual(image_bits, response.body)

def test_adding_video_makes_it_show_up_in_friends_shake(self):
        user2 = User(name='user2', email='user2@mltshp.com', email_confirmed=1, is_paid=1)
        user2.set_password('asdfasdf')
        user2.save()
        user2.subscribe(self.user.shake())

        url = 'https://vimeo.com/20379529'
        request = HTTPRequest(self.get_url('/tools/save-video'), 'POST', {"Cookie":"sid=%s;_xsrf=%s" % (self.sid, self.xsrf)}, "url=%s&_xsrf=%s" % (url_escape(url), self.xsrf))
        self.http_client.fetch(request, self.stop)
        response = self.wait()
        sfs = Sharedfile.from_subscriptions(user2.id)
        self.assertTrue(len(sfs) > 0)
        self.assertEqual(sfs[0].name , url)

#!/usr/bin/python
# -*- coding: utf-8 -*-

import sys
import tornado.ioloop
import tornado.web


class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Hello, world")


application = tornado.web.Application([
    (r"/", MainHandler),
])


if __name__ == "__main__":
    port = int(sys.argv[1])
    application.listen(port, '0.0.0.0')
    print ">> Website running at http://0.0.0.0:%d" % port
    tornado.ioloop.IOLoop.instance().start()

            @gen.coroutine
            def delete(self, id_):
                yield self.sleep()
                return super().delete(id_)

    @gen.coroutine
    def tearDown(self):
        raise gen.Return(None)

def get_app(self):
        if not self.application:
            self.application = Application(
                [(r'/', TestHandler)]
            )

        return self.application

@tornado.gen.coroutine
def make_websocket_connection(host, port):
    global conn_try_count
    url = "ws://%s:%d/ws" % (host, port)
    conn_try_count += 1
    ws = yield tornado.websocket.websocket_connect(url)
    conn_try_count -= 1
    ws.stream.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    connections.add(ws)
    loop_websocket(ws)

    @tornado.gen.coroutine
    def post(self):
        span = self._get_span()
        if span:
            try:
                with span_in_stack_context(span):
                    res = self.client_channel.json(
                        service='handler2',
                        hostport=self.request.body,
                        endpoint="endpoint2",
                    )
                res = yield res
                body = res.body
            except Exception as e:
                traceback.print_exc()
                self.write('ERROR: %s' % e)
                self.set_status(200)

domain_url = 'http://globo.com'

        key1 = 'limit-for-%s' % domain_url
        self.cache.redis.delete(key1)

        key2 = 'limit-for-%s/sa/' % domain_url
        self.cache.redis.delete(key2)

        yield Task(self.cache.redis.zadd, key1, {'a': 1})
        yield Task(self.cache.redis.zadd, key2, {'b': 1})

        keys = yield Task(self.cache.redis.keys, 'limit-for-%s*' % domain_url)
        expect(keys).to_length(2)

        yield Task(self.cache.delete_limit_usage_by_domain, domain_url)
        keys = yield Task(self.cache.redis.keys, 'limit-for-%s*' % domain_url)
        expect(keys).to_length(0)