docker&flask快速構建服務介面(二)
-
系列其他內容
- 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.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創建容器
- 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")
- 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 //pypi.tuna.tsinghua.edu.cn/simple && \
rm -rf /var/cache/apk/*
expose 2222
CMD ["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 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白名單也都是可以配置的。