类Flask框架实现
模板
Title 显示数据<\br> { {id}} { {name}} { {age}}
import refrom io import StringIO,BytesIOd = { 'id':5,'name':'tom','age':20}class Template: _pattern = '{ {([a-zA-Z0-9_]+)}}' regex = re.compile(_pattern) @classmethod def render(cls,template,data:dict): html = StringIO() with open(template,encoding='utf-8') as f: for line in f: start = 0 newline = '' for matcher in cls.regex.finditer(line): newline += line[start:matcher.start()] print(matcher,matcher.group(1)) key = matcher.group(1) tmp = data.get(key,'') newline += str(tmp) start = matcher.end() else: newline += line[start:] html.write(newline) print(html.getvalue()) html.close()#模板渲染filename = 'index.html'Template.render(filename,d) jinja2
文档官网 中文 安装pip install jinja2pip install MarkupSafe
模板构建
Magedu
- 显示数据 {% for id,name,age in userlist %}
- { {loop.index}},{ {id}},{ {name}},{ {age}} {% endfor %}
#################加载模板代码如下示例###################from jinja2 import Environment,PackageLoader,FileSystemLoader# env = Environment(loader=PackageLoader('webarch','templates')) #包加载器env = Environment(loader=FileSystemLoader('webarch/templates'))template = env.get_template('index.html')userlist = [ (3,'tpm',20), (4,'het',23), (7,'asdf',23), (1,'aasf',18)]d = { 'userlist':userlist,'usercount':len(userlist)}print(template.render(**d)) ###############提供模板模块template.py##################from jinja2 import Environment,PackageLoader,FileSystemLoaderenv = Environment(loader=PackageLoader('webarch','templates')) #包加载器# env = Environment(loader=FileSystemLoader('webarch/templates')) #文件系统加载器def render(name,data:dict): """ 模板渲染 :param name: 去模板目录搜索此模板名的文件 :param data: 字典数据 :return: HTML字符串 """ template = env.get_template(name) #搜索模块index.html return template.render(**data) ##############代码中增加###################创建Router对象idx = Router()py = Router('/python')user = Router('/user')#注册App.register(idx,py)App.register(user)py.register_preinterceptor(ip)@user.get(r'^/?$')def userhandler(request): userlist = [ (3, 'tpm', 20), (4, 'het', 23), (7, 'asdf', 23), (1, 'aasf', 18) ] d = { 'userlist':userlist,'usercount':len(userlist)} return render('index.html',d) 模块化
1.创建包webarch2.template.py模块移入此包3.新建web.py模块，将AttrDict、Router、App类放入其中4.将路由定义、注册代码、handler定义移入webarch/__init__.py中5.在项目根目录下，建一个app.py，里面放入启动server的代码。
拦截器interceptor
#加入拦截器功能的方式1、App和Router类直接加入把拦截器的相关方法、属性分别添加到相关类中实现简单2、MixinApp和Router类都需要这个拦截器功能，可以使用Mixin方式、将属性、方法组合进来，但是，App类拦截器适合使用第二种，但是Router的拦截器是每个实例不一样的，所以使用第一种方式实现
def fn(request:Request) -> Request: pass
def fn(request:Request,response:Response) -> Response: pass
IP拦截
#创建Router对象idx = Router()py = Router('/python')user = Router('/user')#注册App.register(idx,py)App.register(user)#ip拦截def ip(request:Request): print(request.remote_addr,'~~~~~') print(type(request.remote_addr)) if request.remote_addr.startswith('192.'): return request else: return None #返回None将截断请求py.register_preinterceptor(ip) Json支持
@py.get(r'^/{id}$')def pythonhandler(request): userlist = [ (3, 'tom', 20), (5, 'jerry', 16), (6, 'sam', 23), (8, 'kevin', 18) ] d = { 'userlist':userlist,'usercount':len(userlist)} res = Response(json=d) return res 
总结
完整代码
################webarch/__init__.py#####################from webob import Request,Response#模板from .template import renderfrom .web import Router,App#创建Router对象idx = Router()py = Router('/python')user = Router('/user')#注册App.register(idx,py)App.register(user)#ip拦截def ip(request:Request): print(request.remote_addr,'~~~~~') print(type(request.remote_addr)) if request.remote_addr.startswith('192.'): return request else: return None #返回None将截断请求py.register_preinterceptor(ip)@idx.get(r'^/$')@idx.route(r'^/{id:int}$') #支持所有方法def indexhandler(request): id = '' if request.groupdict: id = request.groupdict.id return 'magedu.com{}欢迎你
'.format(id)@py.get(r'^/{id}$')def pythonhandler(request): userlist = [ (3, 'tom', 20), (5, 'jerry', 16), (6, 'sam', 23), (8, 'kevin', 18) ] d = { 'userlist':userlist,'usercount':len(userlist)} res = Response(json=d) return res@user.get(r'^/?$')def userhandler(request): userlist = [ (3, 'tpm', 20), (4, 'het', 23), (7, 'asdf', 23), (1, 'aasf', 18) ] d = { 'userlist':userlist,'usercount':len(userlist)} return render('index.html',d)
###########webarch/web.py#############import refrom webob import Request,Responsefrom webob.exc import HTTPNotFoundfrom webob.dec import wsgifyclass AttrDict: def __init__(self,d:dict): self.__dict__.update(d if isinstance(d,dict) else {}) def __setattr__(self, key, value): #不允许修改属性 raise NotImplementedError def __repr__(self): return " ".format(self.__dict__) def __len__(self): return len(self.__dict__)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): 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(t, str) repl += src[start:matcher.start()] #拼接分组前 tmp = '/(?P<{}>{})'.format( matcher.group(1), self.TYPEPATTERNS.get(matcher.group(2), self.TYPEPATTERNS['str']) ) repl += tmp # 替换 start = matcher.end() #移动 else: repl += src[start:] # 拼接分组后的内容 return repl, types #####实例 def __init__(self,prefix:str=''): self.__prefix = prefix.rstrip('/\\') #前缀，例如/product self.__routetable = [] #存四元组，列表，有序的 ##拦截器## self.pre_interceptor = [] self.post_interceptor = [] ##拦截器注册函数## 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): #/student/{name:str}/xxx/{id:int} ==> '/student/(?P [^/]+/xxx/(?P [+-]\\d+))' pattern,trans = self.__parse(rule) # 用户输入规则转换为正则表达式 self.__routetable.append( (tuple(map(lambda x:x.upper(),methods)), 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): return self.route(pattern,'HEAD') def match(self,request:Request): #必须先匹配前缀 if not request.path.startswith(self.__prefix): return None ##局部拦截，此Router的请求，开始拦截，处理request## for fn in self.pre_interceptor: request = fn(request) if not request: return None #请求为None将不再向后传递，截止 #前缀匹配，说明就必须这个Router实例处理，后续匹配不上，依然返回None for methods,pattern,trans,handler in self.__routetable: # not methods表示一个方法都没有定义，就是支持全部方法 if not methods or request.method.upper() in methods: #前提是以__prefix开头了，可以replace，去掉prefix剩下的才是正则表达式需要匹配的路径 matcher = pattern.match(request.path.replace(self.__prefix,'',1)) if matcher:#正则匹配 newdict = {} for k,v in matcher.groupdict().items(): newdict[k] = trans[k](v) #动态增加属性 request.groupdict =AttrDict(newdict)#命名分组组成的字典被属性化 response = handler(request) ##局部拦截响应，依次拦截，处理响应## for fn in self.post_interceptor: response = fn(request,response) return responseclass App: _ROUTERS = [] #存储所有一级对象 ##全局拦截器## 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,*routers:Router): for router in routers: cls._ROUTERS.append(router) @wsgify def __call__(self, request:Request): ##全局拦截请求## for fn in self.PRE_INTERCEPTOR: request = fn(request) #遍历_ROUTERS,调用Router实例的match方法，进行匹配 for router in self._ROUTERS: response = router.match(request) ##全局拦截响应## for fn in self.POST_INTERCEPTOR: response = fn(request,response) if response: #匹配返回非None的Router对象 return response raise HTTPNotFound(' 你访问的页面不存在！
') #########webarch/template.py#############from jinja2 import Environment,PackageLoader,FileSystemLoaderenv = Environment(loader=PackageLoader('webarch','templates')) #包加载器# env = Environment(loader=FileSystemLoader('webarch/templates')) #文件系统加载器def render(name,data:dict): """ 模板渲染 :param name: 去模板目录搜索此模板名的文件 :param data: 字典数据 :return: HTML字符串 """ template = env.get_template(name) #搜索模块index.html return template.render(**data) ###########app.py#########from .web import Appfrom wsgiref.simple_server import make_serverif __name__ == '__main__': ip = '127.0.0.1' port= 9999 server = make_server(ip,port,App()) try: server.serve_forever() # server.handle_request() 一次 except KeyboardInterrupt: server.shutdown() server.server_close()
发布
#######setup.py####from distutils.core import setupimport glob#导入setup函数并传参setup( name='webarch', version='0.1.0', description='Python WSGI Web Framework', author='cy', url='https://www.magedu/com', packages=['webarch'], datafiles=glob.glob('webarch/templates/*.html')#返回列表) 
