53web开发5_路由分组-创新互联

目录

十多年的印江网站建设经验，针对设计、前端、开发、售后、文案、推广等六对一服务，响应快，48小时及时工作处理。全网营销推广的优势是能够根据用户设备显示端的尺寸不同，自动调整印江建站的显示方式，使网站能够适用不同显示终端，在浏览器中调整网站的宽度，无论在任何一种浏览器上浏览网站，都能展现优雅布局与设计，从而大程度地提升浏览体验。成都创新互联公司从事“印江网站设计”,“印江网站推广”以来，每个客户项目都认真落实执行。

路由分组：...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；

str

不包含/的任意字符

[^/]+

word

字母和数字

\w+

int

纯数字，正负数

[+-]?\d+

float

正负号、数字、包含.

[+-]?\d+.\d+

any

包含/的任意字符

.+

例：

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

53web开发5_路由分组

另外有需要云服务器可以了解下创新互联cdcxhl.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。

分享名称：53web开发5_路由分组-创新互联
网址分享：https://www.cdcxhl.com/article38/dsgosp.html

成都网站建设公司_创新互联，为您提供网站制作、网站营销、小程序开发、定制网站、用户体验、网站设计

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

外贸网站建设

虚拟主机知识

各行业网站