Table of Contents
WSGIwsgi服务器作用wsgirefhttp协议无状态,短连接,长连接cookiesessionIDDjango配置项目环境Flask开发环境配置mac安装flask生成requirementsapp相关配置app.run配置实现flask项目配置文件的3种方式读取配置参数app.run本质Router@app.route和app.add_url_rule参数路由与视图函数url_for redirect路由简单使用转换器requestrequest本质response返回json格式数据返回构建信息返回普通字符串返回自定义错误信息返回信息中包含闪现cookieset get delcsrfsessionsession简单使用配置通过全局对象使用session通过字典使用session自定义session第三方sessiong对象中间件自定义中间件类实现请求钩子flask中的信号flask中请求中使用线程局部变量managerJinja2在视图函数中渲染模版xss模版语法模版过滤器表单扩展数据库定义表名的一般格式:连接并配置数据库flask_migrate语句blueprint蓝图的运行机制蓝图的作用循环引用的问题blueprint实现方式1. 在模块中实现2. 在包中实现多个app简单项目与复杂项目的目录结构目录结构及blueprint测试数据库测试部署RestfulREST的特点:API例子性能一、不同角度的网站性能普通用户认为的网站性能开发人员认为的网站性能运维人员认为的网站性能二、性能的指标并发数响应时间常见的系统操作响应时间三、性能的优化Web前端性能优化应用服务器优化总结:celery小知识点
WSGI
WSGI WEB服务器 ,本质上就是一个TCP服务器,监听在特定端口上
wsgi不是服务器???
(当客户端发送请求时,请求经过webserver,经过wsgiserver, 到达wsgiapp,)wsgi是存在于之间的通用网关接口,每收到一个请求,服务器会通过一个可调用对象,application_callable,调用应用,应用处理完数据,会先调用一次start_response, 作为响应头部,然后再返回一个可迭代对象作为body。
application_callable(environ, start_response), environ需要是一个键值对
start_response(status, headers, exec_info) status 可以是200 OK, 400 Not Found等,headers 是一个由 (header_name, header_value) 这样的元祖组成的列表,exec_info 是可选参数,一般在应用出现错误的时候会用到
wsgi服务器作用
监听http服务器端口(TCPServer, 默认端口80)
接收浏览器端http请求,并解析封装成environ环境数据
调用应用程序,将environ数据和start_response方法两个实参传入,给Application_callable函数
将应用程序返回的正文封装成HTTP响应报文,返回浏览器
wsgiref
只用来测试
# WSGI 可调用APP实现
res_str=b'www.baidu.com\n'
# 1 函数实现
defapplication(environ,start_response):
start_response("200 OK", [('Content-Type','text/plain; charset=utf-8')])
return[res_str]
# 2 类实现
classApplication:
def__init__(self,environ,start_response):
self.env=environ
self.start_response=start_response
def__iter__(self):# 对象可迭代
self.start_response('200 OK', [('Content-Type','text/plain; charset=utf-8')])
yieldfrom(res_str,'12234')
# 3 类实现,可调用对象
classApplication:
def__call__(self,environ,start_response):
start_response('200 OK', [('Content-Type','text/plain; charset=utf-8')])
return[res_str]
# yield from environ
# WSGI服务器__wsgiref
fromwsgiref.simple_serverimportmake_server,demo_app
ip='127.0.0.1'
port=9999
# 自己写app_callable
# server = make_server(ip, port, Application())
# 使用默认app_callable
server=make_server(ip,port,demo_app)
server.serve_forever()# server.handle_request() 执行一次
测试服务器,使用curl命令
# -I 使用HEAD方法
# -X 指定方法,-d传输数据
# curl -X POST http://127.0.0.1:9999/ -d {'x':2}
# curl - I http: // 127.0.0.1: 9999 /id = 5
environ: 返回的报文首部包含的信息
CONTENT_TYPE='text/plain'
HTTP_HOST='127.0.0.1:9999'# 客户端地址
HTTP_USER_AGENT='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36'
PATH_INFO='/'# 客户端根目录
QUERY_STRING=''# 查询全部内容
REMOTE_HOST=# 服务器端地址
REQUEST_METHOD='GET'# 请求方法
SERVER_NAME='1.0.0.127.in-addr.arpa'
REMOTE_ADDR='127.0.0.1'
SERVER_PORT='9999'
SERVER_PROTOCOL='HTTP/1.1'
SERVER_SOFTWARE='WSGIServer/0.2'
http协议
无状态,短连接,长连接
最早的http协议是无状态的,即服务器端不会记录浏览器端的任何状态,即使同一个浏览器发送2次请求,服务器端也不能判断出这两次请求,是不是同一个浏览器发出的,不过后面可以通过cookie和sessionID来区分;长连接是指,通信双方先建立3次握手,当数据传输完成,连接也不会断开,等待下一次的数据传输,这样时间长了服务器压力增大;短连接是指连接建立后,保持keep_alive状态,一段时间后再断开,减轻服务器压力。
cookie
是一种客户端,服务器端传递数据的一种技术,当浏览器首次给服务器发送请求时,服务器会在响应头部添加Set_Cookie字段,并生成cookie信息,当浏览器收到响应时,会记录这个cookie信息,当再次向服务器发送请求时会带上这个cookie信息,服务器就可以通过cookie信息,判断着两次请求是不是同一个浏览器发出的。
cookie的缺点:传输的是明文,安全性较差,每次通信时带上cookie信息,消耗流量,有大小限制,不超过4kB。
sessionID
浏览器如果第一次访问,服务器会为其生成一个sessionID,用来唯一识别浏览器,这个ID会通过Set_Cookie发送给浏览器,浏览器再次发送请求时会带上这个ID,这个sessionID是会话级别的,如果会话关闭,浏览器重新发送请求时,就不包含sessionID,服务器则会生成为其生成一个新的sessionID,同样服务器端也只会保存sessionID一段时间,超时后,会自动分配一个新的sessionID给浏览器,并覆盖浏览器会话级别的sessionID。
Django
配置项目环境
pycharm新建项目根目录选择虚拟环境
pipinstalldjango==2.2.8
django-adminstartprojectdata2d
./manage.pystartappdata2dapp
# 配置settings.py
INSTALLED_APPS加data2dapp
DATABASES= {
'default': {
'ENGINE':'django.db.backends.mysql',
'NAME':'blog',
'USER':'jw',
'PASSWORD':'7709',
'HOST':'172.16.177.160',
'PORT':'3306',
}
}
LANGUAGE_CODE='zh-hans'
TIME_ZONE='Asia/Shanghai'
DEBUG=True
LOGGING= {
'version':1,
'disable_existing_loggers':False,
'handlers': {
'console': {
'class':'logging.StreamHandler',
}, },
'loggers': {
'django.db.backends': {
'handlers': ['console'],
'level':'DEBUG',
},
},
}
# 安装mysql
pipinstallmysqlclient
# 新建数据库,和表
# 迁移
./manage.pymigrate
# 创建管理员
manage.pycreatesuperuser
# 启动web server
./manage.pyrunserver
# 后台登陆
127.0.0.1:8000/admin
配置前端环境
Flask
Flask 中的本地线程对象
Flask 内部使用本地线程对象,这样就不必在同一 个请求中考虑线程安全的问题,而是在函数之间传递对象。
开发环境配置
mac安装flask
$mkdirmyproject
$cdmyproject
$python3-mvenvvenv
# 进入虚拟环境
.venv/bin/activate
# 安装Flask
$pipinstallFlask
# 如果遇到socket.timeout: The read operation timed out
# 使用这条命令安装
# 查看已经安装了哪些库 pip freeze
$pip--default-timeout=100install
$pipinstallsimplejson
$pipinstallflask_cache
# 在服务器上启动redis
/home/jw/software/redis/src/redis-server
# 查看是否启动了
ps-ef|grepredis
生成requirements
piplist
pipfreeze
pipfreeze>requirements
pipinstall-rrequirements
app相关配置
app.run配置
# app.run(host='0.0.0.0', port=5000, debug=True)
# app.run(host='127.0.0.1', port=5000, debug=True)
# app.run(host='172.20.10.2', port=5000, debug=True)
实现flask项目配置文件的3种方式
配置参数的3种方式:
1使用配置文件,在运行文件的上级目录下(myproject),建立config.py
app.config.from_pyfile("config.py")
2使用对象配置参数(多)
classConfig(object):
DEBUG=True
app.config.from_object(Config)
3操作字典
app.config['DEBUG'] =True
读取配置参数
app.config本质上是一个字典
from flask import current_app
1 从全局中读取
print(app.config.get('ITCAST'))
2 如果获取不到app,从current_app中获取(与蓝图有关)
print(current_app.config.get('ITCAST'))
url_map
print(app.url_map)
# map例子,包括:路由,请求方法,视图函数名字
# common_view>
# ' (OPTIONS, GET, HEAD) -> static>
app.run本质
每一次请求来都会执行视图函数hello(函数加括号), hello是run_simple的第3个参数,在app.run()函数中同样封装了run_simple方法,只不过第3个参数是app,
所以app加括号,执行的是app实例对象的call方法
from werkzeug.wrappers import Request, Response
@Request.application
def hello(request):
return Response('Hello World!')
if __name__ == '__main__':
from werkzeug.serving import run_simple
run_simple('localhost', 4000, hello)
Router
@app.route和app.add_url_rule参数
@app.route和app.add_url_rule参数:
# 常用
rule, URL规则
view_func, 视图函数名称
defaults=None, 默认值,当URL中无参数,函数需要参数时,使用defaults={'k':'v'}为函数提供参数
endpoint=None, 名称,用于反向生成URL,即: url_for('名称')
methods=None, 允许的请求方式,如:["GET","POST"]
# 不常用
strict_slashes=None, 对URL最后的 / 符号是否严格要求,
如: @app.route('/index',strict_slashes=False), 访问 http://www.xx.com/index/ 或 http://www.xx.com/index均可访问 @app.route('/index',strict_slashes=True)
仅访问 http://www.xx.com/index
redirect_to=None,重定向到指定地址
如: @app.route('/index/', redirect_to='/home/')
def func(adapter, nid):
return "/home/888"
或 @app.route('/index/', redirect_to=func)
subdomain=None,子域名访问
from flask import Flask, views, url_for
app = Flask(import_name=__name__)
# 需要先在/etc/hosts 文件中配置DNS映射
# 如果这么配置正常的路由将访问不了
app.config['SERVER_NAME'] = 'test'
@app.route("/", subdomain="admin")
def static_index():
"""Flask supports static subdomains
This is available at static.your-domain.tld"""
return "static.your-domain.tld"
@app.route("/dynamic", subdomain="")
def username_index(username):
"""Dynamic subdomains are also supported
Try going to user1.your-domain.tld/dynamic"""
return username + ".your-domain.tld"
if __name__ == '__main__':
app.run()
路由与视图函数
url_for redirect
# 视图函数名字可以是其他视图函数名字,也可以是当前视图函数名字
# 也可以是endpoint别名
url = url_for('视图函数名字')
# 后面可以加视图函数参数
url = url_for('视图函数名字', 99)
redirect也可以这么用
redirect('/login')
# 重定向的另一种方法(多)
@app.route('/')
def index():
return 'index page'
@app.route('/login')
def login():
url = url_for('index') # 视图函数的名字
print('url:', url, type(url))
# url='/', 即index函数对应的路径, 字符串
return redirect(url)
# 重定向的另一种方法
@app.route('/login')
def login():
url = '/'
return redirect(url)
路由简单使用
# 视图函数相同,路由不同,结果相同
@app.route('/common_view1')
@app.route('/common_view')
def common_view():
return 'common_view'
# 路由相同,请求方法相同,视图函数不同,会返回先找到的视图函数
# 返回'diff_view1'
@app.route('/diff_view')
def diff_view1():
return 'diff_view1'
@app.route('/diff_view')
def diff_view2():
return 'diff_view2'
# 路由相同,请求方法不同,视图函数不同,会返回接收get方法的视图函数
# 返回'diff_view2'
@app.route('/diff_view11', methods=['POST'])
def diff_view11():
return 'diff_view1'
@app.route('/diff_view11', methods=['GET'])
def diff_view22():
return 'diff_view2'
转换器
flask支持的默认路由转换器:
@app.route('/user/')
@app.route('/post/')
@app.route('/post/')
@app.route('/post/')
@app.route('/login', methods=['GET', 'POST'])
常用路由系统有以上五种,所有的路由系统都是基于一下对应关系来处理:
DEFAULT_CONVERTERS = {
'default': UnicodeConverter,
'string': UnicodeConverter,
'any': AnyConverter,
'path': PathConverter,
'int': IntegerConverter,
'float': FloatConverter,
'uuid': UUIDConverter,
# 注意:
# 1、goods_id与视图函数中的参数必须同名
# 2 冒号后面不能有空格
# 提取字符串
# 不加转换器,默认匹配除了/以外的,所有字符串,而且这个字符串必须存在
@app.route('/goods/')
def goods_detail(goods_id):
return 'goods detail page {}'.format(goods_id)
# 提取整数
@app.route('/goods/')
def goods_detail(goods_id):
return 'goods detail page {}'.format(goods_id)
自定义转换器
同样一个类,切换不同正则表达式,就可以匹配不同数据
re(.*) 可以匹配/index.html /
# 自定义转换器
class RegexConverter(BaseConverter):
def __init__(self, url_map, regex):
super().__init__(url_map)
# 名字可以不是self.regex,
# flask会利用这个属性进行路由的正则匹配
self.regex =regex
app.url_map.converters['re'] = RegexConverter
# app.url_map.converters是字典
# 自定义的转换器,必须加入到flask应用的字典中去
# 例子:'re':
@app.route("/send/")
def send_msm(mobile):
return 'send msm to %s' % mobile
# /call路径与/send路径不同,但是可以使用同一个RegexConverter类,
# 替换不同的正则表达式,就可以匹配多种类型号码
@app.route("/call/")
def call_tel(mobile):
return 'call to {}'.format(mobile)
自定义转换器中的数据流
def to_python(self, value):
pass
# 视图函数正常执行都会执行这一句,对转化器后面的参数转换,比如字符串转换成整数
def to_url(self, value):
pass
# 视图函数只要有url_for(),都会执行
#例子1
# 自定义转换器函数执行顺序:
# 如果原视图函数中有,redirect函数,例如:index函数
# 1 原视图函数
# 2 to_url
# 3 to_python
# 4 redirect指向的视图函数
# 如果原视图函数中没有,redirect函数,例如:test_to_python函数
# 3 to_python
# 1 原视图函数
class RegexConverter(BaseConverter):
def __init__(self, url_map, regex):
super().__init__(url_map)
self.regex =regex
# to_url的返回值会传递给to_python
def to_url(self, value):
print('==============to_url', value)
return '1'
# 在视图函数之前执行,拦截视图函数
def to_python(self, value):
print('===========to_python')
return int(value)
app.url_map.converters['re'] = RegexConverter
@app.route('/')
def index():
print('========index')
url = url_for('test_to_python', mobile='')
return redirect(url)
@app.route("/test_to_python/")
def test_to_python(mobile):
print('========test_to_python')
return 'send msm to %s' % mobile
# 路由参数提取及转换器, 例子2
class RegexConverter(BaseConverter):
def __init__(self, url_map, regex):
super().__init__(url_map)
self.regex =regex
def to_python(self, value):
print('===========to_python')
return value
def to_url(self, value):
print('==============to_url', value)
return '9'
# return url_quote(value, charset=self.map.charset)
app.url_map.converters['re'] = RegexConverter
# re里面写33 44等,会识别成单个的3,或者4
@app.route("/test_to_python/")
def test_to_python(mobile):
print('========test_to_python')
return 'send msm to %s' % mobile
@app.route("/test_to_url/")
def test_to_url(mobile):
print('========test_to_url')
url = url_for('test_to_python', mobile=mobile)
url = url_for('test_to_url', mobile=mobile)
return redirect(url)
request
请求响应信息汇总:
# 如果忘记可以通过类名字来查询
# Request.
# Response.
# 请求相关信息
# request.method
# request.args
# request.form
# request.values
# request.cookies
# request.headers
# request.path
# request.full_path
# request.script_root
# request.url
# request.base_url
# request.url_root
# request.host_url
# request.host
# request.files
# obj = request.files['the_file_name']#obj.save('/var/www/uploads/'+secure_filename(obj.filename))
# 响应相关信息
# return "字符串"
# return render_template('html模板路径',**{})
# return redirect('/index.html')
# return jsonify({'a': 1})
# response =make_response(render_template('index.html'))
# response是flask.wrappers.Response类型
# response.delete_cookie('key')
# response.set_cookie('key', 'value')
# response.headers['X-Something'] = 'A value'
# return response
request 包括前端发送过来的所有请求数据
@app.route('/index', methods=['GET', 'POST'])
def index():
# data bytes
# 前端以raw(也就是json)方式发送过来的数据,都存在request.data当中,
# 是bytes类型,例如:b'{"username": "jw", "pwd": "123"}'
print(request.data)
return 'login success'
# form ImmutableMultiDict([])
# 前端表单发送过来的数据,以类字典的方式,都存在request.form中
print(request.form)
# request.form.get
# 返回字典中的一个数据
print(request.form.get('username', 'jj'))
print(request.form.get('pwd'))
# request.form.getlist []
# 返回包含username的列表,如果username重复,可以返回多个用户名
print(request.form.getlist('username'))
# request.args ImmutableMultiDict([])
# 只用来提取查询字符串中的参数,即url中的参数
# ImmutableMultiDict([('country', "'zhong'")])
print(request.args)
# postman例子
# GET /api/v1.0/houses?aid=6&sd=2020-2-13&ed=2020-2-14&sk=new&p=1
# 路由例子
@api1.route('/houses')
# request.files ImmutableMultiDict([])
# 返回二进制文件对象
file_obj = request.files.get('pic').read()
print(type(file_obj))
with open('test_pic.png', 'wb') as f:
f.write(file_obj.read())
print('success write')
request other method
# request.url
# request.cookies
# request.headers
# request.method
request本质
处理多个request的方式
如果只有一个请求,单进程单线程,基于全局标量处理
处理多个请求,单进程多线程,基于threadinglocal实现
处理多个请求,单进程单线程多协程,即多每个线程分片处理,每一片处理一个请求,因为本质还是多线程,所以threadinglocal解决不了多个协程中,变量共用的问题
例子:在flask内部有一个local类,实现了给每一个线程开辟不同的空间,
当处理不同的request时,每一个请求都有一个单独的空间,内部实现原理与下例类似
# 手动实现类似threading.local()
importthreading
fromwerkzeugimportlocal
try:
fromgreenletimportgetcurrentasget_ident# 协程
exceptImportError:
try:
fromthreadimportget_ident
exceptImportError:
from_threadimportget_ident# 多线程
classLocal:
def__init__(self):
object.__setattr__(self,'__storage__', {})
object.__setattr__(self,'__get_ident__',get_ident)
def__setattr__(self,key,value):
ident=self.__get_ident__()
storage=self.__storage__
try:
storage[ident][key] =value
exceptKeyError:
storage[ident] = {key:value}
def__getattr__(self,item):
try:
returnself.__storage__[self.__get_ident__()].get(item)
exceptKeyError:
raiseAttributeError(item)
local=Local()
deftest_local():
local.x=0
for_inrange(10):
local.x+=1
print(threading.current_thread(),local.x)
print(local.__storage__)
foriinrange(5):
ret=threading.Thread(target=test_local)
ret.start()
response
返回json格式数据
# response tuple.
# The tuple must have the form
# (body, status, headers), (body, status), or (body, headers).
return simplejson.dumps({"name": 'jw'}), 200, {"Content-Type": "application/json"}
# 会自动将Content-Type设置为application/json
return jsonify({"name": 'jw'})
返回构建信息
# make_response构建响应信息
resp = make_response(simplejson.dumps({"name": 'jw'}))
resp.status = "201"
resp.headers['Content-Type'] = 'application/json'
csrf_token = csrf.generate_csrf()
resp.set_cookies = {'csrf_token': csrf_token}
print(type(resp))
return resp
返回普通字符串
return '200' # text/html; charset=utf-8
返回自定义错误信息
@app.route('/index', methods=['GET', 'POST'])
def index():
# 当请求错误时,abort可以立即终止视图函数的执行
# 并返回给前端特定的信息
# 1. 传递状态码,必须是http标准状态码
# 结合下面的视图函数,重新定义返回的错误信息
abort(404)
# # 2. 传递响应体信息, 直接给前端返回,'login failed'
# resp = Response('login failed')
# abort(resp)
@app.errorhandler(404)
def handle_404_error(error):
return '出现了错误信息404,错误信息为: {}'.format(error)
返回信息中包含闪现
flash是一个基于Session实现的用于保存数据的集合,其特点是:使用一次就删除。
from flask import flash, Flask, get_flashed_messages
app = Flask(__name__)
app.secret_key = '00ikkjd'
# 获得一次就没有了
@app.route('/')
def index():
flash('flash1')
return 'test flash'
@app.route("/get_flash")
def get_flash():
msg = get_flashed_messages()
return ','.join(msg)
app.run(debug=True)
from flask import flash, Flask, get_flashed_messages
app = Flask(__name__)
app.secret_key = '00ikkjd'
@app.route('/')
def index():
flag = True
flash('hello3')
if flag:
# 闪现信息
print('first:', get_flashed_messages()) # 3个flash信息
flash('hello1')
print('second:', get_flashed_messages()) # 3个flash信息
flash('hello2')
print('thrid:', get_flashed_messages()) # 3个flash信息
flag = False
return 'test flash'
@app.route("/get_flash")
def get_flash():
msg = get_flashed_messages() # 只得到了2个flash,没有hello3
return ','.join(msg)
app.run(debug=True)
from flask import flash
flag=True
@app.route('/index')
def index():
global flag
if flag:
# 闪现信息
flash('hello1')
flash('hello2')
flag = False
return 'test flash'
cookie
set get del
@app.route('/index', methods=['GET', 'POST'])
def index():
resp = make_response('set cookie')
## 默认浏览器关闭就失效,是临时cookie
# resp.set_cookie("test1", "python2")
#
# # max_age 设置有效期时间为3600秒
# resp.set_cookie("test2", "python2", max_age=3600)
# 在头部信息中set cookie
resp.headers['Set-Cookie'] = "test3=python3; Max-Age=3600; Expires=None"
return resp
@app.route('/get-cookie', methods=['Get', 'Post'])
def get_cookie():
return simplejson.dumps(request.cookies)
@app.route('/del-cookie', methods=['Get', 'Post'])
def del_cookie():
print(request.cookies, type(request.cookies))
resp = make_response('del cookie')
resp.delete_cookie("test1")
resp.delete_cookie("csrftoken")
return resp
csrf
CSRF或XSRF(Cross-site Request Forgery),即跨站请求伪造。它也被称为:one click attack/session riding,是一种对网站的恶意利用。
csrf攻击:当用户登陆了A网站,A网站的cookie信息保存在浏览器(可能是浏览器没有关闭,或者cookie持久化了) ,在这期间又访问了黑客网站,
黑客网站有一些隐藏的运行代码,这些代码一旦执行就向A网站悄悄地发起了请求,由于A网站的cookie信息还有效,而且访问的是A网站,则cookie就一并发给A网站,A网站看到这些cookie就认为是登陆用户发起的合理请求,这就是csrf攻击。
解决办法:在请求头和请求体中分别增加csrf_token字段,浏览器每次访问A网站,都会在请求头和请求体中加上csrf_token, A会对比这两个值,如果相同,说明是安全的请求,而黑客网站B,它只能发起请求,让浏览器自动带上cookie值,不能获得A网站的cookie值,当黑客网站再一次发起对A网站请求时,A网站检查,请求体中没有csrf_token,或者这个值对不上,则验证失败。
SECRET_KEY = "XHS999*Y9dfs9cshd9"
SESSION_TYPE = 'filesystem'
csrf_token = csrf.generate_csrf()
resp.set_cookie('csrf_token', csrf_token)
session
flask的session,可以在不同请求之间存储特定用户信息,默认保存在cookies中,是在cookie基础之上实现的,并且对cookies进行密钥签名,所以使用session之前需要设置一个密钥(不安全)
session简单使用
from flask import Flask, request, render_template, Request,session
app = Flask(__name__)
app.secret_key = "ippppc mdmmem"
@app.route('/')
def index():
session['name'] = 'jjj'
return 'index'
@app.route('/logout')
def logout():
session.pop('jjj')
return 'Already logout'
if __name__ == "__main__":
app.run(debug=True)
配置
app.config['SECRET_KEY'] = '77890jnnijndnwixnniijni'
通过全局对象使用session
@app.route('/login', methods=['GET', 'POST'])
def login():
session['1'] = 123
session['name'] = 'jw'
session['tel'] = '123'
return 'success login'
@app.route('/index', methods=['GET', 'POST'])
def index():
return 'login: {} {} {}'.format(session.get('name'), session.get('tel'),
session.get('1'))
通过字典使用session
session_dict = {'1': 111,
'2': 222,
'name': ''}
自定义session
# pip3 install Flask-Session
# run.py
fromflaskimportFlask
fromflaskimportsession
frompro_flask.utils.sessionimportMySessionInterface
app=Flask(__name__)
app.secret_key='A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
app.session_interface=MySessionInterface()
@app.route('/login.html',methods=['GET',"POST"])
deflogin():
print(session)
session['user1'] ='alex'
session['user2'] ='alex'
delsession['user2']
return"内容"
if__name__=='__main__':
app.run()
# session.py
importuuid
importjson
fromflask.sessionsimportSessionInterface
fromflask.sessionsimportSessionMixin
fromitsdangerousimportSigner,BadSignature,want_bytes
classMySession(dict,SessionMixin):
def__init__(self,initial=None,sid=None):
self.sid=sid
self.initial=initial
super(MySession,self).__init__(initialor())
def__setitem__(self,key,value):
super(MySession,self).__setitem__(key,value)
def__getitem__(self,item):
returnsuper(MySession,self).__getitem__(item)
def__delitem__(self,key):
super(MySession,self).__delitem__(key)
classMySessionInterface(SessionInterface):
session_class=MySession
container= {}
def__init__(self):
importredis
self.redis=redis.Redis()
def_generate_sid(self):
returnstr(uuid.uuid4())
def_get_signer(self,app):
ifnotapp.secret_key:
returnNone
returnSigner(app.secret_key,salt='flask-session',
key_derivation='hmac')
defopen_session(self,app,request):
"""
程序刚启动时执行,需要返回一个session对象
"""
sid=request.cookies.get(app.session_cookie_name)
ifnotsid:
sid=self._generate_sid()
returnself.session_class(sid=sid)
signer=self._get_signer(app)
try:
sid_as_bytes=signer.unsign(sid)
sid=sid_as_bytes.decode()
exceptBadSignature:
sid=self._generate_sid()
returnself.session_class(sid=sid)
# session保存在redis中
# val = self.redis.get(sid)
# session保存在内存中
val=self.container.get(sid)
ifvalisnotNone:
try:
data=json.loads(val)
returnself.session_class(data,sid=sid)
except:
returnself.session_class(sid=sid)
returnself.session_class(sid=sid)
defsave_session(self,app,session,response):
"""
程序结束前执行,可以保存session中所有的值
如:
保存到resit
写入到用户cookie
"""
domain=self.get_cookie_domain(app)
path=self.get_cookie_path(app)
httponly=self.get_cookie_httponly(app)
secure=self.get_cookie_secure(app)
expires=self.get_expiration_time(app,session)
val=json.dumps(dict(session))
# session保存在redis中
# self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime)
# session保存在内存中
self.container.setdefault(session.sid,val)
session_id=self._get_signer(app).sign(want_bytes(session.sid))
response.set_cookie(app.session_cookie_name,session_id,
expires=expires,httponly=httponly,
domain=domain,path=path,secure=secure)
第三方session
"""
pip3 install redis
pip3 install flask-session
"""
from flask import Flask, session, redirect
from flask.ext.session import Session
app = Flask(__name__)
app.debug = True
app.secret_key = 'asdfasdfasd'
app.config['SESSION_TYPE'] = 'redis'
from redis import Redis
app.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379')
Session(app)
@app.route('/login')
def login():
session['username'] = 'alex'
return redirect('/index')
@app.route('/index')
def index():
name = session['username']
return name
if __name__ == '__main__':
app.run()
g对象
是一个请求周期内的全局变量。处理请求时,用于临时存储的变量,每一次请求都会重设这个变量本质是LocalProxy类型,是从app_ctx对象中取出的g对象
print(g) # <flask.g of 'hello'>
中间件
自定义中间件类实现
from flask import flash, Flask, get_flashed_messages
app = Flask(__name__)
# 请求处理之前操作
# 执行视图函数
# 请求处理之后的操作
@app.route('/')
def index():
print('执行视图函数')
return 'index page'
# 每一次请求之前都会调用call方法
# app.__call__
class MiddleWare:
def __init__(self, old_wsgi_app):
self.wsgi_app = old_wsgi_app
def __call__(self, *args, **kwargs):
print('请求处理之前操作')
self.wsgi_app(*args, **kwargs)
print('请求处理之后的操作')
app.wsgi_app = MiddleWare(app.wsgi_app)
app.run(debug=True)
请求钩子
以登陆装饰器为例,如果视图函数太多,就需要多次使用装饰器,但是如果在每一次请求之前判断用户是否登陆,则很方便:
from flask import Flask, session, redirect, url_for,Request, request
app = Flask(__name__)
app.secret_key = '999nii'
@app.route('/')
def index():
return 'index page'
@app.route('/login')
def login():
session['name'] = 'jjj'
return 'already login'
@app.before_request
def process_login():
# request:Request
if request.path == '/login': # 一定要写不然递归
return
if session.get('name') == 'jjj':
return
return redirect('/login')
app.run(debug=True)
当有多个钩子时定义的顺序和相应的顺序
# 请求之前第一个钩子
# 请求之前第二个钩子
# 响应第二个钩子
# 响应第一个钩子
fromflaskimportFlask,session,redirect,url_for,Request,request
app=Flask(__name__)
app.secret_key='999nii'
@app.route('/')
defindex():
return'index page'
@app.route('/login')
deflogin():
session['name'] ='jjj'
return'already login'
@app.before_request
defprocess_login1():
print('请求之前第一个钩子')
ifrequest.path=='/login':
return
ifsession.get('name') =='jjj':
return
returnredirect('/login')
@app.before_request
defprocess_login2():
print('请求之前第二个钩子')
ifrequest.path=='/login':
return
ifsession.get('name') =='jjj':
return
returnredirect('/login')
@app.after_request
defprocess_logout1(response):
print('响应第一个钩子')
returnresponse
@app.after_request
defprocess_logout2(response):
print('响应第二个钩子')
returnresponse
app.run(debug=True)
flask支持4种请求钩子
@app.before_first_request
defhandle_before_first_request():
"""第一次请求处理之前被执行"""
print('第一次请求处理之前被执行')
@app.before_request
defhandle_before_request():
"""每一次请求之前都被执行,包括第一次"""
print('每一次请求之前都被执行,包括第一次')
@app.after_request
defhandle_after_request(response):
"""每一次请求之后(视图函数执行完成,并且没有异常)都被执行"""
print('每一次请求之后, 都被执行, 视图函数有异常不执行')
returnresponse
@app.teardown_request
defhandle_teardown_request(response):
"""每一次请求之后(视图函数执行完成,有无异常)都被执行"""
print(response)# None
# 返回值不影响视图函数返回值,但是参数response必须要有
path=request.path
ifpathin[url_for('index'),url_for('login')]:
print('调用了index视图函数')
return
elifpath==url_for('hello'):
print('调用了hello视图函数')
else:
print('每一次请求之后, 都被执行, 视图函数有异常,也执行')
returnresponse
flask中的信号
使用信号例子:
如果数据库写入数据时需要记录一下,可以通过发送信号给日志系统,记录日志
from flask import Flask, signals
app = Flask(__name__)
def func(sender, **kwargs):
return 'sender is {} , receive_content is {}'.format(sender, kwargs)
# 在信号中注册函数
signals.request_started.connect(func)
# 信号什么时候被触发呢?当调用send方法时
@app.route('/')
def index():
ret = signals.request_started.send('index_view', a='123')
print(ret)
print('index')
return 'page index'
if __name__ == "__main__":
app.run(debug=True)
flask内置的信号
1. 内置信号
request_started = _signals.signal('request-started') # 请求到来前执行
request_finished = _signals.signal('request-finished') # 请求结束后执行
before_render_template = _signals.signal('before-render-template') # 模板渲染前执行
template_rendered = _signals.signal('template-rendered') # 模板渲染后执行
got_request_exception = _signals.signal('got-request-exception') # 请求执行出现异常时执行
request_tearing_down = _signals.signal('request-tearing-down') # 请求执行完毕后自动执行(无论成功与否)
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行(无论成功与否)
appcontext_pushed = _signals.signal('appcontext-pushed') # 应用上下文push时执行
appcontext_popped = _signals.signal('appcontext-popped') # 应用上下文pop时执行
message_flashed = _signals.signal('message-flashed')
# request_started request_finished
class Flask(_PackageBoundObject):
def full_dispatch_request(self):
self.try_trigger_before_first_request_functions()
try:
# 触发request_started 信号
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
response = self.make_response(rv)
response = self.process_response(response)
# request_finished 信号
request_finished.send(self, response=response)
return response
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
ctx.push()
error = None
try:
try:
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.make_response(self.handle_exception(e))
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
# before_render_template template_rendered
def render_template(template_name_or_list, **context):
"""Renders a template from the template folder with the given context.
:param template_name_or_list: the name of the template to be
rendered, or an iterable with template names
the first one existing will be rendered
:param context: the variables that should be available in the context of the template.
"""
ctx = _app_ctx_stack.top
ctx.app.update_template_context(context)
return _render(ctx.app.jinja_env.get_or_select_template(template_name_or_list),context, ctx.app)
def _render(template, context, app):
"""Renders the template and fires the signal"""
# ############### before_render_template 信号 ###############
before_render_template.send(app, template=template, context=context)
rv = template.render(context)
# ############### template_rendered 信号 ###############
template_rendered.send(app, template=template, context=context)
return rv
# got_request_exception
class Flask(_PackageBoundObject):
def handle_exception(self, e):
exc_type, exc_value, tb = sys.exc_info()
# ############### got_request_exception 信号 ###############
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
if handler is None:
return InternalServerError()
return handler(e)
# message_flashed
def flash(message, category='message'):
flashes = session.get('_flashes', [])
flashes.append((category, message))
session['_flashes'] = flashes
# ############### 触发 message_flashed 信号 ###############
message_flashed.send(current_app._get_current_object(),
message=message, category=category)
自定义信号
from flask import Flask
from flask.signals import _signals
app = Flask(__name__)
# 自定义信号
test = _signals.signal('test_signal')
def func():
print('function test signal')
# 在信号中注册函数
test.connect(func)
@app.route('/')
def index():
test.send('123', k='3999')
return 'page index'
if __name__ == "__main__":
app.run()
flask中请求中使用线程局部变量
对于全局变量request,当调用request.form.get('name')时,可以根据不同的线程区分:
request = {
'thread1': {
'form': {'name': 'jw'},
'args': {'id': 1122},
},
'thread2': {
'form': {'name': 'jw'},
'args': {'id': 1122},
}
}
manager
from flask_script import Manager
manager = Manager(app)
manager.run()
$ python hello.py runserver
Jinja2
在视图函数中渲染模版
@app.route('/index')
def index():
data = {'name': 'w',
's': 123,
'n': {'2020': 2020},
'list1': [1, 2, 3, 4, 5]
}
return render_template('index.html', **data)
# return render_template('index.html', name='jw')
# html
name = {{name}}
n = {{n['2020']}}
xss
@app.route('/xss', methods=['POST', 'GET'])
def xss():
text = "text789"
if request.method == 'POST':
text = request.form.get("text", '99999')
return render_template('index.html', text=text)
模版语法
模版过滤器
# 自定义过滤器
# def list_step_2(li):
# return li[::2]
# app.add_template_filter(list_step_2, 'li_filter')
@app.template_filter('li_filter')
def list_step_2(li):
return li[1::2]
# html
list1 = {{list1 | li_filter}}
str1 = {{" test flask " |trim |upper}}
@app.template_global()
def sb(a1, a2):
return a1 + a2
@app.template_filter()
def db(a1, a2, a3):
return a1 + a2 + a3
在模版中调用方式:{{sb(1,2)}} {{ 1|db(2,3)}} # 给函数db先固定一个参数
模版传函数例子1:
Title
{{func('functest') | safe }}
# test.py
from flask import Flask, request, render_template
app = Flask(__name__)
def func(x):
return "".format(x)
@app.route('/')
def index():
return render_template('test.html', func=func )
if __name__ == "__main__":
app.run(debug=True)
表单扩展
# Flask-WTF,可以帮助进行CSRF验证,快速定义表单模版
数据库
定义表名的一般格式:
# 数据库名字缩写_表名
# tbl_表名
连接并配置数据库
建表并生成数据
fromflaskimportFlask,escape,url_for,request,render_template
fromflaskimportResponse,Request,jsonify,render_template
fromflaskimportmake_response,abort,redirect,\
session,current_app,g
fromwerkzeug.utilsimportsecure_filename
fromwerkzeug.contrib.fixersimportLighttpdCGIRootFix
fromwerkzeug.routingimportBaseConverter
importwerkzeug
importsimplejson
fromflask_scriptimportManager
fromflask_sqlalchemyimportSQLAlchemy
fromflask_migrateimportMigrate,MigrateCommand
app=Flask(__name__)
classConfig(object):
SQLALCHEMY_TRACK_MODIFICATIONS=True
SQLALCHEMY_DATABASE_URI="mysql://jw:7709@172.16.177.160:3306/Flask_test"
app.config.from_object(Config)
db=SQLAlchemy(app)
migrate=Migrate(app,db)
manager=Manager(app)
#manager是Flask-Script的实例,在flask-Script中添加一个db命令
manager.add_command('db',MigrateCommand)
classRole(db.Model):
"""角色表"""
__tablename__="tbl_roles"
id=db.Column(db.Integer,primary_key=True)
name=db.Column(db.String(32),unique=True)
users=db.relationship('User',backref='role')
def__repr__(self):
return'Role:'.format(self.name)
classUser(db.Model):
"""用户表"""
__tablename__="tbl_users"
id=db.Column(db.Integer,primary_key=True)# 默认自增
name=db.Column(db.String(32),unique=True)
email=db.Column(db.String(128),unique=True)
password=db.Column(db.String(128))
role_id=db.Column(db.Integer,db.ForeignKey('tbl_roles.id'))
def__repr__(self):
return'User:'.format(self.username)
if__name__=='__main__':
# db.drop_all()
# manager.run()
print('*'*40)
# db.create_all()
#
# role1 = Role(name='admin')
# db.session.add(role1)
# db.session.commit()
#
# role2 = Role(name='stuff')
# db.session.add(role2)
# db.session.commit()
#
# us1 = User(name='wang', email='wang@qq.com', password='123', role_id=role1.id)
# us2 = User(name='zhang', email='zhang@qq.com', password='123', role_id=role1.id)
# us3 = User(name='chen', email='chen@qq.com', password='123', role_id=role2.id)
# us4 = User(name='liu', email='liu@qq.com', password='123', role_id=role2.id)
#
# db.session.add_all([us1, us2, us3, us4])
# db.session.commit()
flask_migrate
flask_script
fromflask_scriptimportManager
manage=Manager(app)
if__name__=='__main__':
manage.run()
flask_migrate
$ sudo pip install flask-migrate
from flask_migrate import Migrate, MigrateCommand
Migrate(app, db)
manage.add_command('db', MigrateCommand)
# 查看帮助
python manage.py db --help
#创建migrations文件夹,所有迁移文件都放在里面。
python manage.py db init
#创建自动迁移脚本, 在migrations文件夹下面的version目录中,生成版本号
# -m后面为版本号的备注信息
# 相当于提交到本地仓库
python manage.py db migrate -m 'initial migration'
# 在提交到本地仓库后,才可以进行upgrade, downgrade
# 在python中,增删改表后,在数据库中也做对应的更改
# 相当于提交到远程数据库
python manage.py upgrade
# 查看提交的版本号
python manage.py db history
# 恢复到指定的版本号
python manage.py db downgrade 版本号
# 非首次更新数据库,只是修改了某个表
python manage.py db migrate -m "modify table name"
python manage.py db upgrade
# 如果恢复表名,则不用再次migrate,直接执行downgrade, 就可以修改
python manage.py db downgrade 版本号
遇到报错Target database is not up to date,表示目标数据库与
项目中的版本不匹配,需要更新数据中的版本
python manage.py db heads #项目中的版本
python manage.py db current # 数据库中的版本
# 更新数据库中的版本到heads
python manage.py db stamp heads
语句
参考数据库md
blueprint
蓝图:用于实现单个应用的视图、模板、静态文件的集合。
蓝图就是模块化处理的类。主要用来实现客户端请求和URL映射
蓝图的运行机制
蓝图是保存了一组将来可以在应用对象上执行的操作。注册路由就是一种操作,当在程序实例上调用route装饰器注册路由时,这个操作将修改对象的url_map路由映射列表。当我们在蓝图对象上调用route装饰器注册路由时,它只是在内部的一个延迟操作,在记录列表defered_functions中添加了一个项。当执行应用对象的 register_blueprint() 方法时,应用对象从蓝图对象的 defered_functions 列表中取出每一项,即调用应用对象的 add_url_rule() 方法,这将会修改程序实例的路由映射列表。
蓝图的作用
蓝图是存储视图,模板,静态文件及操作方法的容器,当蓝图注册到app上之后,app就可以通过蓝图,统一管理这些操作方法,有相同功能的视图函数可以放在同一个模块中,每一个py文件都可以创建蓝图,蓝图可以为这些视图函数创建:
统一的url_prefix,
属于自己的静态文件或模版路径,
请求扩展,如@blue_ap1.before_request
蓝图也可以实现多个app管理
在项目根目录下创建多个app,每个app都拥有自己的蓝图
在一个blueprint中可以调用另一个blueprint的视图函数, 但要加相应的blueprint名.
flask.Blueprint()类并用参数name和import_name初始化。import_name通常用__name__
循环引用的问题
当运行main时,代码停留在第2行,而goods.py中则停留在第一行,
2个py文件相互等着。
# main.py
from flask import Flask
from goods import get_goods
app = Flask(__name__)
@app.route('/')
def index():
return 'index page'
if __name__ == '__main__':
app.run()
# goods.py
from main import app
@app.route('/get_goods')
def get_goods():
return 'get_goods page'
解决办法:推迟一方的导入让另一方先完成。
# main.py
from flask import Flask
from goods import get_goods
app = Flask(__name__)
# 装饰器原理
app.route('/get_goods')(get_goods)
@app.route('/')
def index():
return 'index page'
if __name__ == '__main__':
app.run()
# goods.py
# from main import app
# @app.route('/get_goods')
def get_goods():
return 'get_goods page'
blueprint实现方式
1. 在模块中实现
login main user都是项目根目录下的文件,login与user里面存放了部分视图函数
# main.py
#导入蓝图对象
fromloginimportlogins
fromuserimportusers
app=Flask(__name__)
#注册蓝图,第一个参数logins是蓝图对象,url_prefix参数默认值是根路由,如果指定,会在蓝图注册的路由url中添加前缀。
app.register_blueprint(logins,url_prefix='')
app.register_blueprint(users,url_prefix='')
# login.py
fromflaskimportBlueprint,render_template
#创建蓝图
logins=Blueprint('loginblueprint',__name__)
@logins.route('/login')
deflogin():
returnrender_template('login.html')
# user.py
fromflaskimportBlueprint,render_template
#创建蓝图,第一个参数指定了蓝图的名字。
users=Blueprint('userblueprint',__name__)
@users.route('/user')
defuser():
returnrender_template('user.html')
2. 在包中实现
cart是与main同在项目根目录下的一个包,views是cart包内的视图文件
# main.py
fromcartimportapp_cart
fromflaskimportFlask
app=Flask(__name__)
app.register_blueprint(app_cart,url_prefix='')
# __init__.py
fromflaskimportBlueprint
app_cart=Blueprint('app_cart',__name__,template_folder='templates',)
# 在cart包被导入时,把视图函数加载进来
# 让蓝图与app知道,cart中还有视图函数
from.viewsimportget_cart
# views.py
# 从当前目录,即init文件中
from.importapp_cart
fromflaskimportrender_template
@app_cart.route('/get_cart')
defget_cart():
returnrender_template('views.html')
单个py文件定义蓝图例子:
fromflaskimportBlueprint,session,redirect,request
blue_passport=Blueprint('blue_passport','__name__',
static_folder=None,
template_folder=None,
url_prefix=None)
@blue_passport.route('/login')
deflogin():
session['name'] ='jjj'
return'already login'
@blue_passport.route('/logout')
deflogout():
name=session.pop('name')
return'{} logout'.format(name)
@blue_passport.before_request
defblue_passport_before_request():
ifsession.get('name') =='jjj':
return
ifrequest.path=='/login':
return
returnredirect('/login')
多个app
简单项目与复杂项目的目录结构
参考projects/flask/的两个项目
目录结构及blueprint
admin与web应用分别包含单独的静态文件,视图函数,模版
admin与web的init文件分别包含定义蓝图和导入视图函数
pro_flask的init文件分别包含注册蓝图和定义app
.
├── pro_flask
│ ├── __init__.py
│ ├── admin
│ │ ├── __init__.py
│ │ ├── static
│ │ ├── templates
│ │ └── views.py
│ └── web
│ ├── __init__.py
│ ├── static
│ ├── templates
│ └── views.py
└── run.py
测试
常用的断言方法:
assertEqual 如果两个值相等,则pass
assertNotEqual 如果两个值不相等,则pass
assertTrue 判断bool值为True,则pass
assertFalse 判断bool值为False,则pass
assertIsNone 不存在,则pass
assertIsNotNone 存在,则pass
import unittest
class TestClass(unitest.TestCase):
#该方法会首先执行,相当于做测试前的准备工作
def setUp(self):
pass
#该方法会在测试代码执行完后执行,相当于做测试后的扫尾工作
def tearDown(self):
pass
#测试代码
def test_app_exists(self):
pass
数据库测试
import unittest
from author_book import *
#自定义测试类,setUp方法和tearDown方法会分别在测试前后执行。以test_开头的函数就是具体的测试代码。
class DatabaseTest(unittest.TestCase):
def setUp(self):
app.config['TESTING'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@localhost/test0'
self.app = app
db.create_all()
def tearDown(self):
db.session.remove()
db.drop_all()
#测试代码
def test_append_data(self):
au = Author(name='itcast')
bk = Book(info='python')
db.session.add_all([au,bk])
db.session.commit()
author = Author.query.filter_by(name='itcast').first()
book = Book.query.filter_by(info='python').first()
#断言数据存在
self.assertIsNotNone(author)
self.assertIsNotNone(book)
部署
Restful
REST:Representational State Transfer
REST是设计风格而不是标准。是指客户端和服务器的交互形式
REST的特点:
具象的。一般指表现层,要表现的对象就是资源。比如,客户端访问服务器,获取的数据就是资源。比如文字、图片、音视频等。
表现:资源的表现形式。txt格式、html格式、json格式、jpg格式等。浏览器通过URL确定资源的位置,但是需要在HTTP请求头中,用Accept和Content-Type字段指定,这两个字段是对资源表现的描述。
状态转换:客户端和服务器交互的过程。在这个过程中,一定会有数据和状态的转化,这种转化叫做状态转换。其中,GET表示获取资源,POST表示新建资源,PUT表示更新资源,DELETE表示删除资源。HTTP协议中最常用的就是这四种操作方式。
RESTful架构:
每个URL代表一种资源;
客户端和服务器之间,传递这种资源的某种表现层;
客户端通过四个http动词,对服务器资源进行操作,实现表现层状态转换。
服务器返回的数据格式,应该尽量使用JSON,避免使用XML。
API例子
域名部署
1将api部署在专用域名下:
http://api.example.com
2或者将api放在主域名下:
http://www.example.com/api/
版本号
将API的版本号放在url中。
http://www.example.com/app/1.0/info
http://www.example.com/app/1.2/info
路径
路径表示API的具体网址, 每个网址代表一种资源。网址中不能有动词 ,只能有名词,一般名词要与数据库的表名对应。而且名词要使用复数。
#获取单个商品
http://www.example.com/app/goods/1
#获取所有商品
http://www.example.com/app/goods
使用标准的HTTP方法
GET SELECT :从服务器获取资源。
POST CREATE :在服务器新建资源。
PUT UPDATE :在服务器更新资源。
DELETE DELETE :从服务器删除资源。
例如:
#获取指定商品的信息
GEThttp://www.example.com/goods/ID
#新建商品的信息
POSThttp://www.example.com/goods
#更新指定商品的信息
PUThttp://www.example.com/goods/ID
#删除指定商品的信息
DELETEhttp://www.example.com/goods/ID
响应结果
#返回商品列表
GEThttp://www.example.com/goods
#返回单个商品
GEThttp://www.example.com/goods/cup
#返回新生成的商品
POSThttp://www.example.com/goods
#返回一个空文档
DELETEhttp://www.example.com/goods
状态码
200OK:服务器成功返回用户请求的数据
201CREATED:用户新建或修改数据成功。
202Accepted:表示请求已进入后台排队。
400INVALIDREQUEST:用户发出的请求有错误。
401Unauthorized:用户没有权限。
403Forbidden:访问被禁止。
404NOTFOUND:请求针对的是不存在的记录。
406NotAcceptable:用户请求的的格式不正确。
500INTERNALSERVERERROR:服务器发生错误。
视图函数过滤信息
服务器不能将所有数据一次全部返回给客户端,API需要过滤
#指定返回数据的数量
http://www.example.com/goods?limit=10
#指定返回数据的开始位置
http://www.example.com/goods?offset=10
#指定第几页,以及每页数据的数量
http://www.example.com/goods?page=2&per_page=20
服务器返回的错误信息,以键值对的形式返回
{
error:'Invalid API KEY'
}
使用链接关联相关的资源
在返回响应结果时提供链接其他API的方法,使客户端很方便的获取相关联的信息。
性能
一、不同角度的网站性能
普通用户认为的网站性能
网站性能对于普通用户来说,最直接的体现就是响应时间。用户在浏览器上直观感受到的网站响应速度,即从客户端发送请求,到服务器返回响应内容的时间。
做为网站开发人员来说,网站性能通常会和普通的用户理解的不一样。
普通用户感受到的网站性能,并不只是由网站服务器决定的。它还包括客户端计算机和服务器通信的时间,网站服务器处理响应的时间,客户端浏览器构造请求解析响应数据的时间。甚至,不同的计算机性能、不同浏览器解析HTML的速度、不同网络运营商提供的网络带宽房屋的差异,这些都会导致用户感受到响应时间,可能大于网站服务器处理请求的时间。
开发人员认为的网站性能
开发人员关注的主要是服务器应用程序本身,以及相关配套系统的性能。包括并发处理能力、系统稳定性、响应延迟等技术指标。
对性能优化的主要手段,包括使用缓存加速数据读取,使用集群提高数据吞吐能力,使用异步消息加快请求响应,使用代码改善程序性能。
运维人员认为的网站性能
运维人员关注的主要是服务器基础设施和资源利用率。如服务器硬件的配置、网络运营商的带宽、数据中心的网络架构等。主要优化手段有使用高性价比的服务器、建设优化骨干网络、利用虚拟化技术优化资源利用等。
二、性能的指标
从开发人员的角度,网站性能的指标主要有并发数和响应时间。
并发数
并发数是指系统能够处理请求的数量,对于网站服务器而言,并发数也就是网站并发用户数,指同时提交请求的用户数目。
与并发数相对应的还有网站在线用户数(登录用户数)和网站用户数(一般指注册用户数)。他们的关系一般是:网站用户数>网站用户在线数>网站用户并发数
响应时间
响应时间是最重要的性能指标,直接反映了系统的快慢。
常见的系统操作响应时间
9.Flask中上下文管理主要是涉及到了哪些相关的类?并描述类主要的作用
RequestContext# 封装request, session (,app)等对象
AppContext# 封装app, g对象
Local# 里面有一个大字典,键为唯一标识,值是一个小字典,保存当前线程或者协程的数据。(保存请求上下文和应用上下文对象。)
LocalStack# 存放请求上下文和应用上下文,将Local对象中小字典的值,维护成一个栈(后进先出),通过列表实现的。
LocalProxy# 从LocalStack中获取请求上下文和应用上下文。操作对象的属性,比如传进去request,session,current_app, g等对象。可以设置获取属性。
10.为什么要把Flask Local对象大字典里面的小字典的键stack,维护程一个列表
当测试时,如果出现上下文嵌套的结构,列表中就会出现2个ctx, 比如测试数据库,有2个ctx可以一次性测试2个。
当写网站时永远都不会出现,只有在写脚本时才会出现, 在测试时可以使用,比如测试数据库,先测试完一个req_ctx,再测试另一个req_ctx
对于web程序可以不使用栈(列表),直接用单值,因为无论是一个app还是多个app,
外面的大字典都是通过线程或者协程号来标识的,即使多个app同时发送请求,开启的也是多个线程同时处理,每个线程对应的列表中还是只有一个ctx ;