這篇文章主要介紹“web開發(fā)攔截器的方法是什么”，在日常操作中，相信很多人在web開發(fā)攔截器的方法是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”web開發(fā)攔截器的方法是什么”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！
在請求處理環(huán)節(jié)的某處加入處理，有可能是中斷后續(xù)的處理；
類似java的structs框架中的攔截器，servelet中也有攔截器；
根據(jù)攔截點不同，分：請求掛載；響應(yīng)攔截；
根據(jù)影響面，分：全局?jǐn)r截（在Application中攔截）；局部攔截（在Router中攔截）；
不能在handler中作攔截，即不能在一個完整的功能模塊中攔截；
在__call__()進(jìn)去之后，__call__()return之前作攔截；
注：
依次執(zhí)行，一環(huán)扣一環(huán)，上一步的輸出是下一步的輸入；
攔截器可以是多個，多個攔截器是有順序的；
數(shù)據(jù)response之前執(zhí)行的命名為pre_interceptor，之后的執(zhí)行命名為post_interceptor；
某些特定功能要求最終返回給用戶為404，經(jīng)fn后rerturn None，這樣達(dá)到目的就行；
方式1：
Application和Router類中直接加入；
把攔截器的相關(guān)方法，屬性分別添加到相關(guān)類中；
Router的攔截器是每個實例都不一樣，適合用此種；
方式2：
Mixin；
Application和Router都需要這個攔截器功能，這兩個類有無關(guān)系，可用mixin方式，將屬性方法組合進(jìn)來；
Application適合使用此種；
def fn(app,request:Request)->Request: pass #fn不能影響數(shù)據(jù)繼續(xù)向下一級的傳遞，即是透明的，如handle的輸入要是request，前面的攔截的輸出也要是request，handle處理后是response，經(jīng)攔截器最終到__call__()也要是response
def fn(app,reqeust:Request)->Request: pass #引入app即Application的實例，是為以后從Application上獲取一些全局信息
為把一些應(yīng)用數(shù)據(jù)、配置數(shù)據(jù)、數(shù)據(jù)庫連接等全局共享數(shù)據(jù)提供給所有對象使用，增加一個字典，存儲這些共享數(shù)據(jù)；
為方便訪問，提供字典的屬性化訪問的類，且該字典可寫；
例：
class Context(dict):
def __getattr__(self, item):
try:
return self[item]
except KeyError:
raise AttributeError('Attribute {} Not Found'.fomrat(item))
def __setattr__(self, key, value):
self[key] = value
class Application:
ctx = Context()
ROUTERS = []
def __init__(self, **kwargs):
self.ctx.app = self
for k, v in kwargs:
self.ctx[k] = v
Router的每一個實例中增加上下文屬性，實例自己使用；
Router實例如何使用全局上下文？
用新的處理方法，增加NestedContext類，每一個Router實例的上下文字典內(nèi)部關(guān)聯(lián)一個全局字典的引用，如果自己的字典中找不到，就去全局里找；
Router實例什么時候關(guān)聯(lián)全局字典？
在路由注冊時較合適，只需修改下注冊函數(shù)即可
例，錯誤示例：
class Mixin:
# def __init__(self): pass #僅實現(xiàn)功能，不能有初始化
class A:
def __init__(self): pass
class C(Mixin, A):
def __init__(self):
super().__init__() #錯誤，這樣用的是Mixin的初始化方法，把A的覆蓋掉了
例：
class Context(dict):
def __getattr__(self, item):
try:
return self[item]
except KeyError:
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value):
self[key] = value
class NestedContext(Context):
def __init__(self, globalcontext:Context=None):
super().__init__() #此句可沒有，父類中未初始化，按標(biāo)準(zhǔn)流程應(yīng)寫上
self.relate(globalcontext)
def relate(self, globalcontext:Context=None):
self.globalcontext = globalcontext
def __getattr__(self, item):
if item in self.keys():
return self[item]
return self.globalcontext[item]
ctx = Context()
ctx.x = 6
ctx.y = 'a'
nc = NestedContext()
nc.relate(ctx)
nc.x = 8
print(nc) #
print(nc.x) #自己的
print(nc.y) #全局的
print(nc.z) #KeyError
輸出：
{'globalcontext': {'y': 'a', 'x': 6}, 'x': 8}
8
a
Traceback (most recent call last):
File "E:/git_practice/cmdb/example_wsgi_interceptor.py", line 37, in
print(nc.z)
File "E:/git_practice/cmdb/example_wsgi_interceptor.py", line 24, in __getattr__
return self.globalcontext[item]
KeyError: 'z'
將如下代碼改為單例（單例模式，只允許創(chuàng)建一個實例）；
多線程時，要么鎖要么信號量；
多進(jìn)程時，用進(jìn)程中的信號量；
例：
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 Context(dict):
def __getattr__(self, item):
try:
return self[item]
except KeyError:
raise AttributeError('Attrubute {} Not found'.format(item))
def __setattr__(self, key, value):
self[key] = value
class NestedContext(Context):
def __init__(self, globalcontext:Context=None):
super().__init__()
self.relate(globalcontext)
def relate(self, globalcontext:Context=None):
self.globalcontext = globalcontext
def __getattr__(self, item):
if item in self.keys():
return self[item]
return self.globalcontext[item]
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)]
self.pre_interceptor = []
self.post_interceptor = []
self.ctx = NestedContext()
@property
def prefix(self):
return self.__prefix
def register_preinterceptor(self, fn):
self.pre_interceptor.append(fn)
return fn
def register_postinterceptor(self, fn):
self.post_interceptor.append(fn)
return fn
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 fn in self.pre_interceptor:
request = fn(self.ctx, request)
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:
ctx = Context()
ROUTERS = []
def __init__(self, **kwargs):
self.ctx.app = self
for k, v in kwargs:
self.ctx[k] = v
PRE_INTERCEPTOR = []
POST_INTERCEPTOR = []
@classmethod
def register_preinterceptor(cls, fn):
cls.PRE_INTERCEPTOR.append(fn)
return fn
@classmethod
def register_postinterceptor(cls, fn):
cls.POST_INTERCEPTOR.append(fn)
return fn
@classmethod
def register(cls, router: Router):
router.ctx.relate(cls.ctx)
router.ctx.router = router
cls.ROUTERS.append(router)
@dec.wsgify
def __call__(self, request: Request) -> Response:
for fn in self.PRE_INTERCEPTOR:
request = fn(self.ctx, request)
for router in self.ROUTERS:
response = router.match(request)
for fn in self.POST_INTERCEPTOR:
response = fn(self.ctx, request, response)
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
@Application.register_preinterceptor
def showheaders(ctx: Context, request: Request) -> Response:
print(request.path)
print(request.user_agent)
return request
@py.register_preinterceptor
def showprefix(ctx: Context, request: Request)->Response:
print('~~~~~~~prefix = {}'.format(ctx.router.prefix))
return request
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()
到此，關(guān)于“web開發(fā)攔截器的方法是什么”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
重慶分公司
重慶分公司
