温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

53web开发5_路由分组

发布时间:2020-07-07 13:58:12 来源:网络 阅读:204 作者:chaijowin 栏目:编程语言

 

 

目录

路由分组:... 1

VER1... 2

字典访问属性化:... 4

VER2... 6

正则表达式简化:... 7

 

 

 

 

路由分组:

 

分组捕获:

支持正则表达式的捕获;

什么时候捕获?在框架回调__call__()时,送入request,拿到request.path和正则的模式匹配后,就可提取分组;

如何处理分组?web app就是handler对应的不同的函数,其参数request是一样的,将捕获的数据动态的增加到request对象上;

 

用动态增加属性,为request增加argskwargs属性,在handler中使用时,就可直接从属性中将argskwargs拿出来直接用;

request.args=matcher.group()   #所有分组组成的元组,包括命名的

request.kwargs=matcher.groupdict()   #命名分组组成的字典,用此种

 

所谓路由分组,就是按前缀分别映射;

如下,是不同业务:

/product/tv/1234

/python/student/3456

 

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

/product/python为一级目录,常用的/admin(后台管理),可称为prefix,前缀必须以/开头,不能以分隔符结尾;

 

如何建立prefixurl之间的隶属关系?一个prefix下可有若干个url,这些url都属于这个prefix的;

建立一个Router类,里面保存prefix,同时保存urlhandler的对应关系;

之前,所有注册方法都是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()返回boolstartswith([prefix[,start[,end]])-->boolprefix开头

            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:   #匹配返回非NoneRouter对象,匹配则立即返回

                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.ad['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}

类型设计,支持strwordintfloatany类型;通过这样的设计让用户简化,同时也规范,背后的转换编程者实现;

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不能用splitpartition始终返回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中遍历注册的RouterRoutermatch()方法来判断是不是自己处理,先前缀再注册的规则(规则被装饰器已转换成了命名分组的RE了);

若某个注册的RE匹配,就把获取到的参数放到request中,并调用注册时映射的handler给它传入request

handler处理后,返回responseApplication中拿到这个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_路由分组

 

 

 

 

 


向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI