53web開發5_路由分組

?

?

目錄

路由分組：... 1

VER1：... 2

字典訪問屬性化：... 4

VER2：... 6

正則表達式簡化：... 7

?

?

?

?

路由分組：

?

分組捕獲：

支持正則表達式的捕獲；

什么時候捕獲？在框架回調__call__()時，送入request，拿到request.path和正則的模式匹配后，就可提取分組；

如何處理分組？web app就是handler對應的不同的函數，其參數request是一樣的，將捕獲的數據動態的增加到request對象上；

?

用動態增加屬性，為request增加args、kwargs屬性，在handler中使用時，就可直接從屬性中將args、kwargs拿出來直接用；

request.args=matcher.group()?? #所有分組組成的元組，包括命名的

request.kwargs=matcher.groupdict()?? #命名分組組成的字典，用此種

?

所謂路由分組，就是按前綴分別映射；

如下，是不同業務：

/product/tv/1234

/python/student/3456

?

/product/(\w+)/(?P<id>\d+)

/product和/python為一級目錄，常用的/admin（后臺管理），可稱為prefix，前綴必須以/開頭，不能以分隔符結尾；

?

如何建立prefix和url之間的隸屬關系？一個prefix下可有若干個url，這些url都屬于這個prefix的；

建立一個Router類，里面保存prefix，同時保存url和handler的對應關系；

之前，所有注冊方法都是Application的類方法，即所有映射都保存在ROUTER_TABLE類屬性中，但現在不同前綴就是不同的Router實例，所有注冊方法，都成了實例的方法，路由表由Router實例自己管理；

Application中當前只需要保存所有注冊的Router對象（實例）就行，__call__()依然是回調入口，在其中遍歷所有Router實例，找到路徑匹配的Router實例，讓Router實例返回Response對象；

?

?

VER1：

例：

from wsgiref.simple_server import make_server

from webob import Request, Response, dec, exc

import re

?

?

class Router:

??? def __init__(self, prefix: str=''):

??????? self.__prefix = prefix.rstrip('/\\')?? #/python/或/python\\轉為/python，注意不能用strip()

??????? self.__routertable = []?? #[(methods, re.compile(pattern), handler)]

?

??? @property

??? def prefix(self):?? #為之后使用方便，設為類屬性方式

??????? return self.__prefix

?

??? def route(self, pattern, *methods):

??????? def wrapper(handler):

??????????? self.__routertable.append((methods, re.compile(pattern), handler))

??????????? return handler

??????? return wrapper

?

??? def get(self, pattern):

??????? return self.route(pattern, 'GET')

?

??? def post(self, pattern):

??????? return self.route(pattern, 'POST')

?

??? def head(self, pattern):

??????? return self.route(pattern, 'HEAD')

?

??? def match(self,request:Request)->Response:

??????? if not request.path.startswith(self.prefix):?? #前綴處理，不是對應的前綴直接返回None；字符串方法startswith()返回bool，startswith([prefix[,start[,end]])-->bool，prefix開頭

??????????? return

??????? for methods, regex, handler in self.__routertable:

??????????? print(methods, regex, handler)

??????????? if not methods or request.method.upper() in methods:?? #not methods即支持全部方法

??????????????? matcher = regex.search(request.path.replace(self.prefix, '', 1))?? #request.path路徑一定是prefix開頭，去掉prefix，剩下的才是正則匹配的路徑，replace(old,new[,count])-->new str

??????????????? if matcher:

??????????????????? print(matcher)

??????????????????? request.kwargs = matcher.groupdict()?? #命名分組組成的字典

??????????????????? return handler(request)

??????? # return?? #匹配不上返回None

?

?

class Application:

??? ROUTERS = []

?

??? @classmethod

??? def register(cls, router:Router):

??????? return cls.ROUTERS.append(router)

?

??? @dec.wsgify

??? def __call__(self, request:Request) -> Response:

??????? for router in self.ROUTERS:?? #遍歷ROUTERS，調用Router實例的match()方法看誰匹配

??????????? response = router.match(request)

??????????? if response:?? #匹配返回非None的Router對象，匹配則立即返回

??????????????? return response

??????? raise exc.HTTPNotFound('<h2>the page not found</h2>')

?

idx = Router()

py = Router('/python')

?

Application.register(idx)?? #一定要注冊

Application.register(py)

?

@py.get('/(\w+)')?? #匹配/python/xxxx，只支持get方法

def showpython(request):

??? res = Response()

??? res.body = '<h2>hello python</h2>'.encode()

??? return res

?

@idx.route('^/$')?? #只匹配根，支持所有方法

def index(request):

??? res = Response()

??? res.body = '<h2>welcome</h2>'.encode()

??? return res

?

if __name__ == '__main__':

