类实现wsgi app
from wsgiref.util import setup_testing_defaults from wsgiref.simple_server import make_server class Simple_App: def __init__(self,environ,start_response): self.environ = environ self.start_response =start_response status = '200 ok' headers = [('Content-type','text/plain;charset=utf-8')] self.start_response(status,headers) self.ret = ['{}'.format(v).encode() for k,v in self.environ.items()] def __iter__(self): yield from self.ret with make_server('0.0.0.0',9000,Simple_App) as httpd: httpd.serve_forever()
函数实现wsgi,app的方法
from wsgiref.simple_server import make_server def simple_app(environ, start_response): status = '200 OK' headers = [('Content-type', 'text/plain; charset=utf-8')] qstr = environ.get('QUERY_STRING') # if qstr: # for pair in qstr.split('&'): # k,_,v = pair.partition("=") # print('k={},v={}'.format(k,v)) if qstr: querydict={k:v for k,_,v in map(lambda x:x.partition("="),qstr.split("&"))} print(querydict) start_response(status, headers) ret = [("%s: %s " % (key, value)).encode("utf-8") for key, value in environ.items()] return ret #返回一个可迭代对象,正文就是这个列表的元素 with make_server('', 8000, simple_app) as httpd: print("Serving on port 8000...") print("11111111111") httpd.serve_forever()
理解webob的Request,Response
from webob import Request,Response from wsgiref.simple_server import make_server def simple_server(environ,start_response): request = Request(environ) print(request.params) response = Response() response.body = b'<h1>hello world<h1>' return response(environ,start_response) with make_server('0.0.0.0',9999,simple_server) as httpd: httpd.serve_forever()
wsgify装饰器的实现
from webob import Request,Response from wsgiref.simple_server import make_server from webob.dec import wsgify @wsgify def app(request:Request)->Response: #一个请求对应一个响应 return Response('<h1>welcome to China</h1>')
#返回值会被封装成webob.response类型实例的body属性 # return Response(b'<h2>welcome to China 2 </h2>') # return b'<h3>welcome to China 3</h3>' with make_server('0.0.0.0',9999,app) as httpd: print('success') httpd.serve_forever()
wsgi路由实现
from webob import Response,Request from webob.dec import wsgify from wsgiref.simple_server import make_server from webob.exc import HTTPNotFound import re class Router: def __init__(self,prefix:str=''): self.__prefix = prefix self.routeable = [] #[(method,pattern,handler)] def route(self,pattern,*method): #method is None instead of all of method def wrapper(handler): self.routeable.append((tuple(map(lambda x:x.upper(),method)),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): self.route(pattern,'HEAD') def match(self,request): #receive browser requset if not request.path.startswith(self.__prefix): return for method,pattern,handler in self.routeable: if request.method in method or not method: matcher = pattern.match(request.path.replace(self.__prefix,'',1)) if matcher: request.groupdict = matcher.groupdict() return handler(request) class App: _Routers = [] @classmethod def regesiter(cls, *routers): for router in routers: cls._Routers.append(router) @wsgify def __call__(self, request:Request): for i in self._Routers: result = i.match(request) if result: return result raise HTTPNotFound('this goes out to homepage') root = Router() python = Router('/python') #实例设设置了url的前缀 App.regesiter(root,python) @root.get('^/(?P<id>d+)') @root.get('^/$') def roothandler(request:Request): return Response("<h1>Welcome to roothandler</h1>") @python.get('^/(w+)')#/python/aa 判断aas是否匹配上url,/python在代码中被替换掉成空 def pythonhandler(request:Request): return Response('<h1>Welcome to pythonhandler</h1>') if __name__ == '__main__': with make_server('0.0.0.0',9999,App()) as httpd: httpd.serve_forever()
wsgi更改路由限定方式
from webob import Response,Request from webob.dec import wsgify from wsgiref.simple_server import make_server from webob.exc import HTTPNotFound import re class Router: __regex = re.compile(r'/{([^{}:]+):?([^{}:]*)}') TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'w+', 'int': r'[+-]?d+', 'float': r'[+-]?d+.d+', 'any': r'.+' } def _repl(self, matcher): return '/(?P<{}>{})'.format(matcher.group(1), self.TYPEPATTERNS.get(matcher.group(2), self.TYPEPATTERNS['str'])) def __parse(self,src): return self.__regex.sub(self._repl,src) def __init__(self,prefix:str=''): self.__prefix = prefix self.routeable = [] #[(method,pattern,handler)] def route(self,pattern,*method): #method is None instead of all of method def wrapper(handler): self.routeable.append((tuple(map(lambda x:x.upper(),method)),re.compile(self.__parse(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): self.route(pattern,'HEAD') def match(self,request): #receive browser requset if not request.path.startswith(self.__prefix): return for method,pattern,handler in self.routeable: if request.method in method or not method: matcher = pattern.match(request.path.replace(self.__prefix,'',1)) if matcher: request.groupdict = matcher.groupdict() return handler(request) class App: _Routers = [] @classmethod def regesiter(cls, *routers): for router in routers: cls._Routers.append(router) @wsgify def __call__(self, request:Request): for i in self._Routers: result = i.match(request) if result: return result raise HTTPNotFound('this goes out to homepage') root = Router() python = Router('/python') #实例设设置了url的前缀 App.regesiter(root,python) # @root.get('^/(?P<id>d+)') @root.get('^/$') @root.get(r'^/{id:int}$') def roothandler(request:Request): return Response("<h1>Welcome to roothandler</h1>") # @python.get('^/(w+)')#/python/aa 判断aas是否匹配上url,/python在代码中被替换掉成空 @python.get('^/{id}') def pythonhandler(request:Request): return Response('<h1>Welcome to pythonhandler</h1>') if __name__ == '__main__': with make_server('0.0.0.0',9999,App()) as httpd: httpd.serve_forever()
重写sub
from webob import Response,Request from webob.dec import wsgify from wsgiref.simple_server import make_server from webob.exc import HTTPNotFound import re class Router: __regex = re.compile(r'/{([^{}:]+):?([^{}:]*)}') 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 __parse(self,src: str): #src = {id:int} start = 0 repl = '' types = {} matchers = self.__regex.finditer(src) for i, matcher in enumerate(matchers): name = matcher.group(1) t = matcher.group(2) types[name] = self.TYPECAST.get(matcher.group(2), str) repl += src[start:matcher.start()] tmp = '/(?P<{}>{})'.format(matcher.group(1), self.TYPEATTERNS.get(matcher.group(2), self.TYPEATTERNS['str'])) start = matcher.end() else: repl += src[start:] return repl, types #(r'[^/]+',{id:int}) def __init__(self,prefix:str=''): self.__prefix = prefix self.routeable = [] #[(method,pattern,handler)] def route(self,rule,*method): #method is None instead of all of method def wrapper(handler): pattern,trans = self.__parse(rule) self.routeable.append((tuple(map(lambda x:x.upper(),method)),re.compile(pattern),trans,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): self.route(pattern,'HEAD') def match(self,request): #receive browser requset if not request.path.startswith(self.__prefix): return for method,pattern,trans,handler in self.routeable: if request.method in method or not method: matcher = pattern.match(request.path.replace(self.__prefix,'',1)) #/python/111 ;mathc(111) if matcher:#{id:111} newdict = {} for k,v in matcher.groupdict.items(): newdict[k] = trans[k](v) request.groupdict = matcher.groupdict() return handler(request) class App: _Routers = [] @classmethod def regesiter(cls, *routers): for router in routers: cls._Routers.append(router) @wsgify def __call__(self, request:Request): for i in self._Routers: result = i.match(request) if result: return result raise HTTPNotFound('this goes out to homepage') root = Router() python = Router('/python') #实例设设置了url的前缀 App.regesiter(root,python) # @root.get('^/(?P<id>d+)') @root.get('^/$') @root.get(r'^/{id:int}$') def roothandler(request:Request): return Response("<h1>Welcome to roothandler</h1>") # @python.get('^/(w+)')#/python/aa 判断aas是否匹配上url,/python在代码中被替换掉成空 @python.get('^/{id}') def pythonhandler(request:Request): return Response('<h1>Welcome to pythonhandler</h1>') if __name__ == '__main__': with make_server('0.0.0.0',9999,App()) as httpd: httpd.serve_forever()