python 异步编程 第四章 Tornado

1.1 tcp 服务器

import asyncio, timefrom tornado.iostream import StreamClosedErrorfrom tornado.tcpserver import TCPServerfrom tornado.ioloop import IOLoopclass EchoServer(TCPServer): async def handle_stream(self, stream, address): for i in range(6): try: await stream.write(f"[{time.strftime('%X')}] Count {i}\n".encode("utf-8")) await asyncio.sleep(1) except StreamClosedError: breakserver = EchoServer()server.listen(8888)try: IOLoop.current().start()except KeyboardInterrupt as e: print("Exit")

1.2 http 服务器

import tornado.ioloopimport tornado.webclass HomePage(tornado.web.RequestHandler): def get(self): self.write("hello world") # def post(sefl): # self.write("handle post request")if __name__ == '__main__': app = tornado.web.Application([ ('/', HomePage) ])

1.3 路由

import tornado.ioloopimport tornado.webimport tornado.routingclass HomePage(tornado.web.RequestHandle): def get(self): self.write("Home")class Users(tornado.web.RequestHandle): def get(self): self.wirte("Users")class AppPage(tornado.web.RequestHandler): def get(self): self.write(f"Requst path is: {self.request.path}")if __name__ == '__main__': app = tornado.web.Application([ ("/", HomePage), ("/user", Users), # 使用正则 (r"/app.*", AppPage) ]) app.listen(8888) tornado.ioloop.IOLoop.current().start()

路径参数

import tornado.ioloopimport tornado.webimport tornado.routingclass AppPage(tornado.web.RequestHandler): def get(self, path_arg1, path_arg2): self.write(f"Arg1 is: {path_arg1}, arg2 is: {path_arg2}")if __name__ == '__main__': app = tornado.web.Application([ (r"/app/(.*)/(.*)", AppPage) ]) app.listen(8888) tornado.ioloop.IOLoop.current().start()

多个应用

import tornado.ioloopfrom tornado.web import Applicationfrom tornado.routing import Rule, RuleRouter, PathMatchesclass HandleInApp1(tornado.web.RequestHandler): def get(self): self.write("Handle in app1")app1 = Application([ (r"/app1/handler", HandlerInApp1)])class HandleInApp2(tornado.web.RequestHandler): def get(self): self.write("Handle in app2")app2 = Application([ (r"/app2/handler", HandlerInApp2)])if __name__ == '__main__': server = tornado.web.HTTPServer( RuleRouter([ Rule(PathMatches(r"/app1.*"), app1), Rule(PathMatches(r"/app2.*"), app2) ]) ) server.listen(8888) try: tornado.ioloop.IOLoop.current().start() except KeyboardInterrupt: print("Server stopped by user")

1.4 处理静态文件

from tornado.web import Application, StaticFileHandlerfrom tornado.ioloop import IOLoopimport osAPP_ROOT = os.path.dirname(__file__)if __name__ == '__main__': app = Application([ r"/static/(.*)", StaticFileHandler, dict(path=os.path.join(APP_ROOT, "static")) ]) app.listen(8888) IOLoop.current().start()

index.html

静态文件

1.5 模版渲染

import tornado.ioloopimport tornado.webclass IndexHandler(tornado.web.RequestHandler): async def get(self): await self.render("index.html")if __name__ == '__main__': application = tornado.web.Application([ (r"/", IndexHandler) ]) tornado.ioloop.IOLoop.current().start()

index.html

静态页面

加载模板目录

import tornado.ioloopimport tornado.webimport osSERVER_ROOT = os.path.dirname(__file__)class IndexHandler(tornado.web.RequestHandler): async def get(self): await self.render("index.html")if __name__ == '__main__': application = tornado.web.Application([ (r"/", IndexHandler), ], template_path = os.path.join(SERVER_ROOT, "template")) application.listen(8888) tornado.ioloop.IOLoop.current().start()

layout.html

<!DOCTYPE html><html><head><title> Title </title></head><body> <!-- 与 jinja2 不一样，以 end 结尾 --> {% block body %}{% end %}</body></html>

index.html

{% extends "layout.html" %}{% block body %} hello world {% end %}

模板参数

import tornado.ioloopimport tornado.webimport osSERVER_ROOT = os.path.dirname(__file__)class IndexHandler(tornado.web.RequestHandler): async def get(self): await self.render("server2_index.html", user=['张三', '李四'])if __name__ == '__main__': application = tornado.web.Application([ (r"/", IndexHandler), ],template_path=os.path.join(SERVER_ROOT, "template")) application.listen(8888) tornado.ioloop.IOLoop.current().start()

server2_index.html

{% extends "layout.html" %}{% block body %}<div> Users: <ul> {% for u in users %} <li>{{ u }}</li> {% end %} </url></div>{% end %}

1.6 HTTP 客户端

server.py

import tornado.ioloopimport tornado.webclass MainHandler(tornado.web.RequestHandler): def get(self): name = self.get_query_argument("name","") self.write(f"hello {name}") def post(self): name = self.get_body_argument("name", "") self.write(f"hello {name}")if __name__ == '__main__': application = tornado.web.Application([ (r"/", MainHandler), ]) application.listen(8888) tornado.ioloop.IOLoop.current().start()

client get 请求

import tornado.httpclientimport asyncioclient = tornado.httpclient.AsyncHTTPClient()async def main(): resp = await client.fetch("http://127.0.0.1:8888?name=xiaoming") print(resp.body.decode("utf-8"))asyncio.run(main())

client post 请求

import tornado.httpclientimport asyncioclient = tornado.httpclient.AsyncHTTPClient()async def main(): resp = await client.fetch( tornado.httpclient.HTTPRequest( "http://127.0.0.1:8888", "POST", body="name=xiaoming" ) ) print(resp.body.decode("utf-8"))asyncio.run(main())