python利用flask,gunicorn部署接口全过程

背景

无论开发 Android 还是 Web 端程序,有时候不免需要一些接口供自己调用进行测试,这里记录一下详细的过程。

环境配置

  1. 阿里云的 9.9 的学生特惠机, Ubuntu 14.04 64位
  2. 利用 putty 输入公网 ip 登录到服务器
    先配置编码为 utf-8



    输入账号密码

  3. 要利用 apt-get 包管理工具下载各种包,所以先更新下源,随手把 git 也安一下,安装中途输入 Y 即可

    1
    2
    apt-get update
    apt-get install git

Flask [flɑ:sk]

Flask是一个使用Python编写的轻量级Web应用框架。基于Werkzeug WSGI工具箱和Jinja2 模板引擎。 Flask 官方教程

我已经在本地写了一个简单的 Flask ,并且传到了 Coding

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#coding=utf-8
from flask import Flask,jsonify,request
app = Flask(__name__) #获得 Flask 实例
#写一个 POST 的方法,传一个 name 的参数,返回一个 json 数据
@app.route('/getJson', methods=['POST'])
def getJson():
#bala bala,下边随便返回点东西
name=request.form.get('name', default='windliang')
data={
'name':name,
'place':'wuhan'
}
msg='get success'
code=True
return getBaseReturnValue(data,msg,code)
#返回 json 数据
def getBaseReturnValue(data,msg,code):
json_data = jsonify({'data':data,'msg':msg,'success':code})
return json_data

if __name__ == '__main__':
app.run(host='0.0.0.0') #在服务器上运行

依次执行下边的指令,‘//’后边的仅做注释不用复制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
git clone https://git.coding.net/wind_liang/flasktest.git  //将 coding 上的代码克隆到本地
pip install virtualenv //python 的一个创建隔离环境的工具
mkdir api //创建一个新目录 api,virtualenv 直接用 clone 下的目录会出问题,原因不清楚
cp -i flasktest/api.py api //将 coding 下载下来的 api.py 复制到新目录 api 中
cd api // 进入 api 目录
virtualenv venv // 创建虚拟环境安包的文件夹
source venv/bin/activate //激活虚拟环境 ,这时会看到在最前边有 (venv) 标识 ,以后进入前都得先激活环境
pip install flask // 安装 flask 包
pip freeze > requirements.txt // 将安装的包列表写到 reqirements.txt 中,
以后每次 pip 安装了新的库的时候,都需freeze 一次,
既能知道自己安装了什么库,也方便别人部署时,安装相应的库。
python api.py // 运行程序
ctrl+C 用于终止当前 python 程序
deactivate 用于关闭当前虚拟环境 ,先不用执行此句

virtualenv 是一个将不同项目所需求的依赖分别放在独立的地方的一个工具,它给这些工程创建虚拟的Python环境。它解决了“项目X依赖于版本1.x,而项目Y需要项目4.x”的两难问题,而且使你的全局site-packages目录保持干净和可管理。官方教程

阿里云还得设定开放 5000端口

此时利用 postman (谷歌浏览器的一个插件),或者 curl 等其他能发 post 请求的工具进行测试

ctrl+C 终止程序,进入下一步

gunicorn

Gunicorn是一个Python WSGI UNIX的HTTP服务器。这是一个预先叉工人模式,从Ruby的独角兽(Unicorn)项目移植。该Gunicorn服务器与各种Web框架兼容,我们只要简单配置执行,轻量级的资源消耗,以及相当迅速。现在我们使用 flask 自带的服务器,完成了 web 服务的启动。生产环境下,flask 自带的服务器,无法满足性能要求。我们这里采用 gunicorn 做 wsgi容器,用来部署 python

1
2
3
4
5
6
7
8
9
10
11
pip install gunicorn  //安装 gunicorn
pip freeze > requirements.txt //保存到当前安装的包的列表中
gunicorn -w4 -b0.0.0.0:8000 api:app
此时,我们需要用 8000 的端口进行访问,
原先的5000并没有启用。
其中 gunicorn 的部署中,
-w 表示开启多少个 worker,-b
表示 gunicorn 的访问地址 ,
api 是程序进入的文件名,
app 是代码中获得的 flask 实例
进入阿里云记得开放8000端口


ctrl + c 终止当前程序

微信公众号服务器端 Flask 源码

python 2.7

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
# -*- coding: utf-8 -*-
# filename: main.py
from flask import Flask
import hashlib
import time
from flask import Flask,g,request,make_response
import xml.etree.ElementTree as ET
import requests
import re
import os
from bs4 import BeautifulSoup
import warnings
app = Flask(__name__)
BASE_DIR = os.path.dirname(__file__) #获取当前文件夹的绝对路径
warnings.filterwarnings("ignore") #忽略警告
@app.route("/wx",methods=["GET","POST"])
def wx():
if request.method == "GET": # 判断请求方式是GET请求
my_signature = request.args.get('signature') # 获取携带的signature参数
my_timestamp = request.args.get('timestamp') # 获取携带的timestamp参数
my_nonce = request.args.get('nonce') # 获取携带的nonce参数
my_echostr = request.args.get('echostr') # 获取携带的echostr参数

token = 'helloworld' # 一定要跟微信端填写的token一致

# 进行字典排序
data = [token,my_timestamp ,my_nonce ]
data.sort()

# 拼接成字符串
try:
temp = ''.join(data)
except:
return "success"
# 进行sha1加密
mysignature = hashlib.sha1(temp).hexdigest()

# 加密后的字符串可与signature对比,标识该请求来源于微信
if my_signature == mysignature:
return my_echostr
else:
rec = request.stream.read()
xml_rec = ET.fromstring(rec)
msgType=xml_rec.find("MsgType").text
if msgType != 'text': #只对文字进行回复
return "success"
tou = xml_rec.find('ToUserName').text
fromu = xml_rec.find('FromUserName').text
content = xml_rec.find('Content').text
cc=content.encode('UTF-8') # 用户发送的文字
t="狼吃羊"
if(cc==t):
content="http://windliang.oschina.io/worfeatsheep/"
else:
return "success"
xml_rep = "<xml><ToUserName><![CDATA[%s]]></ToUserName><FromUserName><![CDATA[%s]]></FromUserName><CreateTime>%s</CreateTime><MsgType><![CDATA[text]]></MsgType><Content><![CDATA[%s]]></Content><FuncFlag>0</FuncFlag></xml>"
response = make_response(xml_rep % (fromu,tou,str(int(time.time())), content))
response.content_type='application/xml'
return response



@app.route("/",methods=["GET"])
def index():
return "hello,world"



if __name__ == '__main__':
app.run(host='0.0.0.0',debug='true')

windliang wechat