Shulou(Shulou.com)06/02 Report--

This article mainly introduces "what is the method of web development interceptor". In daily operation, I believe that many people have doubts about the method of web development interceptor. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts of "what is the method of web development interceptor"! Next, please follow the editor to study!

Interceptor, interceptor:

Adding processing somewhere in the request processing process may interrupt subsequent processing

Interceptor in structs framework similar to java, and interceptor in servelet

Classification:

According to the interception point, it can be divided into: request mount; response interception

According to the influence surface, it is divided into: global interception (intercept in Application) and local interception (interception in Router)

Cannot be intercepted in handler, that is, it cannot be intercepted in a complete functional module

Intercept before _ _ call__ () return after _ _ call__ () goes in

Note:

Execute in turn, one link at a time, and the output of the previous step is the input of the next step

Interceptors can be multiple, and multiple interceptors are in order

The previous execution of the data response is named pre_interceptor, and the subsequent execution is named post_interceptor

Some specific functional requirements will eventually be returned to the user for 404 and rerturn None after fn, so that the goal can be achieved.

How to add interceptor functionality:

Method 1:

Add directly to the Application and Router classes

Add the relevant methods and attributes of the interceptor to the related classes respectively

The interceptor of Router is different from each instance, so it is suitable to use this kind of

Mode 2:

Mixin

Both Application and Router need this interceptor function. Whether these two classes are related or not, you can use mixin to combine attribute methods.

Application is suitable for the use of this

The design of interceptor fn function:

Def fn (app,request:Request)-> Request: pass # fn cannot affect the transmission of data to the next level, that is, it is transparent. For example, if the input of handle is request, the output of the previous intercept should also be response after request,handle processing, and finally _ _ call__ () should be response after interceptor.

Def fn (app,reqeust:Request)-> Request: pass # introduces app, that is, an instance of Application, to obtain some global information from Application in the future

Context support:

In order to provide some global shared data such as application data, configuration data, database connections and so on to all objects, add a dictionary to store these shared data.

For easy access, a class for attributive access to the dictionary is provided, and the dictionary can write

Example:

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

A context attribute is added to each instance of Router, and the instance uses its own

How does a Router instance use a global context?

With the new processing method, add the NestedContext class. Each Router instance's context dictionary is associated with a global dictionary reference. If you can't find it in your own dictionary, go to the global dictionary to find it.

When will the Router instance be associated with the global dictionary?

It is more appropriate for route registration. You only need to modify the registration function.

Example, error example:

Class Mixin:

# def _ init__ (self): pass # only implements functions, not initialization

Class A:

Def _ _ init__ (self): pass

Class C (Mixin, A):

Def _ init__ (self):

Super (). _ _ init__ () # error, so the initialization method of Mixin is used to overwrite A

Example:

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__ () # this sentence does not exist. It is not initialized in the parent class and should be written according to the standard process.

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) # own

Print (nc.y) # global

Print (nc.z) # KeyError

Output:

{'globalcontext': {' yearly:'a','x: 6},'x: 8}

eight

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'

Complete code:

Change the following code to a singleton (singleton mode, only one instance is allowed)

When multithreaded, either locks or semaphores

When there are multiple processes, use the semaphores in the process

Example:

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'\ wicked'

'int': r' [+ -]?\ dflowers'

'float': r' [+ -]\ dcats.\ dweeds'

'any': ritual. Accounting.'

}

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,'\ wicked'), 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 ('the page not found')

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 = 'hello python; vars = {}' .format (request.vars.name) .encode ()

Res.body = 'hello python; vars = {}' .format (request.vars.id) .encode ()

Return res

@ idx.route ('^ / $')

Def index (request):

Res = Response ()

Res.body = 'welcome'.encode ()

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 ()

At this point, the study on "what is the method of web development interceptor" is over. I hope to be able to solve everyone's doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.