概述：
Routes是python重新實作的Rails routes system，用來将urls映射到應用具體的action上。routes子產品不僅僅是在openstack中進行使用，flask架構也是對其進行了封裝。
WSGI：
Web伺服器網關接口(python Web Server Gateway Interface)是為python語言定義的web伺服器和web應用程式或架構之間的一種簡單而通用的接口。
Wsgi區分為兩個部分：伺服器和應用程式。在處理一個WSGI請求時，伺服器會為應用程式提供環境資訊及一個回調函數，當應用程式完成處理後，透過前述的回調函數，将結果回傳給伺服器。
WSGI對于app對象有以下要求：
- 必須是一個可調用的對象
- 接收兩個參數必選參數environ，start_response
- 傳回值必須是可疊代的，使用者表示http body。
class APPClass(object):
def __init__(**kwargs):
super(APPCLass, cls).__init__(**kwargs)
def function(environ, start_response):
pass
def __call__(environ, start_response):
pass environ包含請求的是以資訊，下面列舉必須包含的變量：
- REQUEST_METHOD HTTP請求方法，例如GET，POST
- SCRIPT_NAME URL路徑的起始部分對應的應用對象，如果應用程式對象對應伺服器的根，那麼這個值可以為空字元串。
- PATH_INFO URL除淋淋起始部分後的剩餘部分，用于找到對應的應用程式對象。如果請求的路徑是根路徑，這個值為空字元串。
- QUERY_STRING URL路徑中？後面的部分，GET的傳參
- CONTENT_LENGTH HTTP請求中的Content-Length部分
- SERVER_NAME,SERVER_POR
- SERVER_PROTOCOL 用戶端使用的協定，例如HTTP/1.0，HTTP/1.1，它決定了如何處理HTTP請求的頭部。
start_response是HTTP相應的開始，被調用時伺服器檢查headers中的錯誤，禁止start_response直接将response_headers傳遞給用戶端，它必須把它們存儲起來，一直到應用程式第一次疊代傳回一個非空資料後，才能将response_headers傳遞給用戶端。
start_response(status, response_headers, exec_info=None) # wsgiref工具參考文檔：https://docs.python.org/zh-tw/3.11/library/wsgiref.html
from wsgiref.util import setup_testing_defaults
from wsgiref.simple_server import make_server
# A relatively simple WSGI application. It's going to print out the
# environment dictionary after being updated by setup_testing_defaults
def simple_app(environ, start_response):
setup_testing_defaults(environ)
status = '200 OK'
headers = [('Content-type', 'text/plain; charset=utf-8')]
start_response(status, headers)
ret = [("%s: %s\n" % (key, value)).encode("utf-8")
for key, value in environ.items()]
return ret
with make_server('', 8000, simple_app) as httpd:
print("Serving on port 8000...")
httpd.serve_forever() WSGI，uWSGI，uwsgi三者的關系：
WSGI是一個Python Web應用程式與Web伺服器之間的接口規範，它定義了應用程式和伺服器之間的标準接口，使得應用程式可以再不同的Web伺服器上運作。
uWSGI是一個Web伺服器，用C語言編寫的Web應用程式容器。uWSGI伺服器可以作為一個獨立的應用伺服器，也可以與其他Web伺服器(比如Nginx，Apache)一起使用，通過WSGI協定與Python應用程式通信。
uwsgi是一個與uWSGI伺服器相關的協定，uwsgi協定是一種二進制協定，它定義了uWSGI伺服器與應用程式之間的通信協定。uwsgi協定允許uWSGI伺服器與應用程式支架進行雙向通信，進而提供了性能。
webob
Webob是一個封裝了WSGI的請求和應答的python庫。
Webob提供了多個對象，大部分都用來處理HTTP的請求，包括對HTTP頭的解析，内容的處理，建立WSGI的應答（包括HTTP的狀态，頭和body）等。Webob最重要的兩個類是Request和Response，Request主要用來構造和解析HTTP請求，而Response主要用來構造HTTP的應答。
def __call__(self, req, *args, **kw):
"""Call this as a WSGI application or with a request"""
func = self.func
if func is None:
if args or kw:
raise TypeError(
"Unbound %s can only be called with the function it "
"will wrap" % self.__class__.__name__)
func = req
return self.clone(func)
if isinstance(req, dict):
if len(args) != 1 or kw:
raise TypeError(
"Calling %r as a WSGI app with the wrong signature" %
self.func)
environ = req
start_response = args[0]
req = self.RequestClass(environ) #定義RequestClass
req.response = req.ResponseClass() # 定義ResponseClass
try:
args, kw = self._prepare_args(None, None)
resp = self.call_func(req, *args, **kw) # 回調函數
except HTTPException as exc:
resp = exc
if resp is None:
## FIXME: I'm not sure what this should be?
resp = req.response
if isinstance(resp, text_type):
resp = bytes_(resp, req.charset)
if isinstance(resp, bytes):
body = resp
resp = req.response
resp.write(body)
if resp is not req.response:
resp = req.response.merge_cookies(resp)
return resp(environ, start_response) # 傳回資料
else:
args, kw = self._prepare_args(args, kw)
return self.call_func(req, *args, **kw) 方法介紹
routes.Mapper.connect()
routes.url()
routes.url_for()
routes.Mapper.resource()
routes.Mapper.redirect()
routes.Mapper.match()
使用方法
- 注冊路由，路由名稱'zbj'， 路徑是 '/clj', controller為 'main'， action為 'index'# 比對到此條路由URL的請求：交由controller類處理，請求預調用的函數index
map.connect('zbj', '/clj', controller='main', action='index') - 比對路由，match傳回比對的資料以及controller類，routematch比match多傳回route對象。
map.connect('/home/{action:index|jia}/{id:\d+}', controller='home'，action='index')
res = map.match('/home/jia/200')
print(res) # {'action': 'index', 'id': '200', 'controller': 'home'} - 路由集合，具體參數使用方法自行百度
map.resource("message", 'message', controller=controller, path_prefix='/{project_id}',
name_prefix='lala_', collection={"list_many":'GET', 'create_many':'POST'},
member={'update_many':"POST", 'delete_many':"POST"},
new={'preview':"POST"},
parent_resource=dict(member_name='haha', collection_name='heihei')) 範例
參考openstack nova的源碼進行梳理
def _create_controller(main_controller, action_controller_list): # 将執行個體函數注冊到Resourced對象中
controller = wsgi.Resource(main_controller())
for ctl in action_controller_list: # 如果執行個體中存在被wsgi_action裝飾的函數，将該函數屬性儲存。已備調用
controller.register_actions(ctl())
return controller
ROUTE_LIST = (
('/flavors', {
'GET': [flavor_controller, 'index'],
'POST': [flavor_controller, 'create']
})
）
# 使用偏函數完成類的注冊
agents_controller = functools.partial(
_create_controller, agents.AgentController, [])
# paste加載app的入口
class APIRouterV21(base_wsgi.Router): # 繼承父類
def __init__(self, custom_routes=None):
# 初始化執行個體，将map對象傳給父類
super(APIRouterV21, self).__init__(nova.api.openstack.ProjectMapper())
for path, methods in ROUTE_LIST + custom_routes:
for method, controller_info in methods.items():
controller = controller_info[0]() # 控制類
action = controller_info[1] #動作
self.map.create_route(path, method, controller, action) # 将執行個體映射到routes.Mapper.connect()，供url映射調用
@classmethod
def factory(cls, global_config, **local_config):
return cls() # 傳回執行個體對象 class Resource(wsgi.Application):
def __init__(self, controller):
self.controller = controller
self.default_serializers = dict(json=JSONDictSerializer)
self.wsgi_actions = {}
if controller:
self.register_actions(controller)
def register_actions(self, controller): # 被wsgi_action裝飾器裝飾的執行個體函數
actions = getattr(controller, 'wsgi_actions', {}) # 如何實作該方法，可以參考元類定義
for key, method_name in actions.items():
self.wsgi_actions[key] = getattr(controller, method_name)
def get_action_args(self, request_environment): # 解析比對的參數，包含action，GET參數
if hasattr(self.controller, 'get_action_args'):
return self.controller.get_action_args(request_environment)
try:
args = request_environment['wsgiorg.routing_args'][1].copy()
except (KeyError, IndexError, AttributeError):
return {}
return args
def get_body(self, request):
content_type = request.get_content_type()
return content_type, request.body
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, request): # 調用執行個體，調用真正的應用程式
context = request.environ['nova.context']
action_args = self.get_action_args(request.environ) # 擷取get參數
action = action_args.pop('action', None)
try:
content_type, body = self.get_body(request) # 擷取body裡面的資料
accept = request.best_match_content_type()
except exception.InvalidContentType:
msg = _("Unsupported Content-Type")
return Fault(webob.exc.HTTPUnsupportedMediaType(explanation=msg))
return self._process_stack(request, action, action_args, content_type, body, accept)
def _process_stack(self, request, action, action_args, content_type, body, accept):
meth = self.get_method(request, action, content_type, body)
contents = self._get_request_content(body, request)
action_args.update(contents)
response = None
try:
with ResourceExceptionHandler():
action_result = self.dispatch(meth, request, action_args) # 回調執行個體方法，傳回傳回值
except Fault as ex:
response = ex
if not response:
resp_obj = None
if type(action_result) is dict or action_result is None:
resp_obj = ResponseObject(action_result)
elif isinstance(action_result, ResponseObject):
resp_obj = action_result
else:
response = action_result
if resp_obj:
if hasattr(meth, 'wsgi_code'):
resp_obj._default_code = meth.wsgi_code
if resp_obj and not response:
response = resp_obj.serialize(request, accept)
return response
def get_method(self, request, action, content_type, body):
meth = self._get_method(request, action, content_type, body)
return meth
def _get_method(self, request, action, content_type, body):
try:
meth = getattr(self, action) if not self.controller else getattr(self.controller, action)
return meth
except AttributeError:
raise
action_name = action_peek(body) if action == 'action' else action
return (self.wsgi_actions[action_name])
def dispatch(self, method, request, action_args):
return method(req=request, **action_args) class APIMapper(routes.Mapper):
def routematch(self, url=None, environ=None):
if url == "":
result = self._match("", environ)
return result[0], result[1]
return routes.Mapper.routematch(self, url, environ)
def connect(self, *args, **kargs):
if not kargs['requirements'].get('format'):
kargs['requirements']['format'] = 'json|xml'
return routes.Mapper.connect(self, *args, **kargs)
class ProjectMapper(APIMapper):
def create_route(self, path, method, controller, action):
project_id_token = self._get_project_id_token()
self.connect('/%s%s' % (project_id_token, path), conditions=dict(method=[method]),
controller=controller, action=action)
self.connect(path, conditions=dict(method=[method]), controller=controller,
action=action) class Request(webob.Request):
def __init__(self, environ, *args, **kwargs):
if CONF.wsgi.secure_proxy_ssl_header:
scheme = environ.get(CONF.wsgi.secure_proxy_ssl_header)
if scheme:
environ['wsgi.url_scheme'] = scheme
super(Request, self).__init__(environ, *args, **kwargs)
class Router(object):
def __init__(self, mapper):
self.map = mapper
self._router = routes.middleware.RoutesMiddleware(self._dispatch, self.map)
# 調取app，将url與執行個體的映射關系，回調函數傳給RoutesMiddleware。該函數會進行url的比對
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
return self._router #調取RoutesMiddleware
@staticmethod
@webob.dec.wsgify(RequestClass=Request)
def _dispatch(req): # 将url比對成功app進行調用，調取Resouce類的__call__函數
match = req.environ['wsgiorg.routing_args'][1]
app = match['controller'] # Resource對象，封裝真正的類對象
return app # routes.middleware.py
class RoutesMiddleware(object):
def __init__(self, wsgi_app, mapper, use_method_override=True, path_info=True, singleton=True):
self.app = wsgi_app
self.mapper = mapper
self.path_info = path_info
def __call__(self, environ, start_response):
if self.singleton: 是否是單例模式
config = request_config()
config.mapper = self.mapper
config.environ = environ
match = config.mapper_dict
route = config.route
else:
results = self.mapper.routematch(environ=environ)
if results:
match, route = results[0], results[1] # 傳回比對參數（包括action，傳入的參數），以及route對象
else:
match = route = None
url = URLGenerator(self.mapper, environ)
environ['wsgiorg.routing_args'] = ((url), match)
environ['routes.route'] = route
environ['routes.url'] = url
response = self.app(environ, start_response) # 回調函數
return response class APIMapper(routes.Mapper):
def routematch(self, url=None, environ=None): # 在RoutesMiddler中進行調用
if url == "":
result = self._match("", environ)
return result[0], result[1]
return routes.Mapper.routematch(self, url, environ)
def connect(self, *args, **kargs):
if not kargs['requirements'].get('format'):
kargs['requirements']['format'] = 'json|xml'
return routes.Mapper.connect(self, *args, **kargs) 參考文檔
routes安裝：https://pypi.org/project/Routes/
routes文檔：https://routes.readthedocs.io/en/latest/
小計：
- 元類的使用方法
- openstack定時任務
- python3 異步函數
![TestLink導出用例轉換工具(XML2Excel)[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)