首页 > 代码库 > 封装框架1
封装框架1
目的:
1、学习web框架是如何工作的
2、如何编写用户友好的API 除了功能性能 就是友好
先掌握框架的核心内容
最简单的wsgi应用
environ参数:是客户端和服务器端的环境,有容器传递给应用
start_response函数:传递应用要返回的头信息和body体
只能是这两个参数是容器与应用通信的 是位置参数,名字可以取别的
def application(environ, start_response): start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/plain‘)]) # 返回文本格式 return ["hello,world".encode()] # 返回的要是bytes的可迭代对象 if __name__ == ‘__main__‘: from wsgiref.simple_server import make_server server = make_server(‘0.0.0.0‘, 4000, app=application) # Python自带的容器 try: server.serve_forever() except KeyboardInterrupt: server.shutdown()
了解environ参数内容
# 了解environ def application(environ, start_response): for k, v in environ.items(): print(‘{}--->{}‘.format(k, v)) start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/plain‘)]) return ["hello,world".encode()] if __name__ == ‘__main__‘: from wsgiref.simple_server import make_server server = make_server(‘0.0.0.0‘, 4000, app=application) try: server.serve_forever() except KeyboardInterrupt: server.shutdown()
容器是干什么的?
容器是用来做协议转换的
http协议---->转换为WSGI协议
tomcat 也是把http-->servlet 协议
主要关心客户端传来的信息
import os def application(environ, start_response): for k, v in environ.items(): if k not in os.environ.keys(): print(‘{}--->{}‘.format(k, v)) start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/plain‘)]) return ["hello,world".encode()]
REMOTE_ADDR--->127.0.0.1 CONTENT_LENGTH---> wsgi.version--->(1, 0) wsgi.multiprocess--->False wsgi.errors---><_io.TextIOWrapper name=‘<stderr>‘ mode=‘w‘ encoding=‘UTF-8‘> REMOTE_HOST---> SERVER_NAME--->DESKTOP-16F3VS1 wsgi.url_scheme--->http wsgi.file_wrapper---><class ‘wsgiref.util.FileWrapper‘> PATH_INFO--->/favicon.ico HTTP_PRAGMA--->no-cache wsgi.run_once--->False HTTP_CACHE_CONTROL--->no-cache CONTENT_TYPE--->text/plain REQUEST_METHOD--->GET HTTP_ACCEPT_ENCODING--->gzip, deflate, sdch, br SERVER_SOFTWARE--->WSGIServer/0.2 HTTP_ACCEPT--->image/webp,image/*,*/*;q=0.8 QUERY_STRING---> HTTP_USER_AGENT--->Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.98 Safari/537.36 GATEWAY_INTERFACE--->CGI/1.1 wsgi.input---><_io.BufferedReader name=844> SERVER_PROTOCOL--->HTTP/1.1 SERVER_PORT--->4000 HTTP_CONNECTION--->keep-alive HTTP_HOST--->127.0.0.1:4000 HTTP_COOKIE--->csrftoken=0SMasfbAKS3PQIfs2p6JBoRfbphIHVF0; _ga=GA1.1.522430974.1483446120 wsgi.multithread--->True SCRIPT_NAME---> HTTP_ACCEPT_LANGUAGE--->zh-CN,zh;q=0.8 HTTP_REFERER--->http://127.0.0.1:4000/?name=wanglei&age=23&name=duanyu
解析QUERY_STRING
from urllib.parse import parse_qs def application(environ, start_response): query_string = environ.get(‘QUERY_STRING‘) params = parse_qs(query_string) name = params.get(‘name‘, [‘unknown‘])[0] # 不管是不是单个值得到的值是列表 print(name) start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/plain‘)]) # headers return ["hello,world {}".format(name).encode()]
http ---> 代表schema
127.0.0.1 ---> 代表host
8000 -----> 代表port
/... ----> 代表path
?name=wanglei ---> 代表query_string
容器:线程/进程管理 生产中gunicorn容器很好 uwsgi容器必须要和Nginx配合才能使用
协议转换
输入转化为 environ 输出
头信息用start_response返回 body用return 返回
可能产生XSS攻击
from urllib.parse import parse_qs def application(environ, start_response): query_string = environ.get(‘QUERY_STRING‘) params = parse_qs(query_string) name = params.get(‘name‘, [‘unknown‘])[0] print(name) start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/html‘)]) # 当Content-Type 是 ‘text/html‘ 小心XSS攻击
return ["hello,world {}".format(name).encode()]
url : http://127.0.0.1:4000/?name=<script type=‘text/javascript‘ >alert(‘fuck%20it‘);</script>
from urllib.parse import parse_qs from html import escape def application(environ, start_response): query_string = environ.get(‘QUERY_STRING‘) params = parse_qs(query_string) name = params.get(‘name‘, [‘unknown‘])[0] print(name) start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/html‘)]) return ["hello,world {}".format(escape(name)).encode()] # escape函数将字符进行转义
webob封装Request对象 要传入environ参数
from webob import Request from html import escape def application(environ, start_response): request = Request(environ) # Request接受环境作为参数 params = request.params name = params.get(‘name‘, ‘unknown‘) # 默认是最后一个 http://127.0.0.1:4000/?name=wanglei&name=duanyu print(name) start_response(‘200 ok‘, [(‘Content-Type‘, ‘text/html‘)]) return ["hello,world {}".format(escape(name)).encode()] # 结果是 hello,world duanyu
webob封装Response对象
而且不用使用escape来转义字符(内部已经做了处理)
from webob import Request, Response def application(environ, start_response): request = Request(environ) params = request.params name = params.get(‘name‘, ‘unknown‘) print(name) response = Response() response.text = ‘<h1>hello,world {}</h1>‘.format(name) # 设置body体 response.status_code = 200 # 设置状态码 response.content_type = ‘text/html‘ # 设置头信息 return response(environ, start_response)
设置的头信息 可以不通过start_response里面来设置了 Response对象返回的时候需要传递 environ 和 start_response 参数
使用wsgify装饰器
作用: 1. 应用的参数只需要request, 返回Response即可
2.返回的时候不用再给Response传递 environ 和start_response 参数了
from webob import Request, Response from webob.dec import wsgify @wsgify def application(request: Request) -> Response: params = request.params name = params.get(‘name‘, ‘unknown‘) print(name) response = Response() # 只需要处理视图函数的最后返回一个 Response对象即可 response.text = ‘<h1>hello,world {}</h1>‘.format(name) response.status_code = 200 response.content_type = ‘text/html‘ return response
实现wsgify装饰器
from webob import Request, Response from functools import wraps def wsgify(fn): @wraps(fn) def wrapper(environ, start_response): # 装饰器中被装饰的函数不需要和装饰的函数参数保持一致 request = Request(environ) response = fn(request) # 相当于先执行下面的application函数 return response(environ, start_response) # 相当于在这里面给 response传参数 return wrapper @wsgify def application(request: Request) -> Response: params = request.params name = params.get(‘name‘, ‘unknown‘) print(name) response = Response() response.text = ‘<h1>hello,world {}</h1>‘.format(name) response.status_code = 200 response.content_type = ‘text/html‘ return response
最基本的路由
from webob import Request, Response from webob.dec import wsgify @wsgify def application(request: Request) -> Response: response = Response() if request.path == ‘/hello‘: response.text = ‘<h1>hello page</h1>‘ response.status_code = 200 response.content_type = ‘text/html‘ elif request.path == ‘/‘: response.text = ‘<h1>index page </h1>‘ response.status_code = 200 response.content_type = ‘text/html‘ return response
路由的基本思想
一个URL----->函数 将URL和视图函数分开写
静态路由: 现在开发者只需要关系怎么写应用就可以了
from webob import Request, Response from webob.dec import wsgify def hello(request: Request) -> Response: response = Response() response.text = ‘<h1>hello page </h1>‘ response.status_code = 200 response.content_type = ‘text/html‘ return response def index(request: Request) -> Response: return Response(body=‘<h1>index page</h1>‘, status=200, content_type=‘text/html‘) routes = { ‘/hello‘: hello, # 路由和函数映射的字典 ‘/‘: index } @wsgify def application(request: Request) -> Response: return routes[request.path](request) # --->注意这里要传递参数给视图函数
动态路由:
手动添加URL和视图函数的对应关系
from webob import Request, Response from webob.dec import wsgify def hello(request: Request) -> Response: response = Response() response.text = ‘<h1>hello page </h1>‘ response.status_code = 200 response.content_type = ‘text/html‘ return response def index(request: Request) -> Response: return Response(body=‘<h1>index page</h1>‘, status=200, content_type=‘text/html‘) class Application: routes = {} @classmethod def register(cls, path, handler): cls.routes[path] = handler @wsgify def __call__(self, request): return self.routes[request.path](request) if __name__ == ‘__main__‘: from wsgiref.simple_server import make_server Application.register(‘/hello‘, hello) # 注册路由 Application.register(‘/‘, index) server = make_server(‘0.0.0.0‘, 4000, app=Application()) # 这里要特别注意 当函数调用的时候才会 执行__call__方法体的内容 try: server.serve_forever() except KeyboardInterrupt: server.shutdown()
可以实现动态注册路由了 route字典 也就不需要了,不需要事先将对应关系写好了
改写动态路由:用类来改写
def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response def hello(request: Request) -> Response: params = request.params name = params.get(‘name‘, ‘unknown‘) response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ print(params) return response
class Application: route = {} # 路由和处理函数的对应关系 @classmethod def register(cls, path, handler): # 注册路由函数 cls.route[path] = handler @classmethod def default_handler(cls, request: Request) -> Response: #自定义 找不到路由时的处理方法 return Response(body=‘Not found‘, status=404) @wsgify def __call__(self, request: Request) -> Response: return self.route.get(request.path, self.error_handler)(request) # 没有注册的url就返回默认的 handler
webob中已经有了错误的方法 不需要自己编写404错误的函数
from webob import exc
@classmethod def error_handler(cls, request: Request) -> Response: # webob 的错误 raise exc.HTTPNotFound("NOT FOUND")
可以改写为如下,没必要再定义一个错误处理函数,直接出错了就捕获
class Application: route = {} @classmethod def register(cls, path): def inner(handler): cls.route[path] = handler return handler return inner @wsgify def __call__(self, request: Request) -> Response: try: return self.route[request.path](request) except KeyError: raise exc.HTTPNotFound(‘not found .......‘)
我们发现每次都要手动注册url,可以实现在定义视图函数的时候绑定url
class Application: route = {} @classmethod def register(cls, path): # 用装饰器改写注册方法,实现最便捷的注册路由 def inner(handler): cls.route[path] = handler return handler return inner @wsgify def __call__(self, request: Request) -> Response: try: return self.route[request.path](request) except KeyError: raise exc.HTTPNotFound(‘not found .......‘) app = Application() @app.register(‘/‘) # 这样就可以在定义视图的时候绑定url路由了 def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response @app.register(‘/hello‘) def hello(request: Request) -> Response: params = request.params name = params.get(‘name‘, ‘unknown‘) response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ print(params) return response
使用正则表达式匹配 让路由更加具有表现性
class Application: route = {} @classmethod def register(cls, pattern): def inner(handler): cls.route[pattern] = handler return handler return inner @wsgify def __call__(self, request: Request) -> Response: for pattern, handler in self.route.items(): if re.match(pattern, request.path): return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.register(‘^/$‘) # 因为正则表达式匹配是贪婪模式的 所以需要加上 前后牟定 否则这里会全部匹配 def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response @app.register(‘^/hello$‘) # 这里也需要加上牟定 def hello(request: Request) -> Response: params = request.params name = params.get(‘name‘, ‘unknown‘) response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ print(params) return response
事先编译好pattern
class Application: route = [] # 使用列表 而不是字典 @classmethod def register(cls, pattern): def inner(handler): cls.route.append((re.compile(pattern), handler)) # 事先编译好 return handler return inner @wsgify def __call__(self, request: Request) -> Response: for pattern, handler in self.route: if pattern.match(request.path): return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.register(‘^/$‘) def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response
将正则捕获到的参数 放到 request.args 和 request.kwargs里面(可以使用request.kwargs.name来访问字段)
class _Vars: def __init__(self, data=http://www.mamicode.com/None): if data is not None: self._data = data # 注意改名 else: self._data = {} def __getattr__(self, item): try: return self._data[item] # 这样就可以实现request.kwargs.name 这种来访问字段了 except KeyError: raise AttributeError("NO attribute {}".format(item)) def __setattr__(self, key, value): if key != ‘_data‘: raise NotImplemented self.__dict__[‘_data‘] = value # 表示不能对他属性进行修改 class Application: route = [] # 使用列表 而不是字典 @classmethod def register(cls, pattern): def inner(handler): cls.route.append((re.compile(pattern), handler)) return handler return inner @wsgify def __call__(self, request: Request) -> Response: for pattern, handler in self.route: m = pattern.match(request.path) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.register(‘^/$‘) def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘) return response @app.register(‘^/hello/(\w+)$‘) # 测试 args http://127.0.0.1:3001/hello/wanglei def hello(request: Request) -> Response: name = request.args[0] response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ return response
@app.register(‘^/hello/(?P<name>\w+)$‘) # 测试 kwargs http://127.0.0.1:3001/hello/wanglei def hello(request: Request) -> Response: name = request.kwargs.name response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ return response
将method也加进匹配规则里面
把 method 也作为匹配的条件 class _Vars: def __init__(self, data=http://www.mamicode.com/None): if data is not None: self._data = data # 注意改名 else: self._data = {} def __getattr__(self, item): try: return self._data[item] except KeyError: raise AttributeError("NO attribute {}".format(item)) def __setattr__(self, key, value): if key != ‘_data‘: raise NotImplemented self.__dict__[‘_data‘] = value class Application: route = [] # 使用列表 而不是字典 @classmethod def register(cls, pattern, method=‘GET‘): # 只需要在这里加一个 就行了 默认请求方法是GET def inner(handler): cls.route.append((method, re.compile(pattern), handler)) return handler return inner @wsgify def __call__(self, request: Request) -> Response: for method, pattern, handler in self.route: if method.upper() != request.method: continue m = pattern.match(request.path) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) # 也可以使用命名元组来封装 return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.register(‘^/$‘) def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response @app.register(‘^/hello/(?P<name>\w+)$‘,method=‘POST‘) # 测试 kwargs http://127.0.0.1:3001/hello/duanyu def hello(request: Request) -> Response: # params = request.params name = request.kwargs.name response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ return response
实现常见请求方法的注册路由:
请求方法的路由: get put post delete head options
class _Vars: def __init__(self, data=http://www.mamicode.com/None): if data is not None: self._data = data # 注意改名 else: self._data = {} def __getattr__(self, item): try: return self._data[item] except KeyError: raise AttributeError("NO attribute {}".format(item)) def __setattr__(self, key, value): if key != ‘_data‘: raise NotImplemented self.__dict__[‘_data‘] = value class Application: ROUTE = [] # 使用列表 而不是字典 @classmethod def router(cls, pattern, method=‘GET‘): # 只需要在这里加一个 就行了 def inner(handler): cls.ROUTE.append((method, re.compile(pattern), handler)) return handler return inner @classmethod def get(cls, pattern): # 可以通过描述器来实现 不用写7次了 return cls.router(pattern, method=‘GET‘) @classmethod def post(cls, pattern): return cls.router(pattern, method=‘POST‘) @classmethod def put(cls, pattern): return cls.router(pattern, method=‘PUT‘) @classmethod def head(cls, pattern): return cls.router(pattern, method="HEAD") @classmethod def delete(cls, pattern): return cls.router(pattern, method=‘DELETE‘) @classmethod def options(cls, pattern): # 可以通过描述器来实现 不用写7次了 return cls.router(pattern, method=‘OPTIONS‘) @wsgify def __call__(self, request: Request) -> Response: for method, pattern, handler in self.ROUTE: if method.upper() != request.method: continue m = pattern.match(request.path) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) # 也可以使用命名元组来封装 return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.get(‘^/$‘) def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response
一个url可以有多个请求方法
class _Vars: def __init__(self, data=http://www.mamicode.com/None): if data is not None: self._data = data else: self._data = {} def __getattr__(self, item): try: return self._data[item] except KeyError: raise AttributeError("NO attribute {}".format(item)) def __setattr__(self, key, value): if key != ‘_data‘: raise NotImplemented self.__dict__[‘_data‘] = value class Application: ROUTE = [] # 使用列表 而不是字典 @classmethod def router(cls, pattern=‘.*‘, methods=None): def inner(handler): cls.ROUTE.append((methods, re.compile(pattern), handler)) return handler return inner @classmethod def get(cls, pattern): return cls.router(pattern, methods=‘GET‘) @classmethod def post(cls, pattern): return cls.router(pattern, methods=‘POST‘) @classmethod def put(cls, pattern): return cls.router(pattern, methods=‘PUT‘) @classmethod def head(cls, pattern): return cls.router(pattern, methods="HEAD") @classmethod def delete(cls, pattern): return cls.router(pattern, methods=‘DELETE‘) @classmethod def options(cls, pattern): return cls.router(pattern, methods=‘OPTIONS‘) @wsgify def __call__(self, request: Request) -> Response: for method, pattern, handler in self.ROUTE: if method: # 如果传递的不是默认值 if isinstance(method, (list, tuple, set)): # 以列表/元组/集合的方式来指定请求方法 if request.method not in method: continue elif isinstance(method, str): if request.method != method: continue m = pattern.match(request.path) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.router(‘^/$‘) # 为了更大限度的自由 def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response @app.router(‘^/hello/(?P<name>\w+)$‘,methods=(‘GET‘, ‘POST‘)) # 测试 get 和 post 函数是否工作 def hello(request: Request) -> Response: # params = request.params name = request.kwargs.name response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ return response
对上面的优化
class _Vars: def __init__(self, data=http://www.mamicode.com/None): if data is not None: self._data = data # 注意改名 else: self._data = {} def __getattr__(self, item): try: return self._data[item] except KeyError: raise AttributeError("NO attribute {}".format(item)) def __setattr__(self, key, value): if key != ‘_data‘: raise NotImplemented self.__dict__[‘_data‘] = value class Application: ROUTE = [] # 使用列表 而不是字典 @classmethod def router(cls, pattern=‘.*‘, methods=None): def inner(handler): cls.ROUTE.append((methods, re.compile(pattern), handler)) return handler return inner @classmethod def get(cls, pattern): # 可以通过描述器来实现 不用写7次了 return cls.router(pattern, methods=‘GET‘) @classmethod def post(cls, pattern): return cls.router(pattern, methods=‘POST‘) @classmethod def put(cls, pattern): return cls.router(pattern, methods=‘PUT‘) @classmethod def head(cls, pattern): return cls.router(pattern, methods="HEAD") @classmethod def delete(cls, pattern): return cls.router(pattern, methods=‘DELETE‘) @classmethod def options(cls, pattern): return cls.router(pattern, methods=‘OPTIONS‘) @wsgify def __call__(self, request: Request) -> Response: for method, pattern, handler in self.ROUTE: if method: # 如果传递的不是默认值 if isinstance(method, (list, tuple, set))and request.method not in method: # 上面这里的判断层次太深了 continue elif isinstance(method, str) and request.method != method: continue m = pattern.match(request.path) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) return handler(request) raise exc.HTTPNotFound(‘NOT FOUND‘) app = Application() @app.router(‘^/$‘) # 为了更大限度的自由 def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response @app.router(‘^/hello/(?P<name>\w+)$‘,methods=(‘GET‘, ‘POST‘)) # 测试 get 和 post 函数是否工作 def hello(request: Request) -> Response: # params = request.params name = request.kwargs.name response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ return response
实现分组路由
class _Vars: def __init__(self, data=http://www.mamicode.com/None): if data is not None: self._data = data else: self._data = {} def __getattr__(self, item): try: return self._data[item] except KeyError: raise AttributeError("NO attribute {}".format(item)) def __setattr__(self, key, value): if key != ‘_data‘: raise NotImplemented self.__dict__[‘_data‘] = value class Router: def __init__(self, prefix=‘‘): self.__prefix = prefix.rstrip(‘/‘)
self._routes = [] @property def prefix(self): # 不能修改这个前缀 return self.__prefix def route(self, pattern=‘.*‘, method=None): def wrap(handler): self._routes.append((method, re.compile(pattern), handler)) return handler return wrap def get(self, pattern=‘.*‘): return self.route(pattern, ‘GET‘) def put(self, pattern=‘.*‘): return self.route(pattern, ‘PUT‘) def post(self, pattern=‘.*‘): return self.route(pattern, ‘POST‘) def head(self, pattern=‘.*‘): return self.route(pattern, ‘HEAD‘) def delete(self, pattern=‘.*‘): return self.route(pattern, ‘DELETE‘) def options(self, pattern=‘.*‘): return self.route(pattern, ‘OPTIONS‘) def patch(self, pattern=‘.*‘): return self.route(pattern, ‘PATCH‘) def run(self, request: Request): if not request.path.startswith(self.prefix): return for method, pattern, handler in self._routes: if method: # 如果传递的不是默认值 if isinstance(method, (list, tuple, set)) and request.method not in method: continue elif isinstance(method, str) and request.method != method: continue print(method, request.method) m = pattern.match(request.path.replace(self.prefix, ‘‘, 1)) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) return handler(request) class Application: ROUTE = [] @classmethod def register(cls, router: Router): cls.ROUTE.append(router) @wsgify def __call__(self, request: Request) -> Response: for router in self.ROUTE: response = router.run(request) if response: return response raise exc.HTTPNotFound(‘NOT FOUND‘) shop = Router(prefix=‘/shop‘) @shop.get(‘^/$‘) def index(request: Request) -> Response: response = Response(body=‘hello,world‘, content_type=‘text/html‘, ) return response @shop.get(‘^/hello/(?P<name>\w+)$‘) # 测试 get 和 post 函数是否工作 def hello(request: Request) -> Response: name = request.kwargs.name response = Response() response.text = ‘hello,world {}‘.format(name) response.status_code = 200 response.content_type = ‘text/plain‘ return response Application.register(router=shop) if __name__ == ‘__main__‘: from wsgiref.simple_server import make_server server = make_server(‘0.0.0.0‘, 3001, Application()) server.serve_forever()
封装框架1