系列其余内容
- docker疾速创立轻量级的可移植的容器✓
- docker&flask疾速构建服务接口✓
- docker&uwsgi高性能WSGI服务器生产部署必备
- docker&gunicorn高性能WSGI服务器生产部署必备
- docker&nginx&gunicorn实现负载平衡
- docker&ngxtop并实时解析nginx日志
- docker&supervisor监控你的服务
- docker&pyinstaller两步法构建小体积容器
- locust对你的服务做高并发测试
- postman热门的API调试工具
环境依赖
- 本教程是基于redhat linux服务器的
python: 3.8.3click==8.0.1Flask==2.0.1Flask-Limiter==1.4itsdangerous==2.0.1Jinja2==3.0.1limits==1.5.1MarkupSafe==2.0.1six==1.16.0Werkzeug==2.0.1WTForms==2.3.3
本文次要内容
- 包含docker部署flask服务、文件夹挂载、设置flask日志、设置参数验证局部、设置固定ip的申请次数限度、设置ip白名单。
docker&flask创立容器
python文件
- 设置debug=True,当文件更新时,服务会主动重启
import flask, jsonfrom flask import requestimport platform# 创立一个服务,把以后这个python文件当做一个服务app = flask.Flask(__name__)@app.route('/test', methods=['get'])def login(): username = request.values.get('name') pwd = request.values.get('pwd') system = platform.system() systemnode = platform.node() system_info = "平台是{0} & 运行节点是{1}".format(system, systemnode) if username and pwd: if username=='xiaoming' and pwd=='111': resu = {'code': 200, 'message': '登录胜利', 'system':system_info} return json.dumps(resu, ensure_ascii=False) else: resu = {'code': -1, 'message': '账号密码谬误', 'system':system_info} return json.dumps(resu, ensure_ascii=False) else: resu = {'code': 10001, 'message': '参数不能为空', 'system':system_info} return json.dumps(resu, ensure_ascii=False)if __name__ == '__main__': app.run(debug=True, port=2222, host="0.0.0.0")
- Dockerfile文件
FROM python:3.8WORKDIR /home/myfirstapi/RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtimeCOPY . .RUN pip install -r requirements.txt -q -i https://pypi.tuna.tsinghua.edu.cn/simple && \rm -rf /var/cache/apk/*expose 2222CMD ["python3", "flask_test.py"]
为了测试不便咱们特意设置
- python脚本中debug=True,当脚本更新时服务主动重启
- docker容器设置数据卷,使本地的更改能够主动同步到容器中。
# 构建名称为test/dockerflask,版本为1.0的镜像docker build -t test/dockerflask:1.0 . # 通过镜像test/api创立一个后盾运行的容器,且映射端口2222,将本地文件夹/root/first_api/flask_api挂载到容器指定目录下docker run -d -p 2222:2222 --name docker_flask_api -v /root/first_api/flask_api:/home/myfirstapi/ test/dockerflask:1.0
- 失去后果如下:
flask设置日志
能够参考python的logger库
def login(): ... app.logger.debug('this is a DEBUG message') app.logger.info('this is an INFO message') app.logger.warning('this is a WARNING message') app.logger.error('this is an ERROR message') app.logger.critical('this is a CRITICAL message')
- 失去后果如下
flask减少参数验证局部
设置验证局部
from wtforms.fields import simplefrom wtforms import Form, StringField, IntegerFieldfrom wtforms.validators import Length, Regexp, NumberRange, AnyOf, DataRequiredclass parameters_validation(Form): username = StringField(validators=[AnyOf(values = ["xiaoming", "laolitou"])]) pwd = StringField(validators = [DataRequired() Length(max=4,min=2,message="the length of the pwd must between %(min)d and %(max)d"), Regexp(regex="\d+",message="pwd must be start with numbers")], )@app.route('/test', methods=['get'])def login(): form = parameters_validation(request.args) if form.validate(): username = form.username.data pwd = form.pwd.data ... else: return jsonify(form.errors) if __name__ == '__main__': app.run(debug=True, port=2222, host="0.0.0.0")
- 失去后果如下
flask减少ip限度局部
flask设置ip拜访次数
from flask import Flaskfrom flask_limiter import Limiterfrom flask_limiter.util import get_remote_addressapp = Flask(__name__)limiter = Limiter( app, key_func=get_remote_address, default_limits=["5 per day", "2 per hour"])@app.route("/1times")@limiter.limit("1 per day")def slow(): return "每天只能拜访1次"@app.route("/2times") def fast(): return "每天能拜访5次,一小时内只能拜访2次"@app.route("/nolimits")@limiter.exempt # 无拜访速率限度def ping(): return "拜访无次数限度呀"if __name__ == '__main__': app.run(debug=True, port=2222, host="0.0.0.0")
失去后果如下
- 有了这个,对外提供给包月服务的话,这个是不是就能够很容易的加上服务次数的限度了啦
- 有了这个,对外提供给包月服务的话,这个是不是就能够很容易的加上服务次数的限度了啦
flask设置ip白名单
- flask设置ip白名单,只针对局部ip提供服务
from flask import abort, Flask, render_template, requestALLOWED_IPS = ['10.92', '10.91']app = Flask(__name__)@app.errorhandler(403)def permission_error(e): return "没权限呀没权限呀呈现了403谬误: %s"%e@app.before_requestdef limit_remote_addr(): client_ip = str(request.remote_addr) valid = False for ip in ALLOWED_IPS: if client_ip.startswith(ip) or client_ip == ip: valid = True break if not valid: abort(403)@app.route('/', methods = ['GET'])def home(): return "Your IP: {}".format(request.remote_addr)if __name__ == '__main__': app.run(host='0.0.0.0', debug=True)
- 失去如下后果
后续
- 一般而言,咱们正式提供一个服务是须要做负载平衡的,毕竟要思考用户的应用体验;
- 在nginx做负载平衡的过程中,申请限度,ip白名单也都是能够配置的。