Tornado路由與Application的實現(xiàn)
路由原理
在Tornado框架中，路由是指將請求的URL映射到對應的處理函數(shù)上，這個過程需要通過正則表達式來實現(xiàn)。Tornado使用了一種叫做Application的類來封裝整個Web應用程序.我們定制一個Applicaiton繼承tornado的Application
from tornado.web import Application as tornadoApp
class Application(tornadoApp):
? ? """ 定制 Tornado Application 集成日志、sqlalchemy 等功能 """
? ? def __init__(self):
? ? ? ? self.ops_mongo = None ?# ops mongo
? ? ? ? self.redis_cluster = None ?# redis基礎(chǔ)session
? ? ? ? self.redis_manager = None ?# redis_manager session
? ? ? ? self.redis = None
? ? ? ? tornado_settings = {
? ? ? ? ? ? 'autoreload': settings.service.server.autoreload,
? ? ? ? ? ? 'debug': settings.service.server.debug,
? ? ? ? ? ? "default_handler_class": DefaultHandler,
? ? ? ? }
? ? ? ? route = getRoutes(settings)
? ? ? ? self.prepare()
? ? ? ? super(Application, self).__init__(handlers=route, **tornado_settings)
? ? def prepare(self):
? ? ? ? self.redis_cluster = asyncio.get_event_loop().run_until_complete(
? ? ? ? ? ? init_cluster()
? ? ? ? )
? ? ? ? self.ops_mongo = asyncio.get_event_loop().run_until_complete(
? ? ? ? ? ? init_ops_mongodb()
? ? ? ? )
? ? ? ? loop = asyncio.get_event_loop()
? ? ? ? self.redis = RedisPool(loop=loop).get_conn()
? ? ? ? self.redis_manager = RedisManager(self.redis)
? ? ? ? asyncio.get_event_loop().run_until_complete(
? ? ? ? ? ? create_all_indexes()
? ? ? ? )
? ? async def on_close(self, server_conn: object) -> None:
? ? ? ? await self.redis_manager.close()
? ? ? ? logger.info("close redis")
? ? ? ? self.ops_mongo.close()
? ? ? ? logger.info("close mongo")這段代碼定義了一個繼承自 Tornado Application 的自定義 Application 類，并在初始化中進行了一些準備工作。
首先，定義了幾個實例變量，包括 ops_mongo（mongoDB 數(shù)據(jù)庫連接）、redis_cluster（基于 redis 的 session）、redis_manager（管理 redis 連接）和 redis（redis 數(shù)據(jù)庫連接）。
然后，根據(jù)配置文件中的設(shè)置，設(shè)定了 Tornado Application 的一些參數(shù)，并根據(jù)路由設(shè)置得到路由列表。
接下來，調(diào)用了 prepare 方法，該方法中通過異步方法初始化了 ops_mongo（mongoDB 數(shù)據(jù)庫連接）、redis_cluster（基于 redis 的 session）和 redis（redis 數(shù)據(jù)庫連接），并創(chuàng)建了 redis_manager 實例管理 redis 連接。最后，調(diào)用了異步方法 create_all_indexes，創(chuàng)建了所有的索引。
最后，定義了一個異步方法 on_close，在 Web 服務(wù)關(guān)閉時關(guān)閉了 redis_manager 實例和 ops_mongo 實例。同時加入了一些對于關(guān)閉的日志信息。
class DefaultHandler(tornado.web.RequestHandler):
? ? def get(self):
? ? ? ? raise tornado.web.HTTPError(
? ? ? ? ? ? status_code=404,
? ? ? ? ? ? reason="Not Found"
? ? ? ? )
? ? def write_error(self, status_code, exc_info=None, **kwargs):
? ? ? ? self.finish({"error": self._reason})
def getRoutes(setting):
? ? Routes = [
? ? ? ? (r"/redis", home.TestRedisHandler),
? ? ]
? ? return Routes該代碼是使用 Tornado 框架定義一個繼承自 tornado.web.RequestHandler 的 DefaultHandler 類，包含兩個方法 get 和 write_error。
- get 方法會拋出一個 HTTPError，表示請求的頁面不存在或其他錯誤，會返回一個 404 的狀態(tài)碼和 Not Found 的原因。
- write_error 方法會在遇到未處理的異常時自動調(diào)用，可以自定義錯誤響應的方式。這里定義的行為是結(jié)束響應并返回一個包含錯誤信息的 JSON 對象
另外，getRoutes 方法定義了一個以元組形式表示的 URL 路由表，將 /redis 映射到名為 TestRedisHandler 的類。
RequestHandler的功能
在Tornado中，RequestHandler是處理每個HTTP請求的基礎(chǔ)類，所有的請求處理類都應該繼承自它。RequestHandler有如下幾個常用的方法：
- initialize()：初始化方法，在請求處理類實例化時自動調(diào)用，用于設(shè)置一些參數(shù)或進行一些初始化操作。
- prepare()：在處理請求前調(diào)用，用于進行一些安全檢查、身份驗證等操作，并且可以自定義錯誤頁面。
- get()、post()等：根據(jù)HTTP請求方式的不同，提供相應的get()、post()等方法來處理請求，并返回相應的HTTP響應。
- write()、finish()：用于返回響應數(shù)據(jù)，并結(jié)束請求處理。
- redirect()、set_header()等：提供一些常用的HTTP響應輔助方法。
class BaseHandler(RequestHandler, ABC):
????traceid = None
????start_time = None
????# 普通返回
????response_dict = {'code': 0, 'message': "", 'tid': "", 'data': None}
????# 分頁返回
????response_page_dict = {'code': 0, 'message': "", 'tid': "", 'data': None, "pagination":
????????{"pagesize": 0, "pages": 0, "items": 0, "pageno": 0}
??????????????????????????}
????@classmethod
????def return_response(cls):
????????resp = AttrDict(cls.response_dict)
????????return resp
????@classmethod
????def return_page_response(cls):
????????resp = AttrDict(cls.response_page_dict)
????????return resp
????@classmethod
????def create_page(cls, response: AttrDict, total=None, pagesize=0, pageno=None):
????????response.pagination["items"] = total
????????response.pagination["pagesize"] = pagesize
????????response.pagination["pages"] = total // pagesize if total % pagesize == 0 else total // pagesize + 1
????????response.pagination["pageno"] = pageno
????????return response
????def get_request(self):
????????request_body = None
????????if self.request.body:
????????????request_body = tornado.escape.json_decode(self.request.body)
????????return request_body
????def return_json(self, response):
????????*"""*
**返回統(tǒng)一的*json*格式結(jié)果
*****:param*** *response:*
*"""*
** ?
**self.set_header("Content-Type", "application/json; charset=UTF-8")
????????self.response_dict = json.dumps(response, default=json_util.default)
????????self.write(self.response_dict)
????def prepare(self):
????????trace_id = self.request.headers.get("tid", None)
????????self.__class__.start_time = datetime.datetime.now()
????????if not trace_id:
????????????trace_id = str(uuid.uuid4())
????????????self.request.headers["tid"] = trace_id
????????self.__class__.traceid = trace_id
????def write_error(self, status_code, exc_info=None, api_code=None, **kwargs):
????????resp = self.return_response()
????????resp.code = status_code
????????resp.tid = self.traceid
????????ex = exc_info[1]
????????if isinstance(ex, Exception):
????????????err_msg = str(ex)
????????????resp.message = err_msg
????????if isinstance(ex, tornado.web.HTTPError):
????????????err_msg = str(ex)
????????????resp.message = err_msg
????????if isinstance(ex, tornado.web.HTTPError) and ex.log_message:
????????????err_msg = str(ex.log_message)
????????????resp.message = err_msg
????????self.response_dict = json.dumps(resp, default=json_util.default)
????????self.finish(resp)
????def on_finish(self):
????????# 處理完請求后輸出日志，將請求頭信息和響應結(jié)果json化
????????end_time = datetime.datetime.now()
????????request_data = {
????????????"tid": self.request.headers.get("tid", None),
????????????"url": self.request.uri,
????????????"method": self.request.method,
????????????"remote_ip": self.request.remote_ip,
????????????"user_agent": self.request.headers["User-Agent"],
????????????"request_body": json.loads(self.request.body) if self.request.body else ""
????????}
????????response_data = {
????????????"status": self.get_status(),
????????????"response_body": json.loads(self.response_dict) if not isinstance(self.response_dict,
??????????????????????????????????????????????????????????????????????????????dict) else self.response_dict,
????????????"response_time": (end_time - self.start_time).microseconds / 1000
????????}
????????logger = get_logger("access")
????????logger.bind(**request_data).bind(**response_data).info("request success")BaseHandler 的類，它繼承了 tornado.web.RequestHandler 和 abc.ABC 兩個類。該類提供了一些基礎(chǔ)方法，可以被繼承使用來構(gòu)建業(yè)務(wù)處理類。
這個類定義了兩種返回方式：普通返回和分頁返回。普通返回返回的數(shù)據(jù)格式有 code、message、tid 和 data 四個字段，其中 code 表示請求狀態(tài)碼，message 表示請求狀態(tài)消息，tid 表示請求追蹤 id，data 表示響應數(shù)據(jù)。分頁返回在普通返回的基礎(chǔ)上加入了分頁參數(shù)信息，即 pageSize、pages、items 和 pageNo。
該類實現(xiàn)了三個類方法：return_response、return_page_response 和 create_page。return_response 返回普通返回的響應對象，return_page_response 返回分頁返回的響應對象，create_page 用于創(chuàng)建分頁響應。
該類實現(xiàn)了三個實例方法：get_request、return_json 和 write_error。get_request 方法用于獲取請求體的 json 數(shù)據(jù)，return_json 方法用于返回 json 格式的響應，write_error 方法用于處理請求出錯時返回的響應。
該類實現(xiàn)了兩個鉤子函數(shù)：prepare 和 on_finish。prepare 在請求進入處理前被調(diào)用，可以用來獲取請求頭信息等，on_finish 在請求處理完后被調(diào)用，用于記錄日志等操作。
處理錯誤
在Tornado中，異常處理同樣也是一個非常重要的問題，異常處理可以讓我們在遇到錯誤時能夠有更好的反應和處理?？梢酝ㄟ^使用@tornado.web.appadvice()裝飾器或者重寫write_error()方法來進行自定義錯誤處理。
import tornado.web
class CustomErrorHandler(tornado.web.RequestHandler):
????def write_error(self, status_code, **kwargs):
????????if self.settings.get("serve_traceback"):
????????????self.write('Some Exception occurs: {}'.format(kwargs['exc_info'][1]))
????????else:
????????????self.write('Some Exception occurs. We are on it!')
????????self.set_status(status_code)
class ErrorExample(tornado.web.RequestHandler):
????def get(self, error_printer):
????????if error_printer == "raise":
????????????raise Exception("An unknown error occurred.")
????????elif error_printer == "syntax":
????????????raise SyntaxError("There is a syntax error in your code.")
????????else:
????????????self.write("This is normal operation!")上述代碼中，使用write_error()重寫了RequestHandler的默認錯誤處理方式。當我們訪問http://localhost:8888/err?error=syntax 時，會輸出 Some Exception occurs: There is a syntax error in your code
處理完成
tornado的RequestHandler是用于請求處理的基本類，其中on_finish是RequestHandler的一個方法，可以在請求處理完成后被調(diào)用。當客戶端請求處理完成時，可以使用此方法進行一些清理工作或日志記錄等操作，并將tornado 對象釋放回底層IOLoop。在on_finish方法內(nèi)可以進行資源釋放、統(tǒng)計請求處理時間等工作。此外，在建立連接之前初始化某些對象或資源，也可以在on_finish方法中完成清理。
以下是RequestHandler的on_finish方法的聲明:
def on_finish(self) -> None: ????????""" ????????Finish this response, ending the HTTP request. ????????""" ????????pass
Tornado提供了很多的錯誤處理接口來幫助我們進行異常處理。為了使Web程序更加的健壯，我們需要盡可能的考慮到各種可能出現(xiàn)的錯誤情況，并對異常情況進行適當?shù)奶幚怼?/p>
到此這篇關(guān)于Tornado路由與Application的實現(xiàn)的文章就介紹到這了,更多相關(guān)Tornado路由與Application內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
PyQt為事件處理提供了兩種機制高級的信號與槽機制，以及低級的事件處理機制，這篇文章主要介紹了PyQt5?事件處理機制,需要的朋友可以參考下2022-11-11
這篇文章主要介紹了django中資源文件夾的引入,主要包括靜態(tài)資源文件夾的引入及媒體資源文件夾的引入，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-08-08
pd.to_datetime中時間object轉(zhuǎn)換datetime實例
本文主要介紹了pd.to_datetime中時間object轉(zhuǎn)換datetime實例，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-06-06
在之前的編程中，我們的信息打印，數(shù)據(jù)的展示都是在控制臺（命令行）直接輸出的，信息都是一次性的沒有辦法復用和保存以便下次查看，今天我們將學習Python的輸入輸出，解決以上問題2019-04-04
這篇文章主要介紹了如何在Python?中獲取單成員集合中的唯一元素,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-03-03
這篇文章主要介紹了python基礎(chǔ)之Socket套接字詳解,文中有非常詳細的代碼示例,對正在學習python基礎(chǔ)的小伙伴們有很好地幫助,需要的朋友可以參考下2021-04-04
這篇文章主要介紹了python計算圓周長、面積、球體體積并畫出圓(python3+PyObject+Gtk實現(xiàn)界面聯(lián)動),需要的朋友可以參考下2014-04-04
PyTorch、torchvision和Python版本的對應關(guān)系
使用PyTorch時,選擇合適的Python版本是至關(guān)重要的,錯誤的版本組合可能導致各種兼容性問題,本文就來介紹一下PyTorch、torchvision與Python版本匹配,感興趣的可以了解一下2024-03-03
這篇文章主要介紹了FastAPI到底用不用async問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-06-06
基于Python實現(xiàn)Windows桌面定時提醒休息程序
這篇文章為大家詳細主要介紹了如何基于Python實現(xiàn)簡單的Windows桌面定時提醒休息程序,文中的示例代碼講解詳細,有需要的可以參考一下2024-11-11