版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/hzy459176895/article/details/84062967
之前讲了tornado简单案例( get方法)。
个人理解:一般地,get方法只是为了通过url传参(不是传json数据),服务器作出响应,get可以向服务器获取到数据(想要的数据)。而post方法可以向服务器传入一个数据进行处理(你想让服务器对数据做什么,达到你的目的),它的数据是放在request body中需要获取的。
下面介绍一个简单的tornado案例(post方法的)------>
1. service.py (开启服务)
这里设定了端口号,指定了要执行哪一个handler处理器,处理响应。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @author: 小何
import tornado.web
from tornado.web import URLSpec
from tornado.ioloop import IOLoop
from scpy.logger import get_logger
from handler import PostHandler
logger = get_logger(__file__)
HANDLERS = [
URLSpec(r'/example', PostHandler,
name=PostHandler.__name__)
]
if __name__ == '__main__':
SERVER_PORT = 2333
app = tornado.web.Application(handlers=HANDLERS, debug=True)
app.listen(SERVER_PORT)
logger.info("Clean Panel server started on port {SERVER_PORT}".format(SERVER_PORT=SERVER_PORT))
IOLoop.current().start()
2. handler.py (处理)
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @author: 小何
import sys
from controller_example import ControllerExample
reload(sys)
sys.setdefaultencoding('utf-8')
import abc
import json
import tornado.web
from tornado import gen
from tornado.web import HTTPError
from tornado.concurrent import run_on_executor
from concurrent.futures import ThreadPoolExecutor
import scpy.logger
logger = scpy.logger.get_logger(__file__)
NUMBER_OF_EXECUTOR = 6
'''
handler --- post方法测试
'''
class BasePostRequestHandler(tornado.web.RequestHandler):
executor = ThreadPoolExecutor(NUMBER_OF_EXECUTOR)
@tornado.web.asynchronous
@gen.coroutine
def post(self, *args, **kwargs):
try:
result = yield self._post(*args, **kwargs)
self.write(result)
except HTTPError as e:
self.write(e)
except Exception as e:
logger.error(e)
raise HTTPError(404, "No results")
@run_on_executor
def _post(self, *args, **kwargs):
request = self._post_request_arguments(*args, **kwargs)
res = self._request_service(**request)
return res
@abc.abstractmethod
def _post_request_arguments(self, *args, **kwargs):
raise NotImplementedError('call to abstract method %s._get_request_arguments' % self.__class__)
@abc.abstractmethod
def _request_service(self, **kwargs):
raise NotImplementedError('call to abstract method %s._request_service' % self.__class__)
class PostHandler(BasePostRequestHandler):
def _post_request_arguments(self, *args, **kwargs):
'''
获取数据
:param args:
:param kwargs:
:return:
'''
logger.info(self.__class__.__name__)
data = json.loads(self.request.body)
if not data:
raise HTTPError(400, "Query argument cannot be empty string")
return data
def _request_service(self, **kwargs):
'''
处理数据
:param kwargs:
:return:
'''
if kwargs:
c_e = ControllerExample(kwargs)
res = c_e.do_add()
else:
raise HTTPError(400, "Query argument cannot be empty string")
return res
这里边基类中,实现了post请求方法,postHandler实现了获取数据,并将数据传入Controller业务层进行处理。
3 .controller_example.py (业务执行层)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @author : he
class ControllerExample(object):
'''
控制层业务举例
'''
def __init__(self, data):
self.test_data = data
def do_add(self):
self.test_data['aaa'] = self.test_data.get('aaa') + 100
self.test_data['bbb'] = self.test_data.get('bbb') + 100
return self.test_data
这里边执行了一点简单的add操作
4. 测试整个环节 --- test_service.py (一般get方法采用浏览器地址栏输入url访问,post做不到,会报405err,post方法只有使用postman工具查看,或者写脚本测试,我选择了后者)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @createTime : 18-6-19 17:00
# @author : 小何
from unittest import TestCase
import requests
import json
class TestPostHandler(TestCase):
def test_handler_result(self):
content = {'aaa': 123, 'bbb': 456}
json_data = json.dumps(content)
resp = requests.post('http://localhost:2333/example', json_data)
print resp.json()
5. 测试效果,访问通过,并成功执行了业务层中的简单操作!!!
整个tornado简单的过程,也算个经典而简洁的MVC结构了。。。。。。。。。。。。