說明:html
令牌Token認證,在對HTTP形式的API發請求時,大部分狀況咱們不是經過用戶名密碼作驗證,而是經過一個令牌[Token來作驗證]。前端
RESTful API沒法使用Flask-Login擴展來實現用戶認證。由於其沒有客戶端,經過postman請求,沒法設置cookie和session python
RESTful API不保存狀態,沒法依賴Cookie及Session來保存用戶信息。須要使用Flask-HTTPAuth擴展,完成RESTful API的用戶認證工做web
Flask-HTTPAuth提供了幾種不一樣的Auth方法,好比:HTTPBasicAuth、HTTPTokenAuth、MultiAuth、HTTPDigestAuth。 算法
此時,咱們就要使用Flask-HTTPAuth擴展中的HTTPTokenAuth對象中提供sql
」login_required」裝飾器來認證視圖函數、」error_handler」裝飾器來處理錯誤、」verify_token」裝飾器來驗證令牌。數據庫
認證:編程
兩種方式:一種:包含加密過的用戶數據 ;二種:不包含加密的用戶數據flask
一、使用flask的session,但沒有使用flask-session擴展,session數據沒法存儲在服務器上,後端
因此session的數據會被加密後保存到瀏覽器的cookies中。
此種方式至關於 token 中包含加密過的用戶數據 的狀況。
session['uid'] = 123
{'uid': 123}
sadufijklf260398riowehqklasdfnlk;d8rif2309wqeopsk
二、使用flask-session在服務器上存儲session數據,session數據不須要存儲在客戶端中,
可是須要在瀏覽器的cookies 中存儲一個session_id 的值,來標記當前請求對用的session是誰。
session_id = 875456786sdagadh
此種方式至關於 token 中不包含用戶數據
流程:
一、用戶登陸,登陸成功後,服務器爲此用戶生成一個 token(包含表明用戶身份信息的加密字符串)
{'uid': 123} 加密成 asdfjpoqwiefojklsd09823uiowejnfy8ij239-0eopifhdv789uiohjrefd8u9ioj
此加密字符串中也包含 時間信息,用於 token 的過時驗證
二、在登陸請求的響應中,向客戶端返回 上一步 生成的 token
三、客戶端再次請求服務器時,會在 請求頭 中攜帶 token
四、服務器從請求頭中獲取 token,進行解密工做,若是解密失敗:
一、不是服務器加密的數據,會解密失敗
二、超過了有效期
五、若是解密成功,則從 token 數據中得到用戶的身份信息,如:uid,能夠經過 uid 得到當前登陸用戶的用戶對象
加密解密:
itsdangerous庫提供了對信息加簽名(Signature)的功能,咱們能夠經過它來生成並驗證令牌。Flask 默認已經安裝。
1 from flask import Flask, g 2 from flask_httpauth import HTTPTokenAuth 3 from itsdangerous import TimedJSONWebSignatureSerializer as Serializer 4 from itsdangerous import BadSignature, SignatureExpired 5 6 auth = HTTPTokenAuth() 7 8 app = Flask(__name__) 9 app.config['SECRET_KEY'] = '110' 10 # 實例化了一個針對JSON的簽名序列化對象token_serializer。它是有時效性的,60分鐘後序列化後的簽名即會失效,就沒法解密 11 token_serializer = Serializer(app.config['SECRET_KEY'], expires_in=3600) # 參數:key ; 過時時間(秒) 12 13 users = ['jack', 'tom'] 14 for user in users: 15 # .dump(加密信息)對用戶信息進行加密,而後生成token;.loads(要解密的加密值)對數據解密 16 token = token_serializer.dumps({'username': user}).decode('utf-8') 17 print('*** token for {}: {}\n'.format(user, token))
安裝:
爲了簡化,咱們將Token與用戶的關係保存在一個字典中:
使用:
1 import datetime 2 from flask import current_app 3 from flask_sqlalchemy import SQLAlchemy 4 from itsdangerous import TimedJSONWebSignatureSerializer as Serializer 5 6 db = SQLAlchemy() 7 8 # 頻道 1 9 class Channel(db.Model): 10 __tablename__ = "channels" 11 12 id = db.Column(db.Integer, primary_key=True) 13 name = db.Column(db.String(16), unique=True, nullable=False) 14 sort = db.Column(db.Integer, nullable=False) 15 articles = db.relationship('Article', backref='channel', lazy='dynamic') 16 17 # 文章 N 18 class Article(db.Model): 19 __tablename__ = "articles" 20 21 id = db.Column(db.Integer, primary_key=True) 22 created_at = db.Column(db.DateTime, default=datetime.datetime.now()) 23 updated_at = db.Column(db.DateTime, default=datetime.datetime.now(), onupdate=datetime.datetime.now()) 24 title = db.Column(db.String(256), nullable=False) 25 content = db.Column(db.String(5000), nullable=False) 26 27 channel_id = db.Column(db.Integer, db.ForeignKey("channels.id")) 28 author_id = db.Column(db.Integer, db.ForeignKey('users.id')) 29 30 # 用戶 1 31 class User(db.Model): 32 __tablename__ = 'users' 33 34 id = db.Column(db.Integer, primary_key=True) 35 username = db.Column(db.String(32), unique=True, nullable=False) 36 password = db.Column(db.String(256), nullable=False) 37 articles = db.relationship('Article', backref='author', lazy='dynamic') 38 39 # 經過類方法,生成一個加密數據賦值給token變量,進行加密。 40 def generate_auth_token(self, expires_in=3600): 41 # Serializer(key驗證值密鑰,過時時間) 設置key密鑰和expires_in過時時間實例化Serializer。 42 serializer = Serializer(current_app.config['SECRET_KEY'], expires_in=expires_in) 43 # 經過Serializer實例化的serializer對象 把當前用戶的id用序列化器進行了加密,格式爲urf-8。 44 token = serializer.dumps({'user_id': self.id}).decode('utf-8') 45 return token 46 47 @classmethod 48 def check_auth_token(cls, token): 49 serializer = Serializer(current_app.config['SECRET_KEY']) 50 try: 51 data = serializer.loads(token) # 從token中,解密數據 52 except: 53 return None 54 if 'user_id' in data: 55 return cls.query.get(data['user_id']) # User.query.get() 56 else: 57 return None
1 # 在 __init__.py 文件中 2 3 from flask import Flask 4 from flask_restful import Api 5 from app import views, ext 6 from app.apis import ChannelList, ChannelDetail, ArticleList, ArticleDetail, UserRegister, UserLogin 7 from app.views import restful_bp 8 9 10 def create_app(): 11 app = Flask(__name__) 12 # session數據加密:from itsdangerous import TimedJSONWebSignatureSerializer 13 # TimedJSONWebSignatureSerializer 數據加密 14 # generate_password_hash,check_password_hash都會依賴app中的secret_key 15 app.config['SECRET_KEY'] = '110' 16 17 ext.init_db(app) 18 ext.init_migrate(app) 19 20 # 註冊實例化api擴展。prefix前綴名 21 api = Api(app, prefix='/api') 22 # 註冊頻道路由 23 api.add_resource(ChannelList, '/ChannelLists', endpoint='ChannelLists') 24 api.add_resource(ChannelDetail, '/ChannelDetails/<int:id>', endpoint='ChannelDetails') 25 # 註冊文章路由 26 api.add_resource(ArticleList, '/Articles', endpoint='Articles') 27 api.add_resource(ArticleDetail, '/ArticleDetails/<int:id>', endpoint='ArticleDetails') 28 # 用戶註冊登錄 29 api.add_resource(UserRegister, '/auth/register', endpoint='user_register') 30 api.add_resource(UserLogin, '/auth/login', endpoint='user_login') 31 32 # 藍圖 33 app.register_blueprint(blueprint=restful_bp) 34 return app
1 import os 2 from flask_httpauth import HTTPTokenAuth 3 from flask_migrate import Migrate 4 from app.models import db 5 6 migrate = Migrate() 7 auth = HTTPTokenAuth() 8 9 def init_db(app): 10 app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(app.root_path, 'sqlite3.db') 11 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False 12 db.init_app(app=app) 13 14 def init_migrate(app): 15 migrate.init_app(app=app,db=db)
1 import datetime 2 3 from flask import g 4 from flask_restful import Resource, reqparse, fields, marshal_with, abort 5 from requests import auth 6 from werkzeug.security import generate_password_hash, check_password_hash 7 from app.models import Channel, db, Article, User 8 9 # 輸入格式化驗證 10 user_parser = reqparse.RequestParser() 11 user_parser.add_argument('username', required=True, type=str) 12 user_parser.add_argument('password', required=True, type=str) 13 # 輸出格式化設置 14 user_fields = { 15 'id': fields.Integer, 16 'username': fields.String 17 } 18 19 # 用戶註冊 20 class UserRegister(Resource): 21 @marshal_with(fields=user_fields) 22 def post(self): 23 args = user_parser.parse_args() 24 # 須要自定義驗證,驗證用戶名是否重複。未寫 25 user = User() 26 user.username = args.get('username') 27 user.password = generate_password_hash(args['password']) 28 29 db.session.add(user) 30 db.session.commit() 31 return user, 201 32
33 # 用戶登錄 34 class UserLogin(Resource): 35 def post(self): 36 args = user_parser.parse_args() 37 user = User.query.filter_by(username=args['username']).first() 38 if user is None or not check_password_hash(user.password, args['password']): 39 return {'msg': '用戶名密碼錯誤'}, 400 40 token = user.generate_auth_token() # 登錄成功爲該用戶生成一個token認證值,返回給客戶端。 41 return {'token': token} # 返回給客戶端保存 42 # token驗證。裝飾器驗證成功則True;不然False 43 @auth.verify_token 44 def verify_token(token): 46 # 驗證token的回調函數。若是驗證成功則使用token中的用戶身份信息(user_id)從數據庫中查詢當前登陸用戶數據,返回uesr對象;若是驗證不成功,則返回None 51 user = User.check_auth_token(token) 52 if user is None: 53 return False 54 # 驗證成功後,將當前登陸用戶的對象設置到g對象中,供後續使用 55 g.user = user 56 return True
@auth.verify_token和@auth.login_required 幫助咱們對生成返回給客戶端的token進行沿驗證。
auth = HTTPTokenAuth()擴展初始化實例時不須要傳入app對象,也不須要調用auth.init_app(app)注入應用對象
保護:
@auth.login_required對頻道模塊的get請求進行保護
1 # 頻道模塊。get、put、patch、delete 2 class ChannelDetail(Resource): 3 """ 4 GET /channels/123 5 PUT /channels/234 6 PATCH /channels/123 7 DELETE /channels/123 8 """ 9 def get_object(self,id): 10 channel = Channel.query.get(id) 11 if channel is None: 12 return abort(404,message="找不到對象") 13 return channel 14 15 @auth.login_required # 對該函數的請求進行保護,token用戶驗證。若是該用戶的token值錯誤或者過時,該用戶將沒法訪問此函數 16 @marshal_with(fields=channel_article_fields) 17 def get(self,id): 18 channel = self.get_object(id) 19 return channel,200
1 import datetime 2 from flask import g 3 from flask_restful import Resource, reqparse, fields, marshal_with, abort, marshal 4 from werkzeug.security import generate_password_hash, check_password_hash 5 6 from app.ext import auth 7 from app.models import Channel, db, Article, User 8 9 # ============================ N =================================== 10 # 自定義一個類,用於時間格式化 11 class MyDTFmt(fields.Raw): 12 def format(self, value): 13 return datetime.datetime.strftime(value, '%Y-%m-%d %H:%M:%S') 14 15 16 # 定義參數驗證格式 17 article_parser = reqparse.RequestParser() 18 article_parser.add_argument('title', required=True, type=str, help="標題必填") 19 article_parser.add_argument('content', required=True, type=str, help="正文必填") 20 article_parser.add_argument('channel_id', required=True, type=int, help="頻道必填") 21 22 # 定義返回輸出格式 23 article_fields = { 24 'id': fields.Integer, 25 'url': fields.Url(endpoint='ArticleDetails', absolute=True), 26 'title': fields.String, 27 'content': fields.String, 28 # 等同於:'channel':fields.Nested(channel_fields), 29 'channel': fields.Nested({ # 經過Nested將對象解開 30 'name': fields.String, 31 'url': fields.Url(endpoint="ChannelDetails", absolute=True), 32 "sort": fields.Integer, 33 }), 34 "author": fields.Nested({ 35 'id': fields.Integer, 36 'name': fields.String(attribute='username'), 37 }), 38 'created_at': MyDTFmt, # 進行自定義時間格式化 39 'updated_at': fields.DateTime(dt_format="iso8601") 40 } 41 42 43 # 文章模塊。get、post 44 class ArticleList(Resource): 45 46 @auth.login_required 47 @marshal_with(fields=article_fields) 48 def get(self): 49 articles = Article.query.all() 50 return articles, 200 51 52 @auth.login_required 53 @marshal_with(fields=article_fields) 54 def post(self): 55 args = article_parser.parse_args() 56 57 article = Article() 58 article.title = args.get('title') 59 article.content = args.get('content') 60 article.channel_id = args.get('channel_id') 61 article.author_id = g.user.id # 登錄狀態下設置文章做者 62 63 db.session.add(article) 64 db.session.commit() 65 return article, 201
驗證:
經過postman請求登錄時返回的此用戶的token值,在postman中設置上此用戶的token值請求登錄訪問頻道列表
補充:
marshal的使用。api.py
1 import datetime 2 3 from flask import g 4 from flask_restful import Resource, reqparse, fields, marshal_with, abort, marshal 5 from werkzeug.security import generate_password_hash, check_password_hash 6 7 from app.ext import auth 8 from app.models import Channel, db, Article, User 9 10 # 輸出格式化設置 11 user_fields = { 12 'id': fields.Integer, 13 'username': fields.String 14 } 15 16 # 用戶登錄 17 class UserLogin(Resource): 18 def post(self): 19 args = user_parser.parse_args() 20 user = User.query.filter_by(username=args['username']).first() 21 if user is None or not check_password_hash(user.password, args['password']): 22 return {'msg': '用戶名密碼錯誤'}, 400 23 token = user.generate_auth_token() # 登錄成功生成一個token爲該用戶 24 # return {'token': token} # 返回給客戶端保存 25 26 ret = { 27 'code': 0, 28 'msg': '', 29 'data': { 30 # marshal至關於'user':'user.to_dict()' 將對象user與user_fileds中的同名參數進行匹配, 31 # marshal將user對象經過user_fields轉化爲字典形式傳遞出去 32 'user': marshal(user, user_fields), 33 'token': token, 34 } 35 } 36 return ret
原生實現:在實際生產中上傳大型文件,通常採用分段上傳
<form action="/upload" method="post" enctype="multipart/form-data"> <input type="file" name="file"> <input type="submit" value="Upload"> </form> {# 設置表單類型爲enctype="multipart/form-data" 上傳文件才能夠生效#}
經過request對象上的files"file = request.files['file']"獲取文件;使用 save() 方法保存文件到指定位置
1 import os 2 from flask import Flask, request, render_template, send_from_directory, url_for 3 from werkzeug.utils import secure_filename 4 5 app = Flask(__name__) 6 # 設置上傳存放文件的文件夾。uploads是此項目下的手動建立的目錄。注意:手動建立 7 app.config['UPLOAD_FOLDER'] = os.path.join(app.root_path, 'uploads') 8 # 設置最大上傳文件的大小值。10 M;必定要設置此值,不設置可能存在漏洞 9 app.config['MAX_CONTENT_LENGTH'] = 10 * 1024 * 1024 10 11 # 判斷文件的類型。原理:對文件名右邊以點切割,獲取腳標爲1的參數,便是擴展名 12 def allowed_file(filename): 13 return '.' in filename and \ 14 filename.rsplit('.', 1)[1] in {'png', 'jpg', 'jpeg', 'gif'} 15 16 @app.route('/uploads/<filename>') 17 def uploaded_file(filename): 18 """ 19 send_from_directory,從指定的目錄中(第一個參數)查找一個指定名字的文件(第二個參數) 20 若是找到,則將文件讀入內存,而且最爲響應返回給瀏覽器。只會在開發環境中使用 21 """ 22 return send_from_directory(app.config['UPLOAD_FOLDER'],filename) 23 24 @app.route('/upload-test/', methods=['POST', 'GET']) 25 def upload_test(): 26 if request.method == 'POST': 27 # 從表單中得到名爲photo的上傳文件對象,經過request.files.get獲取。 28 # files類型是字典,一個表單中能夠有多個上傳文件,能夠經過遍歷獲取多個上傳文件 29 file = request.files.get('photo') 30 31 if file: 32 # 首先判斷文件的 擴展名 是否被容許 33 if allowed_file(file.filename): 34 # 將文件名作一個安全處理:'my movie.mp4' -> 'my_movie.mp4' 35 filename = secure_filename(file.filename) 36 # 將文件保存到指定目錄(第一個參數),以某個文件名(第二個參數) 37 # 第二個參數是保存文件的文件名,能夠自定義修改。filename = 'abc.def' 38 file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename)) 39 # 經過url_for和uploaded_file視圖函數,反向解析出文件的訪問地址。供前端訪問 40 file_url = url_for('uploaded_file', filename=filename) 41 return render_template('upload.html', photo_url=file_url) 42 else: 43 return render_template('upload.html', err='不支持的文件類型') 44 return render_template('upload.html') 45 46 47 if __name__ == '__main__': 48 app.run()
結合flask-wtf實現:
1 from flask_wtf import Form 2 from flask_wtf.file import FileField, FileAllowed, FileRequired 3 4 class PhotoForm(Form): 5 photo = FileField('photo', validators=[ 6 FileRequired(), # 不能爲空 7 # 驗證文件格式 8 FileAllowed(['jpg', 'png', 'webp'], '只能上傳圖片') 9 ])
form.photo.data.filename和form.photo.data.save進行獲取和保存
1 import os 2 from flask import Flask, render_template, url_for 3 from werkzeug.utils import secure_filename 4 from forms import PhotoForm 5 6 app = Flask(__name__) 7 # flask-wtf中生成csrf token必須依賴SECRET_KEY 8 app.config['SECRET_KEY'] = '110' 9 # 設置上傳文件的文件夾,在static中 10 app.config['UPLOAD_FOLDER'] = os.path.join(app.root_path, 'static/uploads') 11 12 13 @app.route('/upload/', methods=('GET', 'POST')) 14 def upload(): 15 form = PhotoForm() 16 file_url = '' 17 18 if form.validate_on_submit(): 19 # 獲取photo中的數據 20 filename = secure_filename(form.photo.data.filename) 21 form.photo.data.save(os.path.join(app.config['UPLOAD_FOLDER'], filename)) 22 file_url = url_for('static', filename='uploads/' + filename) 23 return render_template('upload.html', form=form, file_url=file_url) 24 25 26 if __name__ == '__main__': 27 app.run()
1 <body> 2 {% if form.errors %} 3 <div> 4 {{ form.errors }} 5 </div> 6 {% endif %} 7 8 <form action="" method="post" enctype="multipart/form-data"> 9 {{ form.csrf_token() }} 10 <input type="file" name="photo"> 11 <input type="submit" value="Upload"> 12 </form> 13 14 <div> 15 photo: <img src="{{ file_url }}" alt=""> 16 </div> 17 </body>
flask-uploads實現:
安裝:
1 import os 2 from flask import Flask, render_template 3 from flask_uploads import UploadSet, configure_uploads, IMAGES, patch_request_class 4 from flask_wtf import FlaskForm 5 from flask_wtf.file import FileField, FileRequired, FileAllowed 6 7 app = Flask(__name__) 8 app.config['SECRET_KEY'] = '110' 9 10 # 爲 flask_uploads 建立一個上傳目錄 11 app.config['UPLOADED_PHOTOS_DEST'] = os.path.join(app.root_path, 'uploads') 12 # 設置上傳文件類型集合。能夠將IMAGES[第二個參數]文件中的文件類型上傳到名字叫photos[第一個參數]的文件中 13 photos = UploadSet('photos', IMAGES) 14 configure_uploads(app, photos) 15 # 設置最大上傳大小, 默認 64 MB 16 patch_request_class(app) 17 18 class UploadForm(FlaskForm): 19 photo = FileField(validators=[ 20 FileAllowed(photos, '只能上傳圖片!'), 21 FileRequired('文件未選擇')]) 22 23 @app.route('/upload/', methods=['GET', 'POST']) 24 def upload_file(): 25 form = UploadForm() 26 if form.validate_on_submit(): 27 # 將 文件(form.photo.data)保存到 photos 上傳集合中 28 filename = photos.save(form.photo.data) 29 # 得到文件的 url 30 file_url = photos.url(filename) 31 # 實際項目中,會把文件的 url 保存到數據庫中 32 else: 33 file_url = None 34 return render_template('upload.html', form=form, file_url=file_url) 35 36 37 if __name__ == '__main__': 38 app.run()
1 <body> 2 <form method="post" enctype="multipart/form-data"> 3 {{ form.csrf_token() }} 4 {{ form.photo }} 5 {% for error in form.photo.errors %} 6 <p>{{ error }}</p> 7 {% endfor %} 8 <input type="submit" value="upload"> 9 </form> 10 11 {% if file_url %} 12 photo:<img src="{{ file_url }}"> 13 {% endif %} 14 </body>
先後端的分離實現:
目前企業中,大多數上傳文件的業務會把文件存儲在 雲存儲 中,每一個項目會向雲存儲申請上傳的 secret_key,經過 secret_key 和 雲存儲上傳api,將用戶的文件上傳到雲存儲中,並返回文件url
介紹:
WSGI:W web、S server、G gateway、I interface。俗稱:web容器,web服務器
Nginx:也是一個服務器軟件,反向代理、負載均衡;Nginx 背後能夠有多個 WSGI 服務器
Nginx 是不不管什麼編程語言均可以使用的代理服務器軟件。性能強悍、用戶多、漏洞少、穩定性強。
它接收用戶的請求後,會轉發給真正的運行python代碼的服務器WSGI容器,WSGI容器內部會去執行python代碼[python代碼放在WSGI容器中才能夠運行起來]
過程:
用戶訪問時經過Nginx,Nginx會負載均衡、反向代理到WSGI容器上。python的代碼是不能夠直接與Ngins交互的,Nginx是不知道如何執行python代碼的,
因此須要python代碼是在WSGI SERVER容器中運行的,經過WSGI運行就能夠提供http服務;執行完後會返回app響應,相應會交給Nginx。最終返回給客戶端
WSGI, 是一個協議,pep 333/3333 規範的協議,規定了服務器如何與應用程序交互web容器,執行咱們的應用程序(好比:flask + 業務代碼)
uWSGI,是一個 服務器軟件,實現了 WSGI 協議,能夠運行標準的 WSGI 應用程序(Flask 應用)。同時也有本身的特有協議:uwsgi
uwsgi, 是 uWSGI 特有的協議
/ WSGI SERVER 1 <======> Python(Flask app) client <-----> Nginx - WSGI SERVER 2 <======> Python(Flask app) \ WSGI SERVER 3 <======> Python(Flask app)
WSGI 就像油箱容器,咱們的程序就像汽油。把程序放到這個容器中,就能夠啓動了
安裝:
配置:
部署: