[Python知识库]docker&flask快速构建服务接口(二)

• 系列其他内容

  1. docker快速创建轻量级的可移植的容器?
  2. docker&flask快速构建服务接口?
  3. docker&uwsgi高性能WSGI服务器生产部署必备
  4. docker&gunicorn高性能WSGI服务器生产部署必备
  5. docker&nginx&gunicorn实现负载均衡
  6. docker&ngxtop并实时解析nginx日志
  7. docker&supervisor监控你的服务
  8. docker&pyinstaller两步法构建小体积容器
  9. locust对你的服务做高并发测试
  10. postman热门的API调试工具

环境依赖

• 本教程是基于redhat linux服务器的

python: 3.8.3
click==8.0.1
Flask==2.0.1
Flask-Limiter==1.4
itsdangerous==2.0.1
Jinja2==3.0.1
limits==1.5.1
MarkupSafe==2.0.1
six==1.16.0
Werkzeug==2.0.1
WTForms==2.3.3

• 本文主要内容
  • 包括docker部署flask服务、文件夹挂载、设置flask日志、设置参数验证部分、设置固定ip的请求次数限制、设置ip白名单。

docker&flask创建容器

1. python文件
   • 设置debug=True，当文件更新时，服务会自动重启

import flask, json
from flask import request
import 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")

2. Dockerfile文件

FROM python:3.8

WORKDIR /home/myfirstapi/

RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
COPY . .

RUN pip install -r requirements.txt -q -i https://pypi.tuna.tsinghua.edu.cn/simple && \
rm -rf /var/cache/apk/*

expose 2222
CMD ["python3", "flask_test.py"]

3. 为了测试方便我们特意设置
   • 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 simple
  from wtforms import Form, StringField, IntegerField
  from wtforms.validators import Length, Regexp, NumberRange, AnyOf, DataRequired
  class 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 Flask
  from flask_limiter import Limiter
  from flask_limiter.util import get_remote_address
  
  app = 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, request
    ALLOWED_IPS = ['10.92', '10.91']
    
    app = Flask(__name__)
    
    @app.errorhandler(403)
    def permission_error(e):
        return "没权限呀没权限呀出现了403错误: %s"%e
    
    @app.before_request
    def 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白名单也都是可以配置的。