owned this note
owned this note
Published
Linked with GitHub
---
tags: Python
---
# Flask 網頁框架
flask是個微框架(microframework),相較於django把所有MTV(model、template、view)架構都打包好了,僅提供最基礎的功能,因此初期上手容易理解,但需要仰賴大量的套件
## 安裝python
下載[python 3.7](https://www.python.org/ftp/python/3.7.9/python-3.7.9-amd64.exe)可支援大多數的套件
- [X] 使用環境變數
```
pip show [某個套件]
pip install pipenv
pipenv shell
pip install -r requirements.txt
pip freeze > requirements.txt
```
## python 資料型態
list [i for i in range(10) if i%2==0]
dictionary {'key':value} keys() values() items()
tuple () zip(a,b)
## os檔案路徑
os.get_cwd()
os.path.join()
os.path.dirname(__file__)
os.path.basename()
os.path.splitext()
## 啟動flask
import os
from flask import Flask, Response, request, jsonify, url_for, redirect, session, send_from_directory
from flask_cors import CORS
app = Flask(__name__)
CORS(app) # 允許使用live server開發
# 資料庫位置(絕對路徑)
app['DATABASE_URI'] = os.path.join(os.path.dirname(__file__), 'flaskr.db')
if __name__ == '__main__':
app.secret_key = os.urandom(16).hex()
app.run(host='localhost', port=5000, debug=True)
## 動態路由
Request方法GET,POST,PUT,DELETE
Response(status codes)100~500
1xx: Information
2xx: Successful
200: OK
3xx: Redirection
4xx: Client Error
404: Not Found
5xx: Server Error
500: Internal Server Error
@app.route('/<path>', methods=['GET'])
靜態檔 send_from_directory('static/js', path)
## 檔案讀寫
import json
import sqlite3
import pandas as pd
with open('') as f:
f.read()
with open('','w',encoding='utf8') as f:
f.write()
json.dump(f,'',ensure_ascii=False)
with sqlite3.connect('') as con:
df = pd.read_sql('',con)
df.to_sql('',if_exists='append')
## SQL操作 - CRUD
(create, read, update, delete)
create
CREATE TABLE IF NOT EXISTS [TABLE] ([SCHEMA]);
INSERT INTO [TABLE]([COLUMNS]) VALUES ([VALUES]);
CREATE TRIGGER [TRIGGER] [BEFORE|AFTER|INSTEAD OF] [INSERT|UPDATE|DELETE] ON [TABLE] WHEN [condition];
CREATE UNIQUE INDEX id ON [TABLE] (...);
FOREIGN KEY [COLUMN] REFERENCES [TABLE] (...) ON DELETE CASCADE ON UPDATE NO ACTION;
read
SELECT * FROM [TABLE] WHERE account = [ACCOUNT];
update
ALTER TABLE [TABLE] RENAME [COLUMN] TO [TABLE|COLUMN];
UPDATE [TABLE] SET [PWD] = [PWD] WHERE id = [ID];
REPLACE INTO [TABLE] ([ID], [PWD]) VALUES ([ID], [PWD]);
delete
DELETE FROM [TABLE] WHERE id = [ID];
DROP TABLE|TRIGGER [TABLE|TRIGGER];
schema
type [INTEGER|REAL|TEXT|BLOB]
constraint [NOT NULL|PRIMARY KEY|AUTOINCREMENT|UNIQUE|DEFAULT]
default [CURRENT_TIMESTAMP]
script
BEGIN TRANSACTION; END; COMMIT;
MYSQL - LOAD DATA INFILE ‘[.csv]’ INTO TABLE [table] FIELDS TERMINATED BY ‘,’ LINES TERMINATED BY ‘\n’ IGNORE 1 ROWS;
postgreSQL - COPY DB.表格(欄位) FROM ‘文字檔的路徑’ WITH CSV HEADER DELIMITER AS ‘,’;
輸出 - SELECT * FROM [table] INTO OUTFILE ‘[.csv]’ FIELDS TERMINATED BY ‘,’ ENCLOSED BY ‘’ LINES TERMINATED BY ‘\n’;
sqlite - .mode csv .output test.csv SELECT * FROM [TABLE] .output stdout
## 登入設定(flask_login)
from flask_login import UserMixin, LoginManager, login_required, current_user, login_user, logout_user
class User(UserMixin):
pass
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'
@login_manager.user_loader
def load_user(user_id):
user = User()
user.id = user_id
return user
def hash_password(str_pw):
m = hashlib.md5(str_pw.encode(encoding='utf-8'))
return m.hexdigest()
## 整合matplotlib
import io
from matplotlib.backends.backend_agg import FigureCanvasAgg
from matplotlib.figure import Figure
fig = Figure()
ax = fig.add_subplot(1, 1, 1)
ax.legend(loc='lower left')
output = io.BytesIO()
FigureCanvasAgg(fig).print_png(output)
return Response(output.getvalue(), mimetype='image/png')
## 檔案上傳
form werkzeug.utils.secure_filename
<form action="http://localhost:5000/upload" method="post">
<input type="file" />
<input type="submit" />
</form>
let data = new FormData();
data.append('image', file);
json = await fetch({
url: 'http://localhost:5000/upload',
method: 'POST',
headers: { 'Content-Type': 'multipart/form-data' },
data: data,
responseType: 'json',
}).then((res) => res.data);
### 伺服器端架flask(ubuntu)
| 動作 | 指令 |
| -------- | -------- |
|建立連線密鑰(本地)|ssh-keygen -R 139.162.17.188|
|連線到伺服器(本地)|ssh root@139.162.17.188|
|加入使用者|adduser <使用者名稱>|
|給予使用者管理者權限|usermod -aG sudo <使用者名稱>|
|更改使用者密碼|passwd <使用者名稱>|
|編輯ssh連線設定|nano /etc/ssh/sshd_config|
|不允許root使用者登入,避免密碼被暴力破解|PermitRootLogin without-password|
|取消註解,使用ssh key登入|PubkeyAuthentication yes|
|重啟服務以生效|service ssh restart|
|切換到新增使用者|su - <使用者名稱>|
|適用防火牆之應用程式|ufw app list|
|允許加密的網絡傳輸|ufw allow OpenSSH|
|套用設定|ufw enable|
|防火牆狀態|ufw status|
|更新套件|sudo apt update|
|安裝代理伺服器|sudo apt install nginx|
|列出所有使用到防火牆的|sudo ufw app list|
|允許超文本傳輸協定|sudo ufw allow "Nginx HTTP"|
|防火牆狀態|sudo ufw status|
|系統服務狀態|systemctl status nginx|
|取得目前IP|curl -4 icanhazip.com|
|安裝python3套件|sudo apt install python3-pip|
|安裝虛擬環境|sudo apt install python3-venv|
|新增資料夾並進入|mkdir ~/myproject && cd $_|
|新增python虛擬環境|python3 -m venv myprojectenv|
|啟用虛擬環境|source myprojectenv/bin/activate|
|安裝核心套件|pip install wheel uwsgi flask|
|新增py檔|nano ~/myproject/myproject.py|
```
from flask import Flask
app = Flask(__name__)
@app.route("/")
def hello():
return "<h1 style='color:blue'>Hello There!</h1>"
if __name__ == "__main__":
app.run(host='0.0.0.0')
```
|依序輸入並儲存離開nano編輯環境|^x y enter|
|-|-|
|開放5000連接阜|sudo ufw allow 5000|
|執行py檔|python myproject.py|
|本地端測試|http://139.162.17.188:5000|
|新增wsgi.py檔|nano ~/myproject/wsgi.py|
```
from myproject import app
if __name__ == "__main__":
app.run()
```
|使用uwsgi執行|uwsgi --socket 0.0.0.0:5000 --protocol=http -w wsgi:app|
|-|-|
|本地端測試|http://139.162.17.188:5000|
|退出虛擬環境|deactivate|
|新增uWSGI起始設定檔|nano ~/myproject/myproject.ini|
```
[uwsgi]
module = wsgi:app
master = true
processes = 5
socket = myproject.sock
chmod-socket = 660
vacuum = true
die-on-term = true
```
|新增系統服務檔|sudo nano /etc/systemd/system/myproject.service|
|-|-|
```
[Unit]
Description=uWSGI instance to serve myproject
After=network.target
[Service]
User=sammy
Group=www-data
WorkingDirectory=/home/<使用者名稱>/myproject
Environment="PATH=/home/<使用者名稱>/myproject/myprojectenv/bin"
ExecStart=/home/<使用者名稱>/myproject/myprojectenv/bin/uwsgi --ini myproject.ini
[Install]
WantedBy=multi-user.target
```
|啟用系統服務|sudo systemctl start myproject|
|-|-|
|套用系統服務|sudo systemctl enable myproject|
|檢視系統服務|sudo systemctl status myproject|
|移除預設|sudo rm /etc/nginx/sites-available/default|
|新增|sudo nano /etc/nginx/sites-available/myproject|
```nginx=
server {
listen 80;
#server_name your_domain www.your_domain;
server_name http://139.162.17.188;
location / {
include uwsgi_params;
uwsgi_pass unix:/home/sammy/myproject/myproject.sock;
}
}
```
|建立檔案連結|sudo ln -s /etc/nginx/sites-available/myproject /etc/nginx/sites-enabled|
|-|-|
|測試代理伺服器|sudo nginx -t|
|重新開啟|sudo systemctl restart nginx|
|關閉5000連接阜|sudo ufw delete allow 5000|
|允許代理伺服器各種連線|sudo ufw allow "Nginx Full"|
|大功告成|http://139.162.17.188|
### SSL憑證(cPanel)
|產生憑證|openssl req -new -newkey rsa:2048 -nodes -keyout server.key -out domain_name.csr|
|-|-|
|依序填入,郵件密碼可跳過|TW、Taiwan、Taichung、ChuBoy Inc.、IT、www.domain_name.com|
|複製私鑰到資料夾下|mv domain_name.key /etc/ssl|
|把整段csr貼到SSL代理商上|-----BEGIN CERTIFICATE REQUEST-----|
|選擇DNS認證,點選GET RECORD複製CNAME|一開始沒有DNS選項,先選擇HTTP based,填入Email後到管理頁再調成DNS認證|
|將CNAME的HOST和Target貼到虛擬機服務商||
|等待通知,憑證商與虛擬機商配對完成後,到SSL代理商網站下載<domain_name>.crt,並丟到/etc/ssl裡||
|在/etc/nginx/sites-enabled修改crt對應位置如下||
```nginx=
server {
listen 443;
ssl on;
ssl_certificate /etc/ssl/<domain_name>.com.crt;
ssl_certificate_key /etc/ssl/<domain_name>.key;
server_name domain_name.com www.my_domain.com.com;
root /var/www/;
access_log /var/log/nginx/nginx.vhost.access.log;
error_log /var/log/nginx/nginx.vhost.error.log;
location / {
# if you are using localhost as a server
proxy_pass https://localhost:3000;
}
}
```
|從apt庫取得certbot|sudo add-apt-repository ppa:certbot/certbot|
|-|-|
|安裝apt套件|sudo apt install python-certbot-nginx|
||sudo certbot --nginx -d IP位址 -d 已註冊域名|
||sudo ufw delete allow 'Nginx HTTP'|
||sudo certbot --nginx|
||sudo certbot renew --dry-run|
||sudo certbot renew --dry-run|
|編輯工作排程|sudo crontab -e|
||30 4 1 * * sudo certbot renew|
### flask restful API(Swagger)
> pip install flask==1.1.2 flask_restplus==0.13.0 sqlalchemy==1.4.22 werkzeug==0.16.1 sqlacodegen
`__init__.py`
```python=
# 空白,告訴python當前資料夾為模組
```
`model.py`
```python=
# 透過以下指令自動產生
# sqlacodegen mssql+pymssql://localhost\SQLEXPRESS/J3WEBAP > model.py
```
`config.py`
```python=
# 配置系統設定
from sqlalchemy import create_engine, orm
engine = create_engine('mssql+pymssql://localhost\\SQLEXPRESS/J3WEBAP')
# engine = create_engine('mysql://account:passwordlocalhost/J3WEBAP')
# engine = create_engine('postgresql://account:passwordlocalhost/J3WEBAP')
# engine = create_engine('sqlite:///J3WEBAP.db')
con = orm.sessionmaker(engine)()
```
`apis/user_api.py`
```python=
from config import engine, con
from model import UserRoleDetail, UserHeader
from flask import request, url_for, redirect, send_from_directory
from flask_restplus import Namespace, Resource
from flask_restplus.fields import Boolean, Integer, String, Float, List
ns = Namespace('oet','公文製作')
# https://docs.sqlalchemy.org/en/14/orm/tutorial.html
# 定義JSON結構
UserRole = ns.model('UserRole', dict(
DepID = String,
DivID = String,
UserID = String,
UserNam = String,
UserAccount = String
))
UserDTO = ns.model('UserDTO', dict(
account = String,
))
# 配置路由(使用marshal_with或marshal_list_with會將Class轉成json)
@ns.route('/user/<UserNam>')
@ns.param('UserNam','使用者名稱')
@ns.response(404, 'Not found')
class UserRoute(Resource):
@ns.doc('讀取')
@ns.marshal_with(UserRole)
def get(self, UserNam):
return con.query(
UserRoleDetail.DepID,
UserRoleDetail.DivID,
UserRoleDetail.UserID,
UserHeader.UserNam,
UserHeader.UserAccount,
).filter(
UserRoleDetail.UserID==UserHeader.UserID,
UserHeader.UserNam==UserNam
).one()
@ns.doc('新增')
@ns.expect(UserDTO)
def post(self, UserNam):
account = request.json['account']
user = UserHeader(UserNam=UserNam, UserAccount=account)
con.add(user)
con.commit()
return 'OK'
@ns.doc('更新')
@ns.expect(UserDTO)
def put(self, UserNam):
account = request.json['account']
user = con.query(UserHeader).filter_by(UserNam=UserNam).one()
user.UserAccount = account
con.commit()
return 'OK'
@ns.doc('刪除')
def delete(self, UserNam):
user = con.query(UserHeader).filter_by(UserNam=UserNam).one()
con.delete(user)
con.commit()
return 'OK'
```
`app.py`
```python=
import os
from apis import user_api
from config import engine
from model import metadata
from flask import Flask, Blueprint
from flask_restplus import Api
metadata.create_all(engine) # 同步資料庫
app = Flask(__name__)
router = Blueprint('api', __name__, url_prefix='/api')
api = Api(router, version='1.0', title='DSICWEBAP', description='', doc='/doc')
api.add_namespace(user_api.ns)
app.register_blueprint(router)
if __name__ == '__main__':
app.secret_key = os.urandom(16).hex()
app.run(host='localhost', port=5000, debug=True)
```
### flask 檔案上傳
```python=
import os
from flask import Flask, request, redirect, url_for, render_template
from flask_cors import CORS
from werkzeug.utils import secure_filename
app = Flask(__name__)
CORS(app)
ALLOWED_EXT = ['txt','pdf','png','jpg','jpeg','gif']
@app.route('/',methods=['GET'])
def home():
if request.method == "GET":
return render_template('uploader.html')
@app.route('/upload',methods=['POST'])
def upload():
if request.method == "POST":
if request.files:
file = request.files['image']
if file.filename == '':
print('No selected file')
return redirect(request.url)
elif file.filename.split('.')[-1] in ALLOWED_EXT:
filename = secure_filename(file.filename)
file.save(os.path.join('', filename))
return 'uploaded'
else:
return 'invalid file type'
if __name__ == '__main__':
app.run(debug=True)
```
```htmlembedded=
<body>
<form action="http://localhost:5000/upload" method="post"
target="_blank" enctype="multipart/form-data">
<input id="file" type="file" name="image" onchange="uploadFile()">
<label id="upload-label" for="image">Select One...</label>
<input type="submit" value="提交">
</form>
<script>
function uploadFile(){
var file = document.getElementById('file').files[0];
var formdata = new FormData();
formdata.append('image',file);
var ajax = new XMLHttpRequest();
ajax.upload.addEventListener('progress', progressHandler);
ajax.open('POST','http://localhost:5000/upload');
ajax.send(formdata);
}
function progressHandler(e){
document.getElementById('upload-label').innerHTML=e.loaded+'bytes'+e.total;
}
</script>
</body>
```
### flask+keras 圖片辨識
```python=
import io
from keras.applications import ResNet50
from keras.preprocessing.image import img_to_array
from keras.applications import imagenet_utils
from PIL import Image
import numpy as np
from flask import Flask, request, jsonify
import tensorflow as tf
app = Flask(__name__)
model = None
def load_model():
global model
global graph
model = ResNet50(weights='imagenet')
graph = tf.get_default_graph()
def preprocess_image(image, target):
if image.mode != 'RGB':
image = image.covert('RGB')
image = image.resize(target)
image = img_to_array(image)
image = np.expand_dims(image, axis=0)
image = imagenet_utils.preprocess_input(image)
return image
@app.route('/predict', methods=['POST'])
def predict():
data = {'success': False}
print('post request')
if request.method == 'POST':
image = request.files['image'].read()
image = Image.open(io.BytesIO(image))
image = preprocess_image(image, target=(224,224))
with graph.as_default():
preds = model.predict(image)
results = imagenet_utils.decode_predictions(preds)
data['predictions'] = []
for (imagenetID, label, prob) in results[0]:
r = {'label':label, 'probability':float(prob)}
data['predictions'].append(r)
data['success'] = True
return jsonify(data)
if __name__ == '__main__':
print('loading keras model')
load_model()
app.run()
```