??? ip = '127.0.0.1'

??? port = 9999

??? server = make_server(ip, port, Application())

??? try:

??????? server.serve_forever()

??? except Exception as e:

??????? print(e)

??? finally:

??????? server.shutdown()

??????? server.server_close()

?

?

?

字典訪問屬性化：

d = { 'a': 8}

改造成這樣用：

d.a

d.a=9

?

例：

class DictObj:

??? def __init__(self, d: dict):

??????? # self._dict = d?? #只要有屬性賦值都要調用__setattr__()方法，如有__setattr__()調用該方法并動態寫入到__dict__中，而該沒實現寫入

??????? if isinstance(d, dict):?? #通常用if not isinstance(d, dict)

??????????? self.__dict__['_dict'] = d

??????? else:

??????????? self.__dict__['_dict'] = {}

?

??? def __getattr__(self, item):

??????? try:

??????????? # print(self._dict)

??????????? return self._dict[item]?? #不能用return getattr(self._dict, item)這種方式??????? ???????????????? except KeyError:

??????????? raise AttributeError('Attribute {} not found'.format(item))

?

??? def __setattr__(self, key, value):?? #不寫該方法則可添加屬性，與要求不符

??????? # self._dict[key] = value

??????? # print(key, value)

??????? raise NotImplementedError

?

d = {'a':8}

do = DictObj(d)

print(do.__dict__)

print(do.a)?? #do.a和d['a']類似DB中的視圖和原本數據

# do.a=9?? #在注釋掉__setattr__()后，可添加屬性進去

# print(do.a)

print(do.b)

輸出：

{'_dict': {'a': 8}}

8

Traceback (most recent call last):

? File "E:/git_practice/cmdb/example_DictObj.py", line 13, in __getattr__

??? return self._dict[item]

KeyError: 'b'

?

During handling of the above exception, another exception occurred:

?

Traceback (most recent call last):

? File "E:/git_practice/cmdb/example_DictObj.py", line 28, in <module>

??? print(do.b)

? File "E:/git_practice/cmdb/example_DictObj.py", line 15, in __getattr__

??? raise AttributeError('Attribute {} not found'.format(item))

AttributeError: Attribute b not found

?

例，錯誤示例：

遞歸：

訪問實例屬性，先找__dict__再找__getattr__，兩處都沒有遞歸一直找；

凡是屬性訪問最后都找__getattr__；

class DictObj:

??? def __init__(self, d: dict):

??????? self._dict = d

??????? # if isinstance(d, dict):

??????? #???? self.__dict__['_dict'] = d

??????? # else:

??????? #???? self.__dict__['_dict'] = {}

?

??? def __getattr__(self, item):

??????? try:

??????????? # print(self._dict)

??????????? return self._dict[item]

??????? except KeyError:

??????????? raise AttributeError('Attribute {} not found'.format(item))

?

??? def __setattr__(self, key, value):

??????? self._dict[key] = value

??????? # print(key, value)

??????? # raise NotImplementedError

?

d = {'a':8}

do = DictObj(d)

print(do.__dict__)

print(do.a)

輸出：

……

? File "E:/git_practice/cmdb/example_DictObj.py", line 13, in __getattr__

??? return self._dict[item]

RecursionError: maximum recursion depth exceeded

?

pycharm中調試程序：

先下斷點；

右鍵Debug "example_DictObj"；

看棧，看變量；

斷點+print語句；

?

VER2：

例：

class DictObj:

??? def __init__(self, d: dict):

??????? if not isinstance(d, dict):

??????????? self.__dict__['_dict'] = {}

??????? else:

??????????? self.__dict__['_dict'] = d

?

??? def __getattr__(self, item):

??????? try:

??????????? return getattr(self._dict, item)

??????? except KeyError:

??????????? raise AttributeError('Attribute {} Not Found '.format(self._dict))

?

??? def __setattr__(self, key, value):

??????? raise NotImplementedError

?

?

class Router:

??? def __init__(self, prefix: str=''):

??????? self.__prefix = prefix.rstrip('/\\')

??????? self.__routertable = []

?

??? @property

??? def prefix(self):

??????? return self.__prefix

?

??? def route(self, pattern, *methods):

??????? def wrapper(handler):

??????????? self.__routertable.append((methods, re.compile(pattern), handler))

??????????? return handler

??????? return wrapper

?

??? def get(self, pattern):

??????? return self.route(pattern, 'GET')

?

??? def post(self, pattern):

??????? return self.route(pattern, 'POST')

?

??? def head(self, pattern):

??????? return self.route(pattern, 'HEAD')

?

??? def match(self,request:Request)->Response:

??????? if not request.path.startswith(self.prefix):

??????????? return

??????? for methods, regex, handler in self.__routertable:

