Python的web开发
Posted Python爱好者666
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python的web开发相关的知识,希望对你有一定的参考价值。
一、Web开发
Tcp udp
Cs即客户端、服务器端编程,客户端和服务器端之间需要使用socket,约定协议、版本(协议使用的是tcp或者udp)。Tcp协议和udp协议,指定地址和端口,就可以通信了。
客户端、服务器端传输数据,数据可以有一定的格式,双方必须先约定好。
1、BS
Bs:在tcp协议之上的http协议开发的,浏览器和服务器,是基于cs开发之上的。
Browser、server开发
Browser浏览器,一种特殊的客户端,支持HTTP(s)协议,能够通过URL向服务器端发起请求,等待服务端返回HTML,并在浏览器内可视化展示的程序。
server,支持HTTP(s)协议,能够介绍众多客户端发起的HTTP协议请求,经过处理,将HTML等数据返回给浏览器。
是特殊的cs,即客户端必须是一种支持HTTP协议且能解析并渲染HTML的软件,服务器端必须是能够接受多客户端http访问的软件服务器。
http协议底层是基于tcp协议实现的。
2、分类
客户器端开发:HTML,css,Javascript
服务器端开发:Python有wsgi,Django,flask,tornado。
3、HTTP协议
1)简介
安装httpd
#yum install httpd
HTTP协议是无状态协议。
同一个客户端的两次请求之间没有任何关系,从服务器端角度来说,不知道两个请求是来自同一个客户端。
2)cookie
键值对信息,浏览器发起每一请求,都会把cookie信息发给服务器端,是一种客户端、服务器端传递数据的技术。
服务器端可以通过判断这些信息,来确定这次请求是否和之前的请求有关联,
一般来说cookie信息是在服务器端生成的,返回给客户端。
客户端可以自己设置cookie信息。
3)URL
URL就是地址,统一资源定位符uniform resource locator。每一个连接指向一个资源供客户端访问。
URL组成:方言,host,path路径,查询字符串。
访问静态资源的时候,通过URL访问的是网站某路径下的index.html文件,而这个文件对应磁盘上的真实文件,就会从磁盘上读取这个文件, 把这个文件发回给浏览器。
Scheme模式、协议:
http、ftp、https、file、mailto等等,都是类似
host:port
www.xxxx.com:80默认端口80不写,域名会使用dns解析,
/path/to.resource
Path,指向资源的路径。
?key1=value&key2=value2
Query string,查询字符串,问号分割,后面的可以=value形式,用&符号分割。
4、HTTP消息
消息分为request和response
Request:浏览器向服务器端发起请求。
Response:服务器对客户端的请求响应。
请求和响应的消息都由请求行,header消息报头,body消息正文组成。
1)请求
(1)请求消息行:请求方法method、请求路径、协议版本、CRLF(回车换行)。Get / http /1.1
(2)请求方法method
Get:请求获取对应URL对应的资源。把请求的内容到放到了header的里面。
Post:提交数据至服务器端,请求报文头部。
Head:和get相似,不过不返回消息正文。
(3)常见传递信息的方式
使用get方法使用querystring
通过查询字符串在URL中传递参数。
Post方法提交数据。
URL本身就包含着信息
2)响应
响应消息行:协议版本,状态码,消息描述CRLF(回车换行)。 http/1.1 200 ok
状态码在响应第一行:
1xx:提示信息,表示请求已被成功接收,继续处理。
2xx:表示正常响应。
200.正常的返回了网页消息
3xx:重定向
301 页面永久移走,永久重定向,返回新的URL,浏览器会根据返回的URL发起新的request请求。
302 临时重定向
304资源未修改,浏览器使用本地缓存。
4xx:客户端请求错误
404 not found 网页找不到,客户端请求的资源有错误。
400 请求语法错误
401 请求要求身份验证
403服务器拒绝请求。
5xx:服务器端错误
500 服务器内部错误
502上游服务器错误
5、无状态、有连接和短连接
无状态:服务器不知道两次连接请求的之间的关系,同一个浏览器两次访问,服务器端不知道两次之间是有联系的。后来可以根据cookie,session判断。
有连接:基于tcp协议,是面向连接的,需要3次握手,4次挥手。
短连接:http1.1之前,都是一个请求一个连接,而tcp的连接创建销毁成本太高,对服务器影响很大。
从http1.1开始,支持keep-live。默认开启,一个连接打开后,会保持一段时间(可以设置),浏览器再次访问该服务器就会使用这个tcp连接,减轻了服务器的压力,提高了效率。
6、wsgi
Wsgi 服务器的通用网关接口。基于http协议的server。
Wsgi主要规定了服务器端和应用程序间的接口
Python Web Server Gateway Interface
通用网关接口cgi
(1)重点
基于http协议,wsgi app和server之间的应用。
对应的函数进行处理。
不同的路径代表不同的请求。函数和请求之间的映射关系。
框架对应URL,
Server把处理逻辑提出去,提给app
浏览器发送请求给server,server把请求头解包,封装成为字典,调用wsgi的app,app进行逻辑处理,处理完毕后直接返回状态码和报文头给浏览器,返回正文到wsgi到server,server发送http协议正文给浏览器。
from wsgiref.util import setup_testing_defaults
from wsgiref.simple_server import make_server
# A relatively simple WSGI application. It‘s going to print out the
# environment dictionary after being updated by setup_testing_defaults
def simple_app(environ, start_response):
setup_testing_defaults(environ)
status = ‘200 OK‘
headers = [(‘Content-type‘, ‘text/plain; charset=utf-8‘)]
start_response(status, headers) #返回正文之间必须先返回状态码
ret = [("%s: %s " % (key, value)).encode("utf-8")
for key, value in environ.items()]
return ret
httpd = make_server(‘‘, 8000, simple_app)
print("Serving on port 8000...")
httpd.serve_forever()
7、wsgi服务器 ----wsgiref
Wsgiref.simple_server模块实现简单的wsgi http服务器端。
Wsgiref.Simple_server. make_server(ip,port,demo_app,server_class=wsgisercer,handler_class=wsgirequesthandler)
启动一个wsgi服务器。
Wsgiref.simple_server.demo_app(environ,start_response)一个函数,完成了wsgi的应用程序端。
###简单的服务器端代码:
from wsgiref.simple_server import make_server,demo_app
ip = ‘127.0.0.1‘
port = 9999
server = make_server(ip,port,demo_app)# demo_app应用程序,可调用
server.serve_forever()
Wsgi服务器的作用:
监听HTTP服务端口(tcpserver,默认端口80)
接受浏览器端的HTTP请求并解析封装成environ环境数据。
负责调用应用程序,将environ和start_response方法传入。
将应用程序响应的正文封装成http响应报文返回浏览器端。
8、wsgi app应用程序端
1)应用程序应该是一个可调用对象。
2)这个可调用对象接收两个参数。
3)可调用对象,返回的是一个可迭代对象。函数和class__call__返回值是[] class中__init__ 要有yield.
def application(environ,start_response):
pass
class Application:
def __init__(self,environ,start_response):
pass
def __iter__(self):
yield self.ret
class Application:
def __call__(self, environ,start_response):
pass
app程序端三种实现:
from wsgiref.simple_server import make_server,demo_app
#
def application(environ,start_response):
status = ‘200 ok‘
headers = [(‘Content-Type‘,‘text/html;charset=utf-8‘)]
start_response(status,headers)
ret_sre1 = b‘welcome to this ‘
ret_sre = ‘{}‘.format(‘welcome‘).encode()
return [ret_sre]
class Application:
def __init__(self,environ,start_response):
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status, headers)
ret_sre1 = b‘welcome to this ‘
# ret_sre = ‘{}‘.format(‘welcome‘).encode()
self.ret
= ret_sre1
def __iter__(self):
yield self.ret
#
class Application:
def __call__(self, environ,start_response):
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status, headers)
ret_sre1 = b‘welcome to this ‘
# ret_sre = ‘{}‘.format(‘welcome‘).encode()
return [ret_sre1]
ip = ‘127.0.0.1‘
port = 9999
server = make_server(ip,port,Application())# demo_app应用程序,可调用
server.serve_forever()
8、environ
是包含HTTP请求信息的dict对象
名称 |
含义 |
REQUEST_METHOD |
请求方法,get、post等 |
PATH_INFO |
URL中的路径部分 |
QUERY_STRING |
查询字符串 |
SERVER_NAME,SERVER_POST |
服务器名、端口 |
HTTP_HOST |
地址和端口 |
SERVER_PROTOCOL |
协议 |
HTTP_USER_AGENT |
Useragent信息 |
9、start_response
是一个可调用对象,有3个参数,
Status:状态码和描述语。
Response_headers:可迭代对象,是一个元素为二元组的列表。[(‘Content-Type‘, ‘text/html;charset=utf-8‘)] 响应头部
Exc_info=None:在错误处理的时候使用.
start_reaponse 应该在返回可迭代对象之前调用,因为其返回的是response header,返回的可迭代对象是response body。
10、服务器端
服务器程序需要调用符合上述定义的可调用对象app,传入environ、start_response,app处理后,返回响应头和可迭代对象的正文,由服务器封装返回浏览器端。
from wsgiref.simple_server import make_server
def application(*args):
environ = args[0]
start_response = args[1]
for k,v in environ.items():
print(k,v)
print(‘====================‘)
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status,headers)
ret = [("%s :%s
" % (key,value)).encode(‘utf-8‘) for key,value
in environ.items()]
return ret
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,application)
server.serve_forever()
测试命令:curl -I 使用head方法。
curl -X 指定方法,-d传输数据
11、Web服务器
本质上就是一个tcp协议,监听在特定的端口上
支持HTTP协议,能够将HTTP请求的报文进行解析,能够把响应数据进行HTTP协议的报文封装并返回浏览器。
实现wsgi协议,协议约定了和应用程序之间接口。
HTTP协议是应用层。
12、app应用程序
遵从wagi协议
本身是一个可调用对象
调用start_response,返回响应头部。
返回包含正文的可迭代对象。
响应的是调用app的调用,response的是status,headers。
app的两个参数environ和start_response
二、flask框架的实现
1、wsgi请求environment处理
Wsgi服务器会帮忙处理http请求报文,但是提供的environ是一个用起来很不方便的字典。
2、query_string查询字符串的解析
from wsgiref.simple_server import make_server
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get(‘QUERY_STRING‘)
print(query_string)
#1字典
# d = {}
# for item in
query_string.split(‘&‘):
#
k,_,v = item.partition(‘=‘)
#
d[k] = v
#
print(‘k={},v={}‘.format(k,v))
#
print(d)
#2字典解析式
d = {k:v for k,_,v in map(lambda
item:item.partition(‘=‘),query_string.split(‘&‘))}
print(d)
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status,headers)
ret = ‘welcome to {}‘.format(‘item‘).encode()
return [ret]
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
name=tome&age=3&age=2
127.0.0.1 - - [26/Jun/2018 16:40:14] "POST /xxx?name=tome&age=3&age=2 HTTP/1.1" 200 15
{‘name‘: ‘tome‘, ‘age‘: ‘2‘}
3、cgi模块查询
import cgi
from wsgiref.simple_server import make_server
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get(‘QUERY_STRING‘)
print(query_string)
#1字典
# d = {}
# for item in query_string.split(‘&‘):
#
k,_,v = item.partition(‘=‘)
#
d[k] = v
#
print(‘k={},v={}‘.format(k,v))
#
print(d)
#2字典解析式
# d = {k:v for k,_,v in map(lambda
item:item.partition(‘=‘),query_string.split(‘&‘))}
# print(d)
qs = cgi.parse_qs(query_string)
print(qs)
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status,headers)
ret = ‘welcome to {}‘.format(‘item‘).encode()
return [ret]
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
name=tome&age=3&age=2
127.0.0.1 - - [26/Jun/2018 17:19:06] "POST /xxx?name=tome&age=3&age=2 HTTP/1.1" 200 15
{‘age‘: [‘3‘, ‘2‘], ‘name‘: [‘tome‘]}
Cgi模块value的值是列表,因为value可以有多个值。
3、urllib库
urllib:请求解析重要的库。
多值是age=1&age=2这个是多值。
age=1,2只是一个值。
Parse_se函数,将同一个名称的多值,保存在字典中,使用列表保存。
import cgi
from urllib.parse import parse_qs
from wsgiref.simple_server import make_server
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get(‘QUERY_STRING‘)
print(query_string)
#1字典
# d = {}
# for item in query_string.split(‘&‘):
# k,_,v = item.partition(‘=‘)
# d[k] = v
# print(‘k={},v={}‘.format(k,v))
# print(d)
#2字典解析式
# d = {k:v for k,_,v in map(lambda item:item.partition(‘=‘),query_string.split(‘&‘))}
# print(d)
# qs = cgi.parse_qs(query_string)
# print(qs)
qs = parse_qs(query_string)
print(qs)
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status,headers)
ret = ‘welcome to {}‘.format(‘item‘).encode()
return [ret]
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
name=tome&age=3&age=2
{‘age‘: [‘3‘, ‘2‘], ‘name‘: [‘tome‘]}
127.0.0.1 - - [26/Jun/2018 17:28:24] "POST /xxx?name=tome&age=3&age=2 HTTP/1.1" 200 15
name=tome&age=3&age=2,3
{‘age‘: [‘3‘, ‘2,3‘], ‘name‘: [‘tome‘]}
4、webob库
环境数据有很多,都是存在字典中,字典的存取方式没有对象的属性访问方便。
利用第三方库webob库,可以把环境数据的解析、封装成为类。
1)webob简介
安装pip install webob
官方文档:https://docs.pylonsproject.org/projects/webob/en/stable/#webob
2)webob.request对象
from wsgiref.simple_server import make_server
import webob
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get(‘QUERY_STRING‘)
#print(query_string)
request = webob.Request(environ)
print(1,request.headers)
print(2,request.method)
print(3,request.path)
print(4,request.query_string)
print(5,request.GET)
print(6,request.POST)
print(‘params={}‘.format(request.params))
status = ‘200 ok‘
headers = [(‘Content-Type‘, ‘text/html;charset=utf-8‘)]
start_response(status,headers)
ret = ‘welcome to {}‘.format(‘item‘).encode()
return [ret]
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
1 <webob.headers.EnvironHeaders object at 0x00000050EDB88F60>
2 POST
3 /xxx
4 name=tome&age=3&age=2,3
5 GET([(‘name‘, ‘tome‘), (‘age‘, ‘3‘), (‘age‘, ‘2,3‘)])
6 <NoVars: Not an HTML form submission (Content-Type: text/plain)>
params=NestedMultiDict([(‘name‘, ‘tome‘), (‘age‘, ‘3‘), (‘age‘, ‘2,3‘)])
127.0.0.1 - - [26/Jun/2018 17:46:55] "POST /xxx?name=tome&age=3&age=2,3 HTTP/1.1" 200 15
3)multidict
是允许一个key存多个值 的类型
from webob.multidict import MultiDict
md = MultiDict()
md.add(1,‘abc‘)
md.add(1,‘cde‘)
md.add(‘a‘,1)
md.add(‘a‘,2)
for pair in md.items():
print(pair)
print(md.getall(1))
print(md.get(‘a‘))
print(md.get(1))
(1, ‘abc‘)
(1, ‘cde‘)
(‘a‘, 1)
(‘a‘, 2)
[‘abc‘, ‘cde‘]
2
Cde
4)webob.response对象
from wsgiref.simple_server import make_server
import webob
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get(‘QUERY_STRING‘)
#print(query_string)
request = webob.Request(environ)
print(1,request.headers)
print(2,request.method)
print(3,request.path)
print(4,request.query_string)
print(5,request.GET)
print(6,request.POST)
print(‘params={}‘.format(request.params))
res = webob.Response()
# res.status_code
= 250
# print(res.content_type)
# print(res.status)
# print(res.headerlist)
# status = ‘200 ok‘
# headers = [(‘Content-Type‘,
‘text/html;charset=utf-8‘)]
# start_response(status,headers)
start_response(res.status,res.headerlist)
ret = ‘welcome to {}‘.format(‘item‘).encode(‘utf-8‘)
# res.body = ret
# return res(environ,start_response)
return [ret]
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
from wsgiref.simple_server import make_server
import webob
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get(‘QUERY_STRING‘)
#print(query_string)
request = webob.Request(environ)
print(1,request.headers)
print(2,request.method)
print(3,request.path)
print(4,request.query_string)
print(5,request.GET)
print(6,request.POST)
print(‘params={}‘.format(request.params))
res = webob.Response() #[(‘Content-Type‘,
‘text/html;charset=utf-8‘)]
res.status_code = 250 #自己可以设置属性
print(res.content_type)
#
print(res.status)
# print(res.headerlist)
# status = ‘200 ok‘
# headers = [(‘Content-Type‘,
‘text/html;charset=utf-8‘)]
# start_response(status,headers)
# start_response(res.status,res.headerlist)
ret = ‘welcome
to {}‘.format(‘item‘).encode(‘utf-8‘)
res.body = ret
return res(environ,start_response) 返回res
# return [ret]
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
5)webob.dec装饰器类型
Wsgify装饰器
https://docs.pylonsproject.org/projects/webob/en/stable/api/dec.html
from webob.dec import wsgify
import webob
from wsgiref.simple_server import make_server
@wsgify
def app(request:webob.Request):
res = webob.Response(‘welcome to item‘)
return res
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,app)
server.serve_forever()
wsgify装饰器装饰的函数应该具有一个参数,这个参数应该是webob。Request类型,是对字典environ的对象化后的实例。
返回值:
是一个webob. Response 类型实例。
可以是一个bytes类型实例,会被封装成webob. Response类型实例的body属性。
可以是一个字符串型实例,会被封装转换为bytes类型实例,然后会被封装成为webob. Response类型实例的body属性。
都会被封装成webob.Response
一个请求,一个响应:
from webob.dec import wsgify
import webob
from wsgiref.simple_server import make_server
@wsgify
def app(request:webob.Request):
res = webob.Response(‘welcome to item‘)
return res
def application(environ,start_response):
request =
webob.Request(environ)
res = webob.Response()
res.status_code = 200
ret = ‘welcome‘.encode(‘utd-8‘)
res.body = ret
return res(environ,start_response)
if __name__ == ‘__main__‘:
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,app)
try:
server.serve_forever()
except Exception
as e:
print(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
封装成类
from webob import Response,Request
from webob.dec import
wsgify
from wsgiref.simple_server import make_server
class App:
@wsgify
def __call__(self,request:Request):
return ‘welcome
to‘
if __name__ == ‘__main__‘:
ip = ‘127.0.0.1‘
port = 9000
server = make_server(ip,port,App())
try:
server.serve_forever()
except Exception
as e:
print(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
三、类flask框架实现
Restfull:request.post
1、路由route简介
路由:路怎么走,按照不同的路径分发数据。
URL代表对不同的资源的访问,认为请求不同的路径对应的数据。对动态网页,不同的路径对应不同的应用程序来处理,返回数据,用户以为是访问的静态网页。
不管是静态web还是动态web服务器,都需要路径和资源或处理程序的映射,最终返回HTML的文本。
静态web服务器,解决路径和文件之间的映射。
动态web服务器,解决路径和应用程序之间的映射。
所有框架都是如此,都是由路由配置。
路由的映射关系。
from webob.dec import wsgify
from wsgiref import
simple_server
from webob import
Request,Response
import logging
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
def indexhandler(request):
return ‘<h1></h1>‘
def pythonhandler(request):
return ‘<h2></h2>‘
class App:
@wsgify
def __call__(self, request:Request):
path = request.path
if path
== ‘/‘:
return indexhandler(request)
elif path
== ‘/python‘:
return pythonhandler(request)
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except Exception
as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
2、路由功能实现
路由功能,使用路由类实现。
路由类实现的功能主要就是完成path到handler函数的映射,使用字典保存合适。
理由映射建立注册方法,两个参数path和handler。
1)普通实现
from webob.dec import wsgify
from wsgiref import
simple_server
from webob import
Request,Response
import logging
from webob.exc import
HTTPNotFound
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Router:
ROUTERTABLE = {}
# @classmethod
def register(self,path,handler):
self.ROUTERTABLE[path]
= handler
return handler
def indexhandler(request):
return ‘<h1></h1>‘
def pythonhandler(request):
return ‘<h2></h2>‘
router = Router()
router.register(‘/‘,indexhandler)
router.register(‘/python‘,pythonhandler)
class App:
_Router = Router
@wsgify
def __call__(self, request:Request):
path = request.path
try:
return self._Router.ROUTERTABLE.get(path)(request)
except Exception
as e:
logging.info(e)
raise HTTPNotFound()
# if path == ‘/‘:
# return indexhandler(request)
# elif path == ‘/python‘:
# return pythonhandler(request)
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except Exception
as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
2)装饰器实现
from webob.dec import wsgify
from wsgiref import
simple_server
from webob import
Response,Request
import logging
from webob.exc import
HTTPNotFound
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Router:
ROUTERABLES = {}
@classmethod
def register(cls,path):
def _register(handler):
cls.ROUTERABLES[path] =
handler
return handler
return _register
@Router.register(‘/‘)
def indexhandler(request):
return ‘<h1></h1>‘
@Router.register(‘/python‘)
def pythonhandler(requset):
return ‘<h2></h2>‘
# router = Router()
# router.register(‘/‘,indexhandler)
# router.register(‘/python‘,pythonhandler)
class App:
_Router = Router
@wsgify
def __call__(self, request:Request):
path = request.path
try:
return self._Router.ROUTERABLES.get(path)(request)
except Exception
as e:
logging.info(e)
raise HTTPNotFound()
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except Exception
as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
3、404处理
Webob.exc提供了异常模块
使用webob.exc.HTTPNoteFound表示路由表找不到对象的处理函数。
4、正则表达式匹配模式
使用字符串匹配的模式,路径匹配很死板,使用正则表达式,可以更好的匹配路径,导入re模块。注册的时候传入模式pattern。而不是字符串。
类app中实现模式和传入的路径 的匹配。
因为字典是无序的,在遍历的匹配的时候需要用到有序的,所以采用列表,列表里面的元素用二元组。(编译后的正则对象,handler)
from webob.exc import HTTPNotFound
from webob.dec import
wsgify
import logging
from webob import
Request,Response
import re
from wsgiref import
simple_server
class Router:
ROUTERABLE = []
@classmethod
def register(cls,pattern):
def _register(hanler):
cls.ROUTERABLE.append((re.compile(pattern),hanler))
return hanler
return _register
@Router.register(‘^/$‘)
def indexhandler(request):
return ‘<h1></h1>‘
@Router.register(‘^python$‘)
def pythonhandler(request):
return ‘<h2></h2>‘
class App:
_Router = Router
@wsgify
def __call__(self, request:Request):
path = request.path
for pattern,handler in self._Router.ROUTERABLE:
if pattern.match(path):
return handler(request)
raise HTTPNotFound()
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except Exception
as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
分组捕获:
from webob.dec import wsgify
from webob.exc import
HTTPNotFound
from webob import
Request,Response
import re
import logging
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,pattern):
def _register(handler):
cls.ROUTERABLES.append((re.compile(pattern),handler))
return handler
return _register
@Rouer.register(‘^/$‘)
def indexhandler(request):
return ‘<h1></h1>‘
@Rouer.register(‘^/python$‘)
@Rouer.register(‘^/python/(?P<id>d+)$‘)
def pythonhandler(request):
return ‘<h2></h2>{}‘.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for pattern,handler
in self._Router.ROUTERABLES:
matcher = pattern.match(path)
if matcher:
request.groups =
matcher.groups() #动态增加属性
request.groupdict =
matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
5、method方法匹配
请求方法,即使是同一个URL,因为请求方法不同,处理方式也不同。
需要按照要求匹配请求方法之后才能决定执行什么函数。
方法 |
含义 |
Get |
请求指定的页面信息,并返回报头和正文 |
HEAD |
类似get请求,只不过返回的响应中没有具体内容,用于获取报头 |
Post |
向指定的资源提交数据进行出路请求(例如提交表单或者上传文件),数据被包含在请求报文中。 Post请求客户你能会导致新的资源建立或已有的资源修改 |
Put |
从客户端向服务器端传递的数据取代指定的文档的内容 |
Delete |
请求服务器删除指定的内容 |
增加注册属性method方法。
from webob.dec import wsgify
from webob.exc import
HTTPNotFound
from webob import
Request,Response
import re
import logging
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,method,pattern):
def _register(handler):
cls.ROUTERABLES.append((method.upper(),re.compile(pattern),handler))
return handler
return _register
@Rouer.register("GET",r‘^/$‘)
def indexhandler(request):
return ‘<h1></h1>‘
@Rouer.register("GET",r‘^/python$‘)
@Rouer.register("GET",r‘^/python/(?P<id>d+)$‘)
def pythonhandler(request):
return ‘<h2></h2>{}‘.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for method, pattern,handler in self._Router.ROUTERABLES:
if not method
== request.method.upper():
continue
matcher = pattern.match(path)
if matcher:
request.groups =
matcher.groups()
request.groupdict =
matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
get的方法和post方法:
from webob.dec import wsgify
from webob.exc import
HTTPNotFound
from webob import
Request,Response
import re
import logging
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,pattern,*method):
def _register(handler):
cls.ROUTERABLES.append((
tuple(map(lambda
x:x.upper(),method)),
re.compile(pattern),handler))
print(cls.ROUTERABLES)
return handler
return _register
@classmethod
def get(cls,pattern):
return cls.register(pattern,"GET")
@classmethod
def post(cls,pattern):
return cls.register(pattern,"POST")
# @Rouer.register(r‘^/$‘,‘GET‘) # 1注册一个
# @Rouer.register(r‘^/(?P<id>d+)$‘)
#method没有指定,支持所有的
# @Rouer.register(r‘^/$‘) #method没有指定,支持所有的
# @Rouer.register(r‘^/$‘,"GET","POST") # 1注册多个
@Rouer.get(r‘/‘)
def indexhandler(request):
return ‘<h1></h1>‘
# @Rouer.register(r‘^/python$‘,‘POST‘) #1注册一个
#
@Rouer.register(r‘^/python/(?P<id>d+)$‘,"GET","HEAD") #2注册多个get head
# # @Rouer.register(r‘^/python/(?P<id>d+)$‘) #3不指定,all
@Rouer.post(r‘/python‘)
def pythonhandler(request):
return ‘<h2></h2>{}‘.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for method, pattern,handler in self._Router.ROUTERABLES:
if not method
or request.method.upper() in method:
matcher =
pattern.match(path)
if matcher:
request.groups =
matcher.groups()
request.groupdict =
matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
map函数转换的结果是惰性求值的,利用tuple。
from webob.dec import wsgify
from webob.exc import
HTTPNotFound
from webob import
Request,Response
import re
import logging
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,pattern,*method):
def _register(handler):
cls.ROUTERABLES.append((
tuple(map(lambda
x:x.upper(),method)),
re.compile(pattern),handler))
print(cls.ROUTERABLES)
return handler
return _register
# @Rouer.register(r‘^/$‘,‘GET‘) # 1注册一个
# @Rouer.register(r‘^/(?P<id>d+)$‘)
#method没有指定,支持所有的
# @Rouer.register(r‘^/$‘) #method没有指定,支持所有的
@Rouer.register(r‘^/$‘,"GET","POST") # 1注册多个
def indexhandler(request):
return ‘<h1></h1>‘
# @Rouer.register(r‘^/python$‘,‘POST‘) #1注册一个
@Rouer.register(r‘^/python/(?P<id>d+)$‘,"GET","HEAD") #2注册多个get head
# # @Rouer.register(r‘^/python/(?P<id>d+)$‘) #3不指定,all
def pythonhandler(request):
return ‘<h2></h2>{}‘.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for method, pattern,handler in
self._Router.ROUTERABLES:
if not method or
request.method.upper() in method:
matcher =
pattern.match(path)
if matcher:
request.groups =
matcher.groups()
request.groupdict =
matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
注册多个URL的method。
四、类flask框架
1、路由分组:
路由分组,就是按照前缀分别映射。
常见的一级目录:
/admin后台管理。 /product 这些目录都是根目录下的第一级。
两次注册:路由注册,和实例的注册。
from webob.dec import wsgify
from webob.exc import
HTTPNotFound
from webob import
Request,Response
import logging
import re
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s %(threadName)s
%(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Router:
def __init__(self,predix:str=‘‘):
self.__prefix
= predix.rstrip(‘/\‘) #前缀的
self.__routables
= [] #存三元组,有序的列表
def route_register(self,pattern,*methods):
def _route_register(handler):
self.__routables.append((tuple(map(lambda x:x.upper(),methods)),
re.compile(pattern),handler))
return handler
return _route_register
def get(self,pattern):
return self.route_register(pattern,‘GET‘)
def post(self,pattern):
return self.route_register(pattern,‘POST‘)
def head(self,pattern):
return self.route_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,patter,handler
in self.__routables:
if not method
or request.method.upper() in method:
matcher = patter.match(request.path.replace(self.__prefix
,‘‘,1))
if matcher:
request.groups =
matcher.groups()
request.groupdict =
matcher.groupdict()
return handler(request)
class App:
_ROUTES = []
@classmethod
def register(cls,*routers:Router):
for router
in routers:
cls._ROUTES.append(router)
@wsgify
def __call__(self,request):
for router
in self._ROUTES:
response =
router.match(request)
if response:#匹配返回非None的router对象
return response #匹配则返回
raise HTTPNotFound()
idx = Router()
py = Router(‘/python‘)
App.register(idx,py)
@idx.get(r‘^/$‘)
@idx.route_register(r‘^/(?P<id>d+)$‘)
def indexhandler(request):
return ‘<h1></h1>‘
@py.get(‘^/(w+)$‘)
@py.route_register(r‘^/(?P<id>d+)$‘)
def pythonhandler(request):
return ‘<h2></h2>‘
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except Exception
as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
2、字典的属性方法
属性变成字典的方式来进行访问。
让kwargs这个字典,不使用[]访问元素,而是使用.点号访问元素。如同属性一样访问。
class AttrDict:
def __init__(self,d):
self.__dict__.update(d)
def __setattr__(self, key, value):
raise NotImplementedError
d = {‘a‘:1,‘b‘:2}
obj = AttrDict(d)
print(obj.a)
加上属性字典:
from webob.exc import HTTPNotFound
from webob.dec import
wsgify
from webob import
Request,Response
import re
from wsgiref import
simple_server
import logging
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class 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 ‘<AttrDict{}>‘.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class Router:
def __init__(self,prefix:str=‘‘):
self._prefix
= prefix.rstrip(‘\/‘) #前缀
self.routerables
= [] #三元组
def router_ables(self,pattern,*methods):
def _router_ables(handler): #三元组
self.routerables.append((
tuple(map(lambda
x:x.upper(),methods)),
re.compile(pattern),handler))
return handler
return _router_ables
def get(self,pattern):
return self.router_ables(pattern,"GET")
def post(self,pattern):
return self.router_ables(pattern,"POST")
def head(self,pattern):
return self.router_ables(pattern,"HEAD")
def matcher(self,request:Request):
if not request.path.startswith(self._prefix):
return None
for method,pattern,handler
in self.routerables:
if not method
or request.method.upper() in method:
matcher =
pattern.match(request.path.replace(self._prefix,‘‘,1))
if matcher:
#匹配上
request.groups = matcher.groups()
request.groupdict =
AttrDict(matcher.groupdict()) #命名分组的字典被属性化
print(request.groupdict.id)
return handler(request)
class App: #一级注册函数
_ROUTERS = []
@classmethod
def app_register(cls,*routers:Router):
for router
in routers:
cls._ROUTERS.append(router)
@wsgify
def __call__(self, request:Request):
for router
in self._ROUTERS:
response =
router.matcher(request) #前缀匹配
if response:
return response #匹配上返回,没返回直接报错404
raise HTTPNotFound()
idx = Router()
py = Router(‘/python‘)
App.app_register(idx,py)
@idx.get(r‘^/$‘)
@idx.router_ables(r‘^/(?P<id>d+)$‘)
def indexhanler(request):
return ‘<h1></h1>‘
# @idx.get(r‘^/$‘)
# @idx.router_register(r‘^/(?P<id>d+)$‘)
# @py.get(‘^/(w+)$‘)
# @py.router_register(r‘^/(?P<id>d+)$‘)
@idx.get(r‘^/(w+)$‘)
@py.router_ables(r‘^/(?P<id>d+)$‘)
def pythonhandler(request):
return ‘<h2></h2>{}‘.format(request.groupdict.id)
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except Exception
as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
3、正则表达式化简
类型 |
含义 |
对应的正则 |
str |
不包含/的任意字符,默认类型 |
[^/]+ |
word |
字母和数字 |
w+ |
int |
纯数字,正负数 |
[+-]?d+ |
float |
正负号,数字,包含 |
[+-]?d+.d+ |
any |
包含/的任意字符。 |
.+ |
类型映射到正则表达式;就是使用format函数的填空模式。
TYPEPATTERNS = {
‘str‘ : r‘[^/]+‘,
‘word‘ : r‘w+‘,
‘int‘ : r‘[+-]?d+‘,
‘float‘: r‘[+-]?d+.d+‘,
‘any‘ : r‘.+‘
}
name = ‘id‘
t = ‘int‘
print(‘/(?P<{}>{})‘.format(name,TYPEPATTERNS.get(t,TYPEPATTERNS[‘str‘])))
捕获匹配用户的输入,捕获到分组名,然后对应正则表达式。
src = ‘/student/{name:str}/{id:int}‘
import re
# pattern = r‘/{(w+):(w+)}‘
pattern = r‘/{([^{}:]*):([^{}:]*)}‘
regix = re.compile(pattern)
matcher = regix.search(src)
def repl(mathcer):
name = matcher.group(1)
t = matcher.group(2)
return ‘/(?P<{}>{})‘.format(name,TYPEPATTERNS.get(t,TYPEPATTERNS[‘str‘]))
# print(matcher.group(1))
# print(matcher.group(2))
print(regix.sub(repl,src))
根据用户输入/{id:int}/{name:str} 把这些信息转化为对应的正则表达式。
from webob.dec import wsgify
from webob.exc import
HTTPNotFound
from webob import
Response,Request
import logging
import re
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class 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 ‘<AttrDict{}>‘.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r‘/{([^{}:]+):?([^{}:]*)}‘
_regex = re.compile(pa)
TYPEPATTERNS = {
‘str‘: r‘[^/]+‘,
‘word‘: r‘w+‘,
‘int‘: r‘[+-]?d+‘,
‘float‘: r‘[+-]?d+.d+‘,
‘any‘: r‘.+‘
}
def repl(self,matcher):
name = matcher.group(1)
t = matcher.group(2)
return ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS[‘str‘]))
def pase(self,src):
return self._regex.sub(self.repl,src)
def __init__(self,prefix:str=‘‘):
self.__prefix
= prefix.rstrip(‘\/‘)
self.__ROUTERABLES
= []
def router_register(self,rule,*method):
def _router_register(handler):
self.__ROUTERABLES.append(
(tuple(map(lambda
x:x.upper(),method)),
re.compile(self.pase(rule)),handler))
print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,‘GET‘)
def post(self,pattern):
return self.router_register(pattern,‘POST‘)
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,pattern,handler
in self.__ROUTERABLES:
if not method
or request.method.upper() in method:
matcher =
pattern.match(request.path.replace(self.__prefix,‘‘,1))
if matcher:
# request.groups = matcher.groups()
#
print(2,request.groups)
request.groupdict =
AttrDict(matcher.groupdict())
print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable
in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
for routerable
in self._ROUTER:
response =
routerable.match(request)
if response:
return response
raise HTTPNotFound()
idx = RouterAbles(‘/‘)
py = RouterAbles(‘/python‘)
App.app_regsieter(idx,py)
# @idx.get(r‘^/$‘)
# @idx.router_register(r‘^/(?P<id>d+)$‘)
@idx.router_register(r‘^/{id:int}$‘)
def handler(request):
id = ‘‘
if request.groupdict:
id =
request.groupdict.id
return ‘<h1></h1>{}‘.format(request.groupdict.id)
# @py.get(r‘^/$‘)
# @py.get(‘^/(w+)$‘)
# @py.router_register(r‘^/(?P<id>d+)$‘)
@py.router_register(r‘^/{id}$‘)
def pythonhandler(request):
if request.groupdict:
id = request.groupdict.id
return ‘<h2></h2>{}‘.format(id)
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
访问方式:
127.0.0.1:9000/python/1234
127.0.0.1:9000/123
上例中缺点是捕获的信息sub函数捕获的全部为str。转化的话。需要改造sub函数。
方法:1、路由注册的时候,将用户指定的rule(@py.router_register(r‘^/{id}$‘))添加入路由表三元组中。添加的时候调用parse函数。
2、parse函数中的返回值为:模式匹配sub匹配用户rule中的name和类型,sub(两个参数,调用另外一个函数depl,rule)。
3、depl函数中,根据信息。提前指定的表格,按照名字和类型,将类型转化为对应的正则表达式类型。
用户输入三种的形式的匹配,没写的话采用默认值。
拼接字符串的形式:
改造parse函数的代码
from webob.dec import
wsgify
from webob.exc import
HTTPNotFound
from webob import
Response,Request
import logging
import re
from wsgiref import
simple_server
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class 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 ‘<AttrDict{}>‘.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r‘/{([^{}:]+):?([^{}:]*)}‘
_regex = re.compile(pa)
TYPEPATTERNS = {
‘str‘: r‘[^/]+‘,
‘word‘: r‘w+‘,
‘int‘: r‘[+-]?d+‘,
‘float‘: r‘[+-]?d+.d+‘,
‘any‘: r‘.+‘
}
# def repl(self,matcher):
#
print(matcher,‘]]]‘)
#
name = matcher.group(1)
#
t = matcher.group(2)
#
s1 = ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t,
self.TYPEPATTERNS[‘str‘]))
#
print(‘----‘,s1)
#
return s1
#
# def pase(self,rule):
#
s2= self._regex.sub(self.repl,rule)
#
print(‘+-+-‘,rule)
#
print(‘=====‘,s2)
#
return s2
def parse(self,src: str):
start = 0
repl = ‘‘
types = {}
TYPECAST = {
‘str‘: str,
‘word‘: str,
‘int‘: int,
‘float‘: float,
‘any‘: str
}
matchers = self._regex.finditer(src)
for i, matcher in enumerate(matchers):
name = matcher.group(1)
t = matcher.group(2)
types[name] = TYPECAST.get(t, str)
repl +=
src[start:matcher.start()]
tmp = ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS[‘str‘]))
repl += tmp
start = matcher.end()
else:
repl += src[start:]
return repl, types
def __init__(self,prefix:str=‘‘):
self.__prefix
= prefix.rstrip(‘\/‘)
self.__ROUTERABLES
= []
def router_register(self,rule,*method):
def _router_register(handler):
pattern ,types
= self.parse(rule)
self.__ROUTERABLES.append(
(tuple(map(lambda
x:x.upper(),method)),
re.compile(pattern),types,handler))
print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,‘GET‘)
def post(self,pattern):
return self.router_register(pattern,‘POST‘)
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,pattern,trans,handler in self.__ROUTERABLES:
if not method
or request.method.upper() in method:
matchers =
pattern.match(request.path.replace(self.__prefix,‘‘,1))
if matchers:
newdict ={}
for k,v in matchers.groupdict().items():
newdict[k] = trans[k](v)
# request.groups = matcher.groups()
#
print(2,request.groups)
request.vars =
AttrDict(matchers.groupdict())
# print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable
in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
for routerable
in self._ROUTER:
response =
routerable.match(request)
if response:
return response
raise HTTPNotFound()
idx = RouterAbles(‘/‘)
py = RouterAbles(‘/python‘)
App.app_regsieter(idx,py)
# @idx.get(r‘^/$‘)
# @idx.router_register(r‘^/(?P<id>d+)$‘)
@idx.router_register(r‘^/{id:int}$‘)
def handler(request):
id = ‘‘
if request.vars:
id = request.vars.id
return ‘<h1></h1>{}‘.format(id)
# @py.get(r‘^/$‘)
# @py.get(‘^/(w+)$‘)
# @py.router_register(r‘^/(?P<id>d+)$‘)
@py.router_register(r‘^/{id}$‘)
def pythonhandler(request):
if request.vars:
id = request.vars.id
return ‘<h2></h2>{}‘.format(id)
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
app中,可以使用字典保存所有的router实例。因为每一个router的实例的前缀不同,完全使用前缀为key,router实例为value组成字典。,以后在call方法中,就不需要遍历列表了。只是需要提取request的path的前缀就可以和字典的key匹配了。提高了效率。
4、框架处理流程
客户端发来http请求,被wsgi服务器处理后传递给app的__call__方法。
App中遍历已经注册的router ,router的match来判断自己能不能处理,前缀匹配,看到注册的规则。(规则是装饰器已经转换成了命名分组的正则表达式了)
如果由某一个注册的正则表达式匹配,就把回去的参数放到request中,调用注册时候的映射handler传入request。
Handler处理以后,返回response。App中拿到response的数据,返回给wsgi。
Handler返回的仅仅是数据,将数据填入HTML中,将新生成的HTML字符串返回给客户端,就是网页技术。模板技术。
五、flask实现
1、模板
HTML:就是格式和数据混在一起的超文本。数据格式的描述。
XML:数据描述。
动态数据很难缓存。缓存是没有用的。
import re
from io import StringIO,BytesIO
d = {‘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)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>wcl</title>
</head>
<body>
显示数据<br>
{{id}}{{name}}{{age}}
</body>
</html>
渲染后的代码:
2 <_sre.SRE_Match object; span=(3, 9), match=‘{{id}}‘> id
2 <_sre.SRE_Match object; span=(9, 17), match=‘{{name}}‘> name
2 <_sre.SRE_Match object; span=(17, 24), match=‘{{age}}‘> age
1 <!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>wcl</title>
</head>
<body>
<h1>显示数据</h1>
<p>5tom20</p>
</body>
</html>
2、jinja
1)简介
基于模块的引擎。设计思想来自Django的模板引擎,与其及其相似。
文档:http://jinja.pocoo.org/docs/2.10/
http://docs.jinkan.org/docs/jinja2/
2)安装
安装:pip install jinjia2
pip install Markupsafe
3)模板构建
当前目录下,新建包webarch,下面闯进一个目录templates,该目录下新建一个HTML模板文件index.html
常用的模式
模板代码:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
显示数据<br>
<ul>
{% for id,name,age in userlist %}
<li>{{loop.index}}{{id}}
,{{name}},{{age}}</li>li>
{% endfor %}
</ul>
total{{usercount}}ren
</body>
</html>
Template文件代码:
from jinja2 import Environment
,PackageLoader,select_autoescape,FileSystemLoader
env = Environment(
loader=PackageLoader(‘webarch‘,‘templates‘),
autoescape=select_autoescape([‘html‘,‘xml‘])
)
# env1 = Environment(loader=FileSystemLoader(‘templates‘))
# d = {‘id‘:5,
# ‘name‘:‘tom‘,
# ‘age‘:20,
# ‘list‘:[]
# }
# d = {
# ‘userlist‘:[
# (1,‘tom‘,20),
# (3,‘jeery‘,23),
# (7,‘ll‘,28)
# ]
# }
def render(name,data:dict):
template = env.get_template(‘index.html‘)
html = template.render(**data)
return html
3、json
from .web import RouterAbles,Request,Response,App
from .template import
render
idx = RouterAbles(‘/‘)
py = RouterAbles(‘/python‘)
App.app_regsieter(idx,py)
# @idx.get(r‘^/$‘)
# @idx.router_register(r‘^/(?P<id>d+)$‘)
@idx.router_register(r‘^/{id:int}$‘)
def handler(request):
userlist = [
(3,‘tom‘,20),
(5,‘jerry‘,23),
(6,‘ll‘,25),
(7,‘ls‘,30)
]
d = {‘userlist‘:userlist,‘usercount‘:len(userlist)}
# res = Response(json=d) #json支持
# return res
return render(‘index.html‘,d)
# id = ‘‘
# if request.vars:
#
id = request.vars.id
# return
‘<h1></h1>{}‘.format(id)
# @py.get(r‘^/$‘)
# @py.get(‘^/(w+)$‘)
# @py.router_register(r‘^/(?P<id>d+)$‘)
@py.router_register(r‘^/{id}$‘)
def pythonhandler(request):
userlist = [
(3, ‘tom‘, 20),
(5, ‘jerry‘, 23),
(6, ‘ll‘, 25),
(7, ‘ls‘, 30)
]
d = {‘userlist‘:
userlist, ‘usercount‘: len(userlist)}
return render(‘index.html‘, d)
# res = Response(json=d) #json支持
# return res
# if request.vars:
#
id = request.vars.id
#
return ‘<h2></h2>{}‘.format(id)
4、模块化
将所有的代码组织成包和模块。
1.app.py
2.webarch(package)
(1)__init__.py
(2)template.py
(3)web.py
(4)templates(package)
Index.html
完整代码:
(1)app.py (server)
from wsgiref import simple_server
from webarch import
App
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
(2)__init__.py (handler函数)
from .web import RouterAbles,Request,Response,App
from .template import
render
idx = RouterAbles(‘/‘)
py = RouterAbles(‘/python‘)
App.app_regsieter(idx,py)
# @idx.get(r‘^/$‘)
# @idx.router_register(r‘^/(?P<id>d+)$‘)
@idx.router_register(r‘^/{id:int}$‘)
def handler(request):
userlist = [
(3,‘tom‘,20),
(5,‘jerry‘,23),
(6,‘ll‘,25),
(7,‘ls‘,30)
]
d = {‘userlist‘:userlist,‘usercount‘:len(userlist)}
return render(‘index.html‘,d)
# id = ‘‘
# if request.vars:
#
id = request.vars.id
# return
‘<h1></h1>{}‘.format(id)
# @py.get(r‘^/$‘)
# @py.get(‘^/(w+)$‘)
# @py.router_register(r‘^/(?P<id>d+)$‘)
@py.router_register(r‘^/{id}$‘)
def pythonhandler(request):
userlist = [
(3, ‘tom‘, 20),
(5, ‘jerry‘, 23),
(6, ‘ll‘, 25),
(7, ‘ls‘, 30)
]
d = {‘userlist‘:
userlist, ‘usercount‘: len(userlist)}
return render(‘index.html‘, d)
# if request.vars:
#
id = request.vars.id
#
return ‘<h2></h2>{}‘.format(id)
(3)template.py (渲染模块加载文件)
from jinja2 import
Environment ,PackageLoader,select_autoescape,FileSystemLoader
env = Environment(
loader=PackageLoader(‘webarch‘,‘templates‘),
autoescape=select_autoescape([‘html‘,‘xml‘])
)
# env1 = Environment(loader=FileSystemLoader(‘templates‘))
# d = {‘id‘:5,
# ‘name‘:‘tom‘,
# ‘age‘:20,
# ‘list‘:[]
# }
# d = {
# ‘userlist‘:[
# (1,‘tom‘,20),
# (3,‘jeery‘,23),
# (7,‘ll‘,28)
# ]
# }
def render(name,data:dict):
"""
:param name: 去模块目录搜索此模板名的文件
:param data: 字典
:return: 字符串
"""
template = env.get_template(‘index.html‘)
html = template.render(**data)
return html
(4)web.py (app,routerable,attrdict类)
from webob.dec import
wsgify
from webob.exc import
HTTPNotFound
from webob import
Response,Request
import logging
import re
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class 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 ‘<AttrDict{}>‘.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r‘/{([^{}:]+):?([^{}:]*)}‘
_regex = re.compile(pa)
TYPEPATTERNS = {
‘str‘: r‘[^/]+‘,
‘word‘: r‘w+‘,
‘int‘: r‘[+-]?d+‘,
‘float‘: r‘[+-]?d+.d+‘,
‘any‘: r‘.+‘
}
# def repl(self,matcher):
#
print(matcher,‘]]]‘)
#
name = matcher.group(1)
#
t = matcher.group(2)
#
s1 = ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t,
self.TYPEPATTERNS[‘str‘]))
#
print(‘----‘,s1)
#
return s1
#
# def pase(self,rule):
#
s2= self._regex.sub(self.repl,rule)
#
print(‘+-+-‘,rule)
#
print(‘=====‘,s2)
#
return s2
def parse(self,src: str):
start = 0
repl = ‘‘
types = {}
TYPECAST = {
‘str‘: str,
‘word‘: str,
‘int‘: int,
‘float‘: float,
‘any‘: str
}
matchers = self._regex.finditer(src)
for i, matcher in enumerate(matchers):
name = matcher.group(1)
t = matcher.group(2)
types[name] = TYPECAST.get(t, str)
repl +=
src[start:matcher.start()]
tmp = ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS[‘str‘]))
repl += tmp
start = matcher.end()
else:
repl += src[start:]
return repl, types
def __init__(self,prefix:str=‘‘):
self.__prefix
= prefix.rstrip(‘\/‘)
self.__ROUTERABLES
= []
def router_register(self,rule,*method):
def _router_register(handler):
pattern ,types = self.parse(rule)
self.__ROUTERABLES.append(
(tuple(map(lambda
x:x.upper(),method)),
re.compile(pattern),types,handler))
#print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,‘GET‘)
def post(self,pattern):
return self.router_register(pattern,‘POST‘)
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,pattern,trans,handler in self.__ROUTERABLES:
if not method
or request.method.upper() in method:
matchers =
pattern.match(request.path.replace(self.__prefix,‘‘,1))
#print(‘+++‘,trans)
if matchers:
newdict ={}
for k,v in matchers.groupdict().items():
newdict[k] =
trans[k](v)
#print(‘+-+-‘,trans[k](v))
#print(1,matchers.groupdict().items())
#print(k,v)
#print(‘---‘,newdict)
# request.groups =
matcher.groups()
#
print(2,request.groups)
request.vars = AttrDict(newdict)
# print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable
in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
for routerable
in self._ROUTER:
response =
routerable.match(request)
if response:
return response
raise HTTPNotFound()
(5)index.html文件内容,渲染模板
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
显示数据<br>
<ul>
{% for id,name,age in userlist %}
<li>{{loop.index}}{{id}}
,{{name}},{{age}}</li>li>
{% endfor %}
</ul>
total{{usercount}}ren
</body>
</html>
5、拦截器interceptor
拦截器是在请求处理的环节的某一处加入处理,有可能是中断后续的处理
分类:链式、洋葱式的。
拦截点不同:
1)请求时候拦截的
2)响应时候拦截
影响面:
1)全局拦截,在app中拦截
2)局部拦截,在router中拦截
拦截器可以是多个,多个拦截器是有顺序的,数据的response前执行的命名为preinterceptor,之后命名为postinterceptor。
加入拦截器功能的方式
(1)app和router类直接加入。
把拦截器相关方法,属性分别添加到相关的类中。
实现简单
(2)Mixin
App和router类都需要这个拦截器功能,这个两个类没有什么关系,可以使用Mixin方式,将属性、方法组合进来。App拦截器适合使用第二种,但是router的拦截器是每个实例不一样的,所以使用第一种方式实现。
拦截器函数的设计,fn不能影响数据向下一级的传递,也就是透明的。
Ip拦截的简单代码;
from webob.exc import Request
def ip(request:Request):
if request.remote_addr.startswish(‘192‘):
return request
else:
return None
6、发布的模式
from distutils.core import setup
import glob
setup(name=‘webarch‘, #
名字
version=‘0.0.1‘, #版本
description=‘python wsgi web framework‘, #描述信息
author=‘wcl‘, #作者
author_email=‘[email protected]‘,
#作者邮箱
url=‘www.‘,
#包的主页
packages=[‘webarch‘], #打包列表,
data_files=glob.glob(‘webarch/templates/*‘) #返回列表 配置文件,图片等文件列表
)
#打包
##python setup.py sdist
# 安装
#pip install webarch
7、总结
(1)熟悉wsgi的编程接口
(2)强化模块、类封装的思想
(3)增强业务分析的能力。
框架基本具备了wsgi web框架 的基本实现。
权限验证,SQL注入的检测功能都需要使用拦截器功能实现
完整代码:
(1)__init__.py文件
from .web import RouterAbles,Request,Response,App
from .template import
render
idx = RouterAbles(‘/‘)
py = RouterAbles(‘/python‘)
App.app_regsieter(idx,py)
def ip(request:Request):
if request.remote_addr.startswish(‘192‘):
return request
else:
return None
py.register_preinterceprot(ip)
# @idx.get(r‘^/$‘)
# @idx.router_register(r‘^/(?P<id>d+)$‘)
@idx.router_register(r‘^/{id:int}$‘)
def handler(request):
userlist = [
(3,‘tom‘,20),
(5,‘jerry‘,23),
(6,‘ll‘,25),
(7,‘ls‘,30)
]
d = {‘userlist‘:userlist,‘usercount‘:len(userlist)}
# res = Response(json=d) #json支持
# return res
return render(‘index.html‘,d)
# id = ‘‘
# if request.vars:
#
id = request.vars.id
# return ‘<h1></h1>{}‘.format(id)
# @py.get(r‘^/$‘)
# @py.get(‘^/(w+)$‘)
# @py.router_register(r‘^/(?P<id>d+)$‘)
@py.router_register(r‘^/{id}$‘)
def pythonhandler(request):
userlist = [
(3, ‘tom‘, 20),
(5, ‘jerry‘, 23),
(6, ‘ll‘, 25),
(7, ‘ls‘, 30)
]
d = {‘userlist‘:
userlist, ‘usercount‘: len(userlist)}
return render(‘index.html‘, d)
# res = Response(json=d) #json支持
# return res
# if request.vars:
#
id = request.vars.id
#
return ‘<h2></h2>{}‘.format(id)
(2)类文件web.py
from webob.dec import
wsgify
from webob.exc import
HTTPNotFound
from webob import
Response,Request
import logging
import re
FORMAT = ‘%(asctime)s
%(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)
class 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 ‘<AttrDict{}>‘.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r‘/{([^{}:]+):?([^{}:]*)}‘
_regex = re.compile(pa)
TYPEPATTERNS = {
‘str‘: r‘[^/]+‘,
‘word‘: r‘w+‘,
‘int‘: r‘[+-]?d+‘,
‘float‘: r‘[+-]?d+.d+‘,
‘any‘: r‘.+‘
}
# def repl(self,matcher):
#
print(matcher,‘]]]‘)
#
name = matcher.group(1)
#
t = matcher.group(2)
#
s1 = ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t,
self.TYPEPATTERNS[‘str‘]))
#
print(‘----‘,s1)
#
return s1
#
# def pase(self,rule):
#
s2=
self._regex.sub(self.repl,rule)
#
print(‘+-+-‘,rule)
#
print(‘=====‘,s2)
#
return s2
def parse(self,src: str):
start = 0
repl = ‘‘
types = {}
TYPECAST = {
‘str‘: str,
‘word‘: str,
‘int‘: int,
‘float‘: float,
‘any‘: str
}
matchers = self._regex.finditer(src)
for i, matcher in enumerate(matchers):
name = matcher.group(1)
t = matcher.group(2)
types[name] = TYPECAST.get(t, str)
repl +=
src[start:matcher.start()]
tmp = ‘/(?P<{}>{})‘.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS[‘str‘]))
repl += tmp
start = matcher.end()
else:
repl += src[start:]
return repl, types
def __init__(self,prefix:str=‘‘):
self.__prefix
= prefix.rstrip(‘\/‘)
self.__ROUTERABLES
= []
#拦截器
self.pre_interceptor
= []
self.post_interceptor
= []
def register_preinterceprot(self,fn):
self.pre_interceptor.append(fn)
return fn
def register_postinterport(self,fn):
self.post_interceptor.append(fn)
return fn
def router_register(self,rule,*method):
def _router_register(handler):
pattern ,types = self.parse(rule)
self.__ROUTERABLES.append(
(tuple(map(lambda
x:x.upper(),method)),
re.compile(pattern),types,handler))
#print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,‘GET‘)
def post(self,pattern):
return self.router_register(pattern,‘POST‘)
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
#请求拦截,处理request
for fn in self.pre_interceptor:
request = fn(request)
if not request:
return None
for method,pattern,trans,handler in self.__ROUTERABLES:
if not method
or request.method.upper() in method:
matchers =
pattern.match(request.path.replace(self.__prefix,‘‘,1))
#print(‘+++‘,trans)
if matchers:
newdict ={}
for k,v in matchers.groupdict().items():
newdict[k] =
trans[k](v)
#print(‘+-+-‘,trans[k](v))
#print(1,matchers.groupdict().items())
#print(k,v)
#print(‘---‘,newdict)
# request.groups = matcher.groups()
#
print(2,request.groups)
request.vars = AttrDict(newdict)
# print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
#全局拦截
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)
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable
in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
#全局请求拦截
for fn in self.PRE_INTERCEPTOR:
request = fn(request)
#全局拦截响应
for routerable
in self._ROUTER:
response =
routerable.match(request)
for fn in self.POST_INTERCEPTOR:
response = fn(request,response)
if response:
return response
raise HTTPNotFound()
(3)template.py文件
from jinja2 import
Environment ,PackageLoader,select_autoescape,FileSystemLoader
env = Environment(
loader=PackageLoader(‘webarch‘,‘templates‘),
autoescape=select_autoescape([‘html‘,‘xml‘])
)
# env1 =
Environment(loader=FileSystemLoader(‘templates‘))
# d = {‘id‘:5,
# ‘name‘:‘tom‘,
# ‘age‘:20,
# ‘list‘:[]
# }
# d = {
# ‘userlist‘:[
# (1,‘tom‘,20),
# (3,‘jeery‘,23),
# (7,‘ll‘,28)
# ]
# }
def render(name,data:dict):
"""
:param name: 去模块目录搜索此模板名的文件
:param data: 字典
:return: 字符串
"""
template = env.get_template(‘index.html‘)
html = template.render(**data)
return html
(4)app.py 文件
from wsgiref import
simple_server
from webarch import
App
if __name__ == ‘__main__‘:
server = simple_server.make_server(‘127.0.0.1‘,9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
(5)index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
显示数据<br>
<ul>
{% for id,name,age in userlist %}
<li>{{loop.index}}{{id}}
,{{name}},{{age}}</li>li>
{% endfor %}
</ul>
total{{usercount}}ren
</body>
</html>
以上是关于Python的web开发的主要内容,如果未能解决你的问题,请参考以下文章