13518219792
目录
路由分组：...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
/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对象；
例：
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('
idx = Router()
py = Router('/python')
Application.register(idx) #一定要注册
Application.register(py)
@py.get('/(\w+)') #匹配/python/xxxx，只支持get方法
def showpython(request):
res = Response()
res.body = '
return res
@idx.route('^/$') #只匹配根，支持所有方法
def index(request):
res = Response()
res.body = '
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
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语句；
例：
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
# '/(?<{}>{})'.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
# 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
('/student/xxx/(?P
('/student/(?P
('/student/xxx/133456', {})
('/student/(?P
目前处理流程：
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('
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 = '
res.body = '
return res
@idx.route('^/$')
def index(request):
res = Response()
res.body = '
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('^/$') {}
/python/test/456
('GET',) re.compile('/(?P
<_sre.SRE_Match object; span=(0, 9), match='/test/456'>
{'name': 'test', 'id': 456}
/favicon.ico
() re.compile('^/$') {}
/favicon.ico
另外有需要云服务器可以了解下创新互联cdcxhl.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。