???? ???????print(methods, regex, handler)

??????????? if not methods or request.method.upper() in methods:

??????????????? matcher = regex.search(request.path.replace(self.prefix, '', 1))

??????????????? if matcher:

??????????????????? print(matcher)

???????? ???????????request.kwargs = DictObj(matcher.groupdict())

??????????????????? return handler(request)

??????? # return

?

?

?

正則表達式簡化：

目前路由匹配使用正則表達式定義，不友好，很多用戶不會使用正則，能否簡化？

生產中，url是規范的，不能隨便寫，路徑是有意義的，尤其是restful風格，所以要對url規范，如/product/123456，第1段是業務，第2段是ID；

?

設計：

/student/{name:str}/{id:int}

類型設計，支持str、word、int、float、any類型；通過這樣的設計讓用戶簡化，同時也規范，背后的轉換編程者實現；

另raw類型，直接支持RE；

?

?

例：

import re

?

pattern = '/({[^{}:]+:?[^{}:]*})'

regex = re.compile(pattern)

?

s = '/student/{name:str}/xxx/{id:int}'

s1 = '/student/xxx/{id:int}/yyy'

s2 = '/student/{name:}/xxx/{id}'

s3 = '/student/xxx/133456'

s4 = '/student/{name:}/xxx/{id:aaa}'

?

# /{id:int} => /(?P<id>[+-]?\d+)

# '/(?<{}>{})'.format('id', TYPEPATTERNS['int'])

?

TYPEPATTERNS = {

??? 'str': r'[^/]+',

??? 'word': r'\w+',

??? 'int': r'[+-]?\d+',

??? 'float': r'[+-]?\d+.\d+',

??? 'any': r'.+'

}

?

TYPECAST = {

??? 'str': str,

??? 'word': str,

??? 'int': int,

??? 'float': float,

??? 'any': str

}

?

def transform(kv: str):

??? name, _, type = kv.strip('/{}').partition(':')?? #/{id:int}=>/(?P<id>[+-]\d+)

???????? # name, type = kv.strip('/{}').split(':')?? #'/{id}'.strip('/{}').split(':')，split后返回['id']一個元素，type會拿不到值，報ValueError: not enough values to unpack (expected 2, got 1)，所以此處只能用partition不能用split，partition始終返回3個元素

??? return '/(?P<{}>{})'.format(name, TYPEPATTERNS.get(type, '\w+')), name, TYPECAST.get(type, str)

?

def parse(src: str):

??? start = 0

??? res = ''

??? translator = {}

??? while True:

??????? matcher = regex.search(src, start)

??????? if matcher:

??????????? res += matcher.string[start:matcher.start()]

??????????? tmp = transform(matcher.string[matcher.start():matcher.end()])

??????????? res += tmp[0]

??????????? translator[tmp[1]] = tmp[2]

??????????? start = matcher.end()

??????? else:

??????????? break

??? if res:

??????? return res, translator

??? else:

??????? return src, translator

?

print(parse(s))

print(parse(s1))

print(parse(s2))

print(parse(s3))

print(parse(s4))

輸出：

('/student/(?P<name>[^/]+)/xxx/(?P<id>[+-]\\d+)', {'id': <class 'int'>, 'name': <class 'str'>})

('/student/xxx/(?P<id>[+-]\\d+)', {'id': <class 'int'>})

('/student/(?P<name>\\w+)/xxx/(?P<id>\\w+)', {'id': <class 'str'>, 'name': <class 'str'>})

('/student/xxx/133456', {})

('/student/(?P<name>\\w+)/xxx/(?P<id>\\w+)', {'id': <class 'str'>, 'name': <class 'str'>})

?

?

VER3：

目前處理流程：

b發來請求，被wsgi server調度給Application的__call__()；

Application中遍歷注冊的Router，Router的match()方法來判斷是不是自己處理，先前綴再注冊的規則（規則被裝飾器已轉換成了命名分組的RE了）；

若某個注冊的RE匹配，就把獲取到的參數放到request中，并調用注冊時映射的handler給它傳入request；

handler處理后，返回response，Application中拿到這個response數據，返回給原始的wsgi server；

?

例：

?

from wsgiref.simple_server import make_server

from webob import Request, Response, dec, exc

import re

?

?

class DictObj:

??? def __init__(self, d: dict):

??????? if not isinstance(d, dict):

??????????? self.__dict__['_dict'] = {}

??????? else:

??????????? self.__dict__['_dict'] = d

?

??? def __getattr__(self, item):

??????? try:

?????? ?????return self._dict[item]

??????? except KeyError:

??????????? raise AttributeError('Attribute {} Not Found '.format(self._dict))

?

??? def __setattr__(self, key, value):

??????? raise NotImplementedError

