flask 之(七) --- 認證|文件|部署

登錄註冊

說明: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))

  

安裝:

  •  pip install flask-httpauth
  • 爲了簡化,咱們將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實現:

  • 表單驗證 form
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.filenameform.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實現:

安裝:

  • pip install 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

  1. 前端向後端發起申請上傳的請求
  2. 後端根據雲存儲的 secret_key 及 雲存儲要求的特定算法(一會會由雲存儲的sdk提供)計算一個包含有效期的上傳令牌(token),並將此 token 返回給前端
  3. 前端拿到上傳 token 後,從本地直接將文件上傳到雲存儲服務器(上傳時會一塊兒將 token 提交),上傳成功後,雲存儲返回 url
  4. 前端將上傳後的文件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 就像油箱容器,咱們的程序就像汽油。把程序放到這個容器中,就能夠啓動了 

安裝: 

 

配置:

 

部署: 

相關文章
相關標籤/搜索