Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
A server model with middleware support by using closure
# 上下文类,本例中主要用于存储当前调用的下文,即内层中间件
class Context():
def __init__(self):
self._next = []
@property
def next(self):
return self._next
class Server():
def __init__(self):
self._middlewares = [] # 所有添加中间件
self._funcs = {} # 所有添加的核心件
def add_middleware(self, middleware_func):
self._middlewares.append(middleware_func)
return middleware_func
def add_func(self, name):
def decorate(func):
self._funcs.setdefault(name, func)
return func
return decorate
def _load_middleware(self, ctx, func):
def next(*args, **kwargs):
return func(ctx, *args, **kwargs)
for middleware in reversed(self._middlewares):
# 使用闭包来封装中间件
def f(middleware=middleware, next=next):
def new_next(*args, **kwargs):
ctx._next = next
return middleware(ctx, *args, **kwargs)
new_next.__name__ = getattr(middleware, '__name__')
return new_next
next = f()
return next
def _wrap(self, func):
def f(*args, **kwargs):
ctx = Context()
return self._load_middleware(ctx, func)(*args, **kwargs)
return f
def initilize(self):
for name, func in self._funcs.items():
self.__setattr__(name, self._wrap(func))
server = Server()
@server.add_middleware
def the_first_middleware(ctx, *args, **kwargs):
print("The first one")
return ctx.next(*args, **kwargs)
@server.add_middleware
def the_second_middleware(ctx, *args, **kwargs):
print("The second one")
return ctx.next(*args, **kwargs)
@server.add_middleware
def the_last_middleware(ctx, *args, **kwargs):
print("The last one")
return ctx.next(*args, **kwargs)
@server.add_func('core_func')
def core_func(ctx, *args, **kwargs):
return "The Core Function"
server.initilize()
print(server.core_func())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.