?

?

class Router:

??? pattern = '/({[^{}:]+:?[^{}:]*})'? # /{name:str}

??? regex = re.compile(pattern)

?

??? TYPEPATTERNS = {

??????? 'str': r'[^/]+',

??????? 'word': r'\w+',

??????? 'int': r'[+-]?\d+',

??????? 'float': r'[+-]\d+.\d+',

??????? 'any': r'.+'

??? }

?

??? TYPECAST = {

??????? 'str': str,

??????? 'word': str,

??????? 'int': int,

??????? 'float': float,

??????? 'any': str

??? }

?

??? def _transform(self, kv: str):

??????? name, _, type = kv.strip('/{}').partition(':')

??????? return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str)

?

??? def _parse(self, src: str):

??????? start = 0

??????? res = ''

??????? translator = {}

??????? while True:

??????????? matcher = self.regex.search(src, start)

??????????? if matcher:

??????????????? res += matcher.string[start: matcher.start()]

??????????????? tmp = self._transform(matcher.string[matcher.start():matcher.end()])

??????????????? res += tmp[0]

??????????????? translator[tmp[1]] = tmp[2]

??????????????? start = matcher.end()

??????????? else:

??????????????? break

??????? if res:

??????????? return res, translator

??????? else:

??????????? return src, translator

?

??? def __init__(self, prefix: str=''):

??????? self.__prefix = prefix.rstrip('/\\')

??????? self.__routertable = []?? #[(methods, regex, translator, handler)]

?

??? @property

??? def prefix(self):

??????? return self.__prefix

?

??? def route(self, rule, *methods):

??????? def wrapper(handler):

??????????? pattern, translator = self._parse(rule)

??????????? self.__routertable.append((methods, re.compile(pattern), translator, handler))

??????????? return handler

??????? return wrapper

?

??? def get(self, pattern):

??????? return self.route(pattern, 'GET')

?

??? def post(self, pattern):

??????? return self.route(pattern, 'POST')

?

??? def head(self, pattern):

??????? return self.route(pattern, 'HEAD')

?

??? def match(self, request: Request)->Response:

??????? print(request.path)

??????? if not request.path.startswith(self.prefix):

???????? ???return

??????? for methods, regex, translator, handler in self.__routertable:

??????????? print(methods, regex, translator, handler)

??????????? if not methods or request.method.upper() in methods:

??????????????? matcher = regex.search(request.path.replace(self.prefix, '', 1))

??????????????? if matcher:

??????????????????? print(matcher)

??????????????????? newdict = {}

??????????????????? for k, v in matcher.groupdict().items():

??????????????????????? newdict[k] = translator[k](v)

?????????????????? ?print(newdict)

??????????????????? request.vars = DictObj(newdict)

??????????????????? return handler(request)

??????? # return

?

?

class Application:

??? ROUTERS = []

?

??? @classmethod

??? def register(cls, router: Router):

??????? return cls.ROUTERS.append(router)

?

??? @dec.wsgify

??? def __call__(self, request: Request) -> Response:

??????? for router in self.ROUTERS:

??????????? response = router.match(request)

??????????? if response:

??????????????? return response

???? ???raise exc.HTTPNotFound('<h2>the page not found</h2>')

?

idx = Router()

py = Router('/python')

?

Application.register(idx)

Application.register(py)

?

# @py.get('/{name:str}')

# @py.get('/{id:int}')

@py.get('/{name:str}/{id:int}')

def showpython(request):

? ??res = Response()

??? # print(request.__dict__)

??? # res.body = '<h2>hello python; vars = {}</h2>'.format(request.vars.name).encode()

??? res.body = '<h2>hello python; vars = {}</h2>'.format(request.vars.id).encode()

??? return res

?

@idx.route('^/$')

def index(request):

??? res = Response()

??? res.body = '<h2>welcome</h2>'.encode()

??? return res

?

if __name__ == '__main__':

??? ip = '127.0.0.1'

??? port = 9999

??? server = make_server(ip, port, Application())

??? try:

??????? server.serve_forever()

??? except Exception as e:

??????? print(e)

??? finally:

??????? server.shutdown()

??????? server.server_close()

輸出：

/python/test/456

() re.compile('^/$') {} <function index at 0x00000000033B1BF8>

/python/test/456

('GET',) re.compile('/(?P<name>[^/]+)/(?P<id>[+-]?\\d+)') {'name': <class 'str'>, 'id': <class 'int'>} <function showpython at 0x00000000033B1B70>

<_sre.SRE_Match object; span=(0, 9), match='/test/456'>

{'name': 'test', 'id': 456}

/favicon.ico

() re.compile('^/$') {} <function index at 0x00000000033B1BF8>

/favicon.ico

?

?

?

?

?