Python全栈6 - RESTful API 开发

以你之姓@ 2022-07-03 12:15 212阅读 0赞

6、Python RESTful API 开发

**1、RESTful API 概述**

1-1、展示微博开放平台的 RESTfulAPI

介绍微博开放平台

open.weibo.com

在linux或mac中，$ curl 请求URL**?**请求参数

pro.jsonlint.com上

通过HTTP请求，请求到JSON的过程，就是RESTful的调用

1-2、RESTful 设计理念

REST - Representational State Transfer 表现层状态转化

资源– resources – 网络上的具体信息，如文本，图片

URI – 统一资源标识符，用来唯一的**标识**一个资源

URL – 统一资源定位器，用来定位某个特定的资源，如一个**网址**

表现层（Representation）- 把资源具体呈现出来的形式

如：纯文本，HTML，JSON

状态转移 - State Transfer

HTTP协议，是一个**无状态**的协议，所有的状态都保存在服务器端

GET– 获取资源

POST – 新建资源

PUT– 更新资源

DELETE – 删除资源

一个RESTful请求：$ curl **–X GET**https://api.weibo.com/2/users/show.json

REST构架设计6原则

UniformInferface

Stateless

Cacheable

Client-Server

Layered System

Code on Demand

1-3、Python微型Web框架Flask简介

Flasks是一个基于Werkzeug，Jinja2以及”善意”构建的Python微型Web框架，并且基于BSD 开源证书！

1-4、一个例程和总结

Debug模式：

If \_\_name\_\_ == ’\_\_main\_\_’:

App.run(debug=True)

Debug模式下，可以看到代码，可以直接调试

在浏览器中，使用POST方法较麻烦，但是可以在命令行中进行

$ curl **–X POST** 127.0.0.1:5000/index/yx

**2、Python RESTful API开发工具介绍及应用**

2-1 Chrome 开发者工具介绍

**Network**

**Elements**

**Console**

使用console.table()方法，创建一个表格

使用console.timeEnd()方法，统计一个操作所花费的时间

实例：

进行移动端的开发，点击**手机图标**

2-2 Python HTTP 库 Requests 介绍

[http://requests-docs-cn.readthedocs.org/zh\_CN/latest/user/quickstart.html][http_requests-docs-cn.readthedocs.org_zh_CN_latest_user_quickstart.html]

实例：

\[root@localhost ~\]\# python2.7

Python 2.7.11 (default, Apr  6 2016, 00:07:19)

\[GCC 4.1.2 20080704 (Red Hat 4.1.2-55)\] onlinux2

Type "help","copyright", "credits" or "license" for moreinformation.

>>> import requests

>>> r =requests.get('https://github.com/timeline.json')  **//使用requests发送网络请求**

/usr/local/lib/python2.7/site-packages/requests/packages/urllib3/util/ssl\_.py:315:SNIMissingWarning: An HTTPS request has been made, but the SNI (Subject NameIndication) extension to TLS is not available on this platform. This may causethe server to present an incorrect TLS certificate, which can cause validationfailures. For more information, seehttps://urllib3.readthedocs.org/en/latest/security.html\#snimissingwarning.

SNIMissingWarning

>>> r.text

u'\{"message":"Hello there,wayfaring stranger. If you\\u2019re reading this then you probably didn\\u2019tsee our blog post a couple of years back announcing that this API would goaway: http://git.io/17AROg Fear not, you should be able to get what you needfrom the shiny new Events API instead.","documentation\_url":"https://developer.github.com/v3/activity/events/\#list-public-events"\}'

>>> payload = \{'key1': 'value1','key2': 'value2'\}  **//为URL传递参数**

>>> r = requests.get("http://httpbin.org/get",params=payload)

>>> r.text  **//读取服务器响应的内容，unicode字符集**

u'\{\\n "args": \{\\n   "key1": "value1", \\n    "key2": "value2"\\n  \}, \\n  "headers":\{\\n    "Accept":"\*/\*", \\n   "Accept-Encoding": "gzip, deflate", \\n    "Host": "httpbin.org",\\n    "User-Agent":"python-requests/2.9.1"\\n  \},\\n  "origin": "114.255.40.54",\\n  "url":"http://httpbin.org/get?key2=value2&key1=value1"\\n\}\\n'

>>> r.url ** //访问URL**

u'http://httpbin.org/get?key2=value2&key1=value1'

>>> r.json()  **//返回服务器JSON格式的内容**

\{u'origin': u'114.255.40.54', u'headers':\{u'Host': u'httpbin.org', u'Accept-Encoding': u'gzip, deflate', u'Accept':u'\*/\*', u'User-Agent': u'python-requests/2.9.1'\}, u'args': \{u'key2': u'value2',u'key1': u'value1'\}, u'url':u'http://httpbin.org/get?key2=value2&key1=value1'\}

>>> **type**(r.text)

>>> **type**(r.json())

>>>

C:\\Users\\Administrator>python

Python 2.7.11 (v2.7.11:6d1b6a68f775,Dec  5 2015, 20:32:19) \[MSC v.1500 32 bit(

Intel)\] on win32

Type "help","copyright", "credits" or "license" for moreinformation.

>>> from PIL import Image   **//以请求返回的二进制数据创建一张图片**

>>> from StringIO import StringIO

>>> import requests

>>> r =requests.get('http://requests-docs-cn.readthedocs.org/zh\_CN/latest/\_stat

ic/requests-sidebar.png')

>>> i =Image.open(StringIO(r.content))

>>> i.show()

>>>

//定制请求头 – 模拟一个浏览器

\# requests\_t.py - **Browser**

import requests

hdr =\{"User-Agent":"Mozilla/5.0 (Windows NT 6.1; WOW64)AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.110Safari/537.36"\}

r = requests.get(**'http://127.0.0.1:5000/test1'**, headers = hdr)

**print** r.text

\# app.py - **Server**

from flask import Flask,request

app = Flask(\_\_name\_\_)

@app.route('/')

def index():

print request.headers

return 'Hello restful'

if \_\_name\_\_ == '\_\_main\_\_':

app.run(host='0.0.0.0', port=5000, debug=True)

2-3 实践：动手编写一个验证登录的程序

token是一个**加密的字符串**，其中包含用户名，过期时间，以及一些随机信息

base64对token进行加密

代码：

\# app.py

import base64  
import random  
import time  
from flask import Flask, request  
  
  
app = Flask(\_\_name\_\_)  
  
  
users = \{  
    "magigo": \["123456"\]  
\}  
  
  
def gen\_token(uid):  
    token = base64.b64encode(':'.join(\[str(uid), str(random.random()), str(time.time() + 7200)\]))  
    users\[uid\].append(token)  
    return token  
def verify\_token(token):  
    \_token = base64.b64decode(token)  
    if not users.get(\_token.split(':')\[0\])\[-1\] == token:  
        return -1  
    if float(\_token.split(':')\[-1\]) >= time.time():  
        return 1  
    else:  
        return 0  
  
  
@app.route('/index', methods=\['POST', 'GET'\])  
def index():  
    print request.headers  
    return 'hello'  
  
  
@app.route('/login', methods=\['POST', 'GET'\])  
def login():  
    uid, pw = base64.b64decode(request.headers\['Authorization'\].split(' ')\[-1\]).split(':')  
    if users.get(uid)\[0\] == pw:  
        return gen\_token(uid)  
    else:  
        return 'error'  
  
  
@app.route('/test1', methods=\['POST', 'GET'\])  
def test():  
    token = request.args.get('token')  
    if verify\_token(token) == 1:  
        return 'data'  
    else:  
        return 'error'  
  
  
if \_\_name\_\_ == '\_\_main\_\_':  
    app.run(debug=True)  
   
  
  
\#requests\_r.py  
  
  
import requests  
\# r = requests.get('http://127.0.0.1:5000/login', auth=('magigo', '123456'))  
\# print r.text  
  
  
token ='bWFnaWdvOjAuMzE4MTUxNTA1MjQ4OjE0MjU4MzkzMjMuODk='  
r = requests.get('http://127.0.0.1:5000/test1', params=\{'token': token\})  
print r.text  
\# bWFnaWdvOjAuMzE4MTUxNTA1MjQ4OjE0MjU4MzkzMjMuODk=

**3、OAuth 2.0介绍和实现（上）**

3-1 OAuth 2.0的原理介绍

OAuth – 开放授权 – 一个正式的互联网标准协议

三方协作的过程：用户、网站、微博

Token – 令牌

OAuth概述 – 其中，B步骤是关键

OAuth四种授权模式

授权码模式

简化模式

密码模式

客户端模式

授权码模式

3-2 实现OAuth 2.0协议中的必选方法

1）实现重定向的机制

\# app.py

@app.route('/client/login')

def client\_login():

uri = 'http://localhost:5000/oauth'

return redirect(uri)

\# request\_t.py

r =requests.get('http://localhost:5000/client/login')

print r.text

print '======='

print r**.history**    //通过history属性，取得**重定向**情况

\# app.py

@app.route('/oauth')

def oauth():

return 'Please login'

返回结果：

Please login

=======

\[<Response \[**302**\]>\]   //表示重定向

2）实现授权码的机制

auth\_code = \{\}

def gen\_code(uri):

code = random.randint(0,10000)

auth\_code\[code\] = uri

return code

3）实现token发放的机制

@app.route('/oauth')

def oauth():

if request.args.get('code'):

if auth\_code.get(int(request.args.get('code'))) ==request.args.get('redirect\_uri'):

return gen\_token(request.args.get('client\_id'))

return 'Pleaselogin'

3-3 编写OAuth授权服务器

知识点：

清除OAuth 2.0协议的内容

掌握Flask的重定向操作

掌握OAuth 2.0协议中的授权模式

代码：

*\#app.py*  
**import** base64  
**import** random  
**import** time  
  
**from** flask **import**Flask, request, redirect  
  
app = Flask(\_\_name\_\_)  
  
users = \{  
    **"yx"**: \[**"35"**\]  
\}  
  
redirect\_uri=**'http://localhost:5000/client/passport'**  
  
client\_id = **'35'**  
users\[client\_id\] = \[\]  
auth\_code = \{\}  
oauth\_redirect\_uri = \[\]  
  
*\# 授权码生成器*  
**def** gen\_auth\_code(uri):  
    code = random.randint(0,10000)  
    auth\_code\[code\] = uri  
    **return** code  
  
**def** gen\_token(uid):  
    token = base64.b64encode(**':'**.join(\[str(uid),str(random.random()), str(time.time() +7200)\]))  
    users\[uid\].append(token)  
    **return** token  
  
**def** verify\_token(token):  
    \_token = base64.b64decode(token)  
    **if not** users.get(\_token.split(**':'**)\[0\])\[-1\] == token:  
        **return** \-1  
    **if** float(\_token.split(**':'**)\[-1\]) >= time.time():  
        **return** 1  
    **else**:  
        **return** 0  
  
@app.route(**'/index'**,methods=\[**'POST'**,**'GET'**\])  
**def** index():  
    **print** request.headers  
    **return 'hello'**  
  
@app.route(**'/login'**,methods=\[**'POST'**,**'GET'**\])  
**def** login():  
    uid, pw =base64.b64decode(request.headers\[**'Authorization'**\].split(**' '**)\[-1\]).split(**':'**)  
    **if** users.get(uid)\[0\] == pw:  
        **return** gen\_token(uid)  
    **else**:  
        **return 'error'**  
  
@app.route(**'/oauth'**,methods=\[**'POST'**,**'GET'**\])  
**def** oauth():  
    *\# 登录  
   * **if** request.args.get(**'user'**):  
        **if** users.get(request.args.get(**'user'**))\[0\] == request.args.get(**'pw'**)**and** oauth\_redirect\_uri:  
            uri = oauth\_redirect\_uri\[0\] + **'?code=%s'** % gen\_auth\_code(oauth\_redirect\_uri\[0\])  
            **return** redirect(uri)  
    *\# 验证授权码，发放token  
   * **if** request.args.get(**'code'**):  
        **if** auth\_code.get(int(request.args.get(**'code'**))) == request.args.get(**'redirect\_uri'**):  
            **return** gen\_token(request.args.get(**'client\_id'**))  
    *\#  
   * **if** request.args.get(**'redirect\_uri'**):  
       oauth\_redirect\_uri.append(request.args.get(**'redirect\_uri'**))  
    **return 'please login'**  
  
*\# 客户端*  
@app.route(**'/client/login'**,methods=\[**'POST'**,**'GET'**\])  
**def** client\_login():  
    uri = **'http://localhost:5000/oauth?response\_type=code&client\_id=%s&redirect\_uri=%s'**% (client\_id, redirect\_uri)  
    **return** redirect(uri)  
  
@app.route(**'/client/passport'**,methods=\[**'POST'**,**'GET'**\])  
**def** client\_passport():  
    code = request.args.get(**'code'**)  
    uri = **'http://localhost:5000/oauth?grant\_type=authorization\_code&code=%s&redirect\_uri=%s&client\_id=%s'**% (code, redirect\_uri, client\_id)  
    **return** redirect(uri)  
  
@app.route(**'/test1'**,methods=\[**'POST'**,**'GET'**\])  
**def** test():  
    token = request.args.get(**'token'**)  
    **if** verify\_token(token) ==  1:  
        **return 'data'  
    else**:  
        **return 'error'  
  
if** \_\_name\_\_ == **'\_\_main\_\_'**:  
    app.run(debug=True)

\# requests.py

*\#!/usr/bin/env python  
\# -\*- coding: utf-8 -\*-*  
  
**import** requests  
  
r = requests.get(**'http://127.0.0.1:5000/client/login'**)  
**print** r.text  
**print '========='  
print** r.history  
**print '========='  
print** r.url  
**print '========='**  
uri\_login = r.url.split(**'?'**)\[0\] +**'?user=yx&pw=35'**  
r2 = requests.get(uri\_login)  
**print** r2.text  
**print** r2.history  
**print '========='**  
r3 = requests.get(**'http://127.0.0.1:5000/test1'**,params=\{ **'token'**: r2.text\})  
**print** r3.text

**4、OAuth 2.0介绍和实现（下）**

4-1 Flask渲染页面及Cookies

Flask中的表单

提交表单时，使用**POST方法**

[ http://dormousehole.readthedocs.org/en/latest/quickstart.html\#sessions][http_dormousehole.readthedocs.org_en_latest_quickstart.html_sessions]

Flask响应对象

ResponseHeaders – 浏览器的响应头

http://dormousehole.readthedocs.org/en/latest/quickstart.html\#about-responses

Cookies简介 – 小型文本文件

1）分类：Cookies总是存在客户端中，按在客户端中的存储位置，可分为内存Cookies和硬盘Cookies

2）用途：Cookies可以用于购物车，以及免登陆

3）缺陷：Cookies增加了网络流量，造成安全问题，无法存储大量数据

http://dormousehole.readthedocs.org/en/latest/quickstart.html\#id11

4-2 Token的设计以及加密方法

知识点：

无需核对用户名密码的token验证机制 – 无法被伪造的token

hmac（哈希运算消息认证码）简介

Token的基本设计原则：

Token不携带用户敏感信息

**无需查询数据库**，Token可以进行自我验证

**hmac**简介

作用：验证消息是否被篡改

公式：

使用Python的hmac模块直接计算：hmac.new(‘secret123’,value).digest()

缺陷：无法抵御重放攻击

验证token的有效性：**真的**，**没过期**，**合法请求**

验证验证码是否一致

验证是否过期 – 生存期10分钟左右

验证这个token是否属于被请求数据的用户

4-3 最终的编码

总结：

掌握Flask渲染页面和处理响应对象

了解hmac加密验证的内容

安全建议：使用HTTPS传输，使用请求头而不是参数传递Token

实战：制作一个基于OAuth 2.0的验证登录服务

在Chrome开发者工具中，resources-> Cookies->localhost中，可以看到用户名和密码

代码：

*\#!/usr/bin/env python  
\# -\*- coding: utf-8 -\*-*  
  
**import** base64  
**import** random  
**import** time  
**import** json  
**import** hmac  
**from** datetime **import**datetime,timedelta  
  
**from** flask **import**Flask, request, redirect, make\_response  
  
app = Flask(\_\_name\_\_)  
  
users = \{  
    **"yx"**: \[**"35"**\]  
\}  
  
redirect\_uri=**'http://localhost:5000/client/passport'**  
client\_id = **'123456'**  
users\[client\_id\] = \[\]  
auth\_code = \{\}  
oauth\_redirect\_uri = \[\]  
  
TIMEOUT = 3600 \* 2  
  
  
*\# 新版本的token生成器*  
**def** gen\_token(data):  
    *'''  
   * **:param** *data: dict type  
   * **:return***: base64 str  
    '''  
   * data = data.copy()  
    **if "salt"  not in** data:  
        data\[**"salt"**\] =  unicode(random.random()).decode(**"ascii"**)  
    **if "expires"not in** data:  
        data\[**"expires"**\] = time.time() + TIMEOUT  
    payload = json.dumps(data).encode(**"utf8"**)  
    *\# 生成签名  
   * sig =\_get\_signature(payload)  
    **return** encode\_token\_bytes(payload + sig)  
  
*\# 授权码生成器*  
**def** gen\_auth\_code(uri, user\_id):  
    code = random.randint(0,10000)  
    auth\_code\[code\] = \[uri, user\_id\]  
    **return** code  
  
*\# 新版本的token验证*  
**def** verify\_token(token):  
    *'''  
   * **:param** *token: base64 str  
   * **:return***: dict type  
    '''  
   * decoded\_token =decode\_token\_bytes(str(token))  
    payload = decoded\_token\[:-16\]  
    sig = decoded\_token\[-16:\]  
    *\# 生成签名  
   * expected\_sig =\_get\_signature(payload)  
    **if** sig != expected\_sig:  
        **return** \{\}  
    data = json.loads(payload.decode(**"utf8"**))  
    **if** data.get(**'expires'**) >=time.time():  
        **return** data  
    **return** 0  
  
*\# 使用hmac为消息生成签名*  
**def** \_get\_signature(value):  
    *"""Calculatethe HMAC signature for the given value."""  
   * **return** hmac.new(**'secret123456'**, value).digest()  
  
*\# 下面两个函数将base64编码和解码单独封装*  
**def** encode\_token\_bytes(data):  
    **return** base64.urlsafe\_b64encode(data)  
  
**def** decode\_token\_bytes(data):  
    **return** base64.urlsafe\_b64decode(data)  
  
  
*\# 验证服务器端*  
@app.route(**'/index'**,methods=\[**'POST'**,**'GET'**\])  
**def** index():  
    **return 'hello'**  
  
@app.route(**'/login'**,methods=\[**'POST'**,**'GET'**\])  
**def** login():  
    uid, pw =base64.b64decode(request.headers\[**'Authorization'**\].split(**' '**)\[-1\]).split(**':'**)  
    **if** users.get(uid)\[0\] == pw:  
        **return** gen\_token(dict(user=uid,pw=pw))  
    **else**:  
        **return 'error'**  
  
@app.route(**'/oauth'**,methods=\[**'POST'**,**'GET'**\])  
**def** oauth():  
    *\# 处理表单登录， 同时设置Cookie  
   * **if** request.method ==**'POST' and** request.form\[**'user'**\]:  
        u = request.form\[**'user'**\]  
        p = request.form\[**'pw'**\]  
        **if** users.get(u)\[0\] == p**and** oauth\_redirect\_uri:  
            uri = oauth\_redirect\_uri\[0\] + **'?code=%s'** % gen\_auth\_code(oauth\_redirect\_uri\[0\], u)  
            expire\_date = datetime.now()+ timedelta(minutes=1)  
            resp =make\_response(redirect(uri))  
            resp.set\_cookie(**'login'**,**'\_'**.join(\[u, p\]), expires=expire\_date)  
            **return** resp  
    *\# 验证授权码，发放token  
   * **if** request.args.get(**'code'**):  
        auth\_info = auth\_code.get(int(request.args.get(**'code'**)))  
        **if** auth\_info\[0\] ==request.args.get(**'redirect\_uri'**):  
            *\# 可以在授权码的auth\_code中存储用户名，编进token中  
           * **return** gen\_token(dict(client\_id=request.args.get(**'client\_id'**),user\_id=auth\_info\[1\]))  
    *\# 如果登录用户有Cookie，则直接验证成功，否则需要填写登录表单  
   * **if** request.args.get(**'redirect\_uri'**):  
       oauth\_redirect\_uri.append(request.args.get(**'redirect\_uri'**))  
        **if** request.cookies.get(**'login'**):  
            u, p = request.cookies.get(**'login'**).split(**'\_'**)  
            **if** users.get(u)\[0\] == p:  
                uri = oauth\_redirect\_uri\[0\] + **'?code=%s'** % gen\_auth\_code(oauth\_redirect\_uri\[0\], u)  
                **return** redirect(uri)  
        **return '''  
            <form action=""method="post">  
                <p><inputtype=text name=user>  
                <p><inputtype=text name=pw>  
                <p><inputtype=submit value=Login>  
            </form>  
        '''**  
  
  
*\# 客户端*  
@app.route(**'/client/login'**,methods=\[**'POST'**,**'GET'**\])  
**def** client\_login():  
    uri = **'http://localhost:5000/oauth?response\_type=code&client\_id=%s&redirect\_uri=%s'**% (client\_id, redirect\_uri)  
    **return** redirect(uri)  
  
@app.route(**'/client/passport'**,methods=\[**'POST'**,**'GET'**\])  
**def** client\_passport():  
    code = request.args.get(**'code'**)  
    uri = **'http://localhost:5000/oauth?grant\_type=authorization\_code&code=%s&redirect\_uri=%s&client\_id=%s'**% (code, redirect\_uri, client\_id)  
    **return** redirect(uri)  
  
  
*\# 资源服务器端*  
@app.route(**'/test1'**,methods=\[**'POST'**,**'GET'**\])  
**def** test():  
    token = request.args.get(**'token'**)  
    ret = verify\_token(token)  
    **if** ret:  
        **return** json.dumps(ret)  
    **else**:  
        **return 'error'  
  
if** \_\_name\_\_ == **'\_\_main\_\_'**:  
    app.run(debug=True)

**5、Flask-RESTful插件介绍及应用**

5-1 Flask-RESTful插件介绍

Flask-RESTful概述

Flask-RESTful 是一个Flask扩展，它添加了快速构建REST APIs的支持。它当然也是一个能够跟你现有的ORM/库协同工作的轻量级的扩展。Flask-RESTful鼓励以最小设置的最佳实践。如果你熟悉 Flask 的话，Flask-RESTful 应该很容易上手。

安装：>pip install flask-restful

资源路由

Flask-restful不在使用装饰器作为路由，处理请求的部分是类，而不是flask的函数

可插拔视图：

[http://dormousehole.readthedocs.org/en/latest/views.html][http_dormousehole.readthedocs.org_en_latest_views.html]

代码：

from flask import Flask  
from flask.ext import restful  
  
app = Flask(\_\_name\_\_)  
api = restful.Api(app)  
  
class HelloWorld(restful.Resource):  
    def get(self):  
        return \{'hello': 'world'\}  
  
api.add\_resource(HelloWorld, '/')  
  
if \_\_name\_\_ == '\_\_main\_\_':  
    app.run(debug=True)  
   
  
  
参数解析  
  
  
代码：  
  
  
\# 1. 关于参数解析的部分  
\# 一组虚拟的数据  
TODOS = \{  
    'todo1': \{'task': 1\},  
    'todo2': \{'task': 2\},  
    'todo3': \{'task': 3\},  
\}  
  
  
\# 定义允许的参数为task，类型为int，以及错误时的提示  
parser = reqparse.RequestParser()  
parser.add\_argument('task', type=int, help='Please set a int task content!')  
  
  
\# 真正处理请求的地方  
class TodoList(Resource):  
    def get(self):  
        return TODOS, 200, \{'Etag': 'some-opaque-string'\}  
  
  
    def post(self):  
        args = parser.parse\_args()  
        todo\_id = int(max(TODOS.keys()).lstrip('todo')) + 1  
        todo\_id = 'todo%i' % todo\_id  
        TODOS\[todo\_id\] = \{'task': args\['task'\]\}  
        return TODOS\[todo\_id\], 201  
  
  
\# 实际定义路由的地方  
api.add\_resource(TodoList, '/todos', '/all\_tasks')

Linux下测试：

% curl http://localhost:5000/todos -d “task=4” –X POST

% curl http://localhost:5000/todos

响应域

代码：

\# 2. 关于响应域的部分  
  
  
\# ORM的数据模型  
class TodoDao(object):  
    def \_\_init\_\_(self, todo\_id, task):  
        self.todo\_id = todo\_id  
        self.task = task  
  
  
        \# 这个域不会被返回  
        self.status = 'active'  
  
  
\# marshal-蒙版  
resource\_fields = \{  
    'task':   fields.String,  
    'uri':    fields.Url('todo\_ep')  
\}  
  
  
\# 真正处理请求的地方  
class Todo(Resource):  
    \# 蒙版  
    @marshal\_with(resource\_fields)  
    def get(self, todo\_id):  
        return TodoDao(todo\_id=todo\_id, task='Remember the milk'), 200  
  
  
\# 实际定义路由的地方  
api.add\_resource(Todo, '/todos/<todo\_id>', endpoint='todo\_ep')

Linux下测试：

% curl http://localhost:5000/todos/4

结果：

"task": "Remember themilk",

"uri": "/todos/4"

5-2 Flask-RESTful请求解析

知识点：

必选参数与多值参数的设定

指定获取参数的位置

代码：

from flask import Flask  
from flask.ext.restful importreqparse, Api,Resource  
  
  
app = Flask(\_\_name\_\_)  
api = Api(app)  
  
  
USERS = \{  
    'row1': \{'name':'jilu', 'rate': \[70,65\]\},  
    'row2': \{'name':'bob', 'rate': \[80,90, 68\]\},  
    'row3': \{'name':'tim', 'rate': \[90,80\]\},  
\}  
  
  
parser = reqparse.RequestParser()  
parser.add\_argument('name', type=str, required=True)  //必选参数  
parser.add\_argument('rate', type=int, help='rate is a number',action='append')  
parser.add\_argument('User-Agent',type=str, location='headers') //指定获取参数的位置  
  
  
  
  
  
  
\# 真正处理请求的地方  
class UserInfo(Resource):  
    def get(self):  
        return USERS, 200  
  
  
    def post(self):  
        args = parser.parse\_args()  
        user\_id = int(max(USERS.keys()).lstrip('row')) + 1  
        user\_id = 'row%i' % user\_id  
        USERS\[user\_id\] = \{'name': args\['name'\], 'rate': args\['rate'\]\}  
        USERS\[user\_id\]\['ua'\] = args.get('User-Agent')  
        return USERS\[user\_id\], 201  
  
  
api.add\_resource(UserInfo, '/')  
  
  
if \_\_name\_\_ == '\_\_main\_\_':  
    app.run(debug=True)  
   
  
  
参数解析器的继承  
  
  
代码：  
  
  
\# 解析器继承  
parser\_copy = parser.copy()  
parser\_copy.add\_argument('bar', type=int)  
parser\_copy.replace\_argument('bar', type=str, required=True) // 覆盖原先已经设定好的参数  
parser\_copy.remove\_argument('User-Agent') //删除原先已有的参数

5-3 Flask-RESTful的响应域

Flask-RESTful提供的一种控制响应对象的方法，可以和各种ORM结合，过滤掉不想对用户暴露的内部数据结构。

知识点：

重命名域名和设置默认域名

定制响应域的字段

生成复杂、嵌套结构的响应

代码：

import json

from datetime import datetime

from flask.ext.restful import Resource,fields, marshal\_with, marshal

\# 基本例子

resource\_fields = \{

'name': fields.String,

'address': fields.String,

'date\_updated': fields.DateTime(dt\_format='rfc822'),

class UserInfo(object):

def \_\_init\_\_(self, name, address, date\_updated=datetime.now()):

self.name = name

self.address = address

self.date\_updated = date\_updated

print json.dumps(**marshal**(UserInfo('magi', 'beijing'),resource\_fields))**//使用marshal()而不使用marshal\_with()**

\# class Todo(Resource):

\#    @marshal\_with(resource\_fields, envelope='resource')

\#    def get(self, \*\*kwargs):

\#        return UserInfo('magi', 'beijing')

\# 输出域别名

resource\_fields2 = \{

'open\_name': fields.String(**attribute='name'**),

'address': fields.String,

'date\_updated': fields.DateTime(dt\_format='rfc822'),

print json.dumps(marshal(UserInfo('magi','beijing'), resource\_fields2))

\# 输出域默认值

class UserInfo2(object):

def \_\_init\_\_(self, address):

self.address = address

resource\_fields3 = \{

'open\_name': fields.String(**default='add\_magi'**),

'address': fields.String,

'date\_updated': fields.DateTime(dt\_format='rfc822', **default=str(datetime.now())**),

printjson.dumps(marshal(UserInfo2(address='beijing'), resource\_fields3))

\# 自定义输出域

class UserInfo2(object):

def \_\_init\_\_(self, address, flag, date\_updated=datetime.now()):

self.address = address

self.date\_updated = date\_updated

self.flags = flag

class UrgentItem**(fields.Raw)**:

def format(self, value):

return "Urgent" if value & 0x01 else "Normal"

resource\_fields4 = \{

'open\_name': fields.String(default='add\_magi'),

'address': fields.String,

'date\_updated': fields.DateTime(dt\_format='rfc822'),

'priority': UrgentItem(attribute='flags'),

printjson.dumps(marshal(UserInfo2(address='beijing', flag=1), resource\_fields4))

\# 复杂结构

resource\_fields = \{'name': fields.String\}

resource\_fields\['address'\] = **\{\}**

resource\_fields\['address'\]\['line 1'\] =fields.String(attribute='addr1')

resource\_fields\['address'\]\['line 2'\] =fields.String(attribute='addr2')

resource\_fields\['address'\]\['city'\] =fields.String

resource\_fields\['address'\]\['state'\] =fields.String

resource\_fields\['address'\]\['zip'\] =fields.String

data = \{'name': 'bob', 'addr1': '123 fakestreet', 'addr2': '', 'city': 'New York', 'state': 'NY', 'zip': '10468'\}

print json.dumps(marshal(data,resource\_fields))

\# 嵌套域

address\_fields = \{\}

address\_fields\['line 1'\] =fields.String(attribute='addr1')

address\_fields\['line 2'\] =fields.String(attribute='addr2')

address\_fields\['city'\] =fields.String(attribute='city')

address\_fields\['state'\] =fields.String(attribute='state')

address\_fields\['zip'\] =fields.String(attribute='zip')

resource\_fields = \{\}

resource\_fields\['name'\] = fields.String

resource\_fields\['billing\_address'\] = **fields.Nested**(address\_fields)

resource\_fields\['shipping\_address'\] =fields.Nested(address\_fields)

address1 = \{'addr1': '123 fake street','city': 'New York', 'state': 'NY', 'zip': '10468'\}

address2 = \{'addr1': '555 nowhere', 'city':'New York', 'state': 'NY', 'zip': '10468'\}

data = \{ 'name': 'bob', 'billing\_address':address1, 'shipping\_address': address2\}

print json.dumps(marshal(data,resource\_fields))

5-4 重构程序

对资源服务器进行：

添加数据模型

使用marshal

将flask函数替换为restful形式的类

**from** flask.ext **import** restful  
**from** flask.ext.restful **import**fields,marshal\_with  
  
app = Flask(\_\_name\_\_)  
api = restful.Api(app)

*\# 资源服务器端  
\# 数据模型*  
**class** Test1Data(object):  
    **def** \_\_init\_\_(self, client\_id, expires, salt, user\_id):  
        self.client\_id = client\_id  
        self.expires = expires  
        self.salt = salt  
        self.user\_id = user\_id  
  
*\# marshal-蒙版*  
resource\_fields = \{  
    **'client\_id'**: fields.String(default=**''**),  
    **'expires'**: fields.Float(default=0.0),  
    *\#'salt':fields.Float(default=0.0),  
   * **'user\_id'**: fields.String(default=**''**),  
    *\#'date':fields.DateTime(default=str(datetime.now()))*  
  
\}  
  
*\# 新的资源服务器*  
**class** Test1(restful.Resource):  
   @marshal\_with(resource\_fields)  
    **def** get(self):  
        token = request.args.get(**'token'**)  
        ret = verify\_token(token)  
        **if** ret:  
            **return** ret  
        **else**:  
            **return 'error'**  
  
api.add\_resource(Test1, **'/test1'**)

知识点总结：

掌握Flask-RESTful的基本使用

掌握Flask-RESTful的如何解析请求

掌握Flask-RESTful的如何处理响应域

**6、HTTPs以及Flask-OAuthlib插件使用**

1 HTTPs介绍及搭建 Flask HTTPs 环境

HTTPs介绍

HTTPS（全称：Hyper Text Transfer Protocol over Secure Socket Layer），是以**安全**为目标的[HTTP][]通道，简单讲是HTTP的安全版。是超文本传输协议和SSL/TLS的组合，用以提供加密通讯及对网络服务器身份的鉴定。HTTPS连接经常被用于万维网上的交易支付和企业信息系统中敏感信息的传输。这个系统的最初研发由[网景][Link 1]公司(Netscape)进行，并内置于其浏览器NetscapeNavigator中，提供了身份验证与加密[通讯][Link 2]方法。

HTTP访问

HTTPs访问

证书

HTTPs握手过程

在Linux或mac下，搭建Flask HTTPs开发环境

需要的软件：

OpenSSL

stunnel

$ openssl req -new -x509 -days 365 -nodes-out magigo.pem -keyout magigo.pem

$ chmod 600 magigo.pem

$ vim https\_st

代码：

1 pid =

2cert = /home/yuxiang/magigo.pem

3debug = 7

4foreground = yes

6\[https\]

7accept = 443

8connect = 5000

$ sudo stunnel https\_st

然后启动程序，输入**https**://localhost:5000/client/login，会出现安全连接问题

2 使用Flask-OAuthlib插件搭建OAuth 2Server

Flask-OAuthlib的基本使用

[https://flask-oauthlib.readthedocs.org/en/latest/][https_flask-oauthlib.readthedocs.org_en_latest]

安装：

pipinstall Flask-OAuthlib

原理：

\#client.py – 客户端（8000）

1）  将用户重定向到oauth授权服务器

2）  使用得到的授权码去换取token

\#app.py - 授权服务器（5000）

使用Flask-OAuthlib搭建OAuth2 Server

代码：

\# client.py

*\# coding: utf-8*  
  
from flask import Flask, url\_for, session, request,jsonify  
from flask\_oauthlib.client import OAuth  
  
  
**CLIENT\_ID**= 'LyofOAKrBZnSW5GQlp7xcg9DtbgK8lo6p641lY8t'  
**CLIENT\_SECRET** ='uCysaCWYh4aGUPIE19zMstcom9kYVUz9oXIrNwmMuyU1Y6hKl6'  
  
  
app = Flask(\_\_name\_\_)  
app.debug = True  
app.secret\_key = 'secret'  
oauth = OAuth(app)  
  
**remote**= oauth.remote\_app(  
    'remote',  
    consumer\_key=CLIENT\_ID,  
    consumer\_secret=CLIENT\_SECRET,  
    request\_token\_params=\{'scope':'email'\},  
   base\_url='http://127.0.0.1:5000/api/',  
    request\_token\_url=None,  
   access\_token\_url='http://127.0.0.1:5000/oauth/token',  
   authorize\_url='http://127.0.0.1:5000/oauth/authorize'  
)  
  
\# 相当于/client/login，用于重定向用户登录  
@app.route('/')  
def index():  
    if 'remote\_oauth' in session:  
        resp = remote.get('me')  
        return jsonify(resp.data)  
    next\_url = request.args.get('next')or request.referrer or None  
    return remote.authorize(  
        callback=url\_for('authorized',next=next\_url, \_external=True)  
    )  
  
\# 相当于/client/passport，用于获取**token，并存储在Session中**  
@app.route('/authorized')  
def authorized():  
    resp = remote.authorized\_response()  
    if resp is None:  
        return 'Access denied: reason=%serror=%s' % (  
            request.args\['error\_reason'\],  
           request.args\['error\_description'\]  
        )  
    print resp  
    session\['remote\_oauth'\] =(resp\['access\_token'\], '')  
    returnjsonify(oauth\_token=resp\['access\_token'\])  
  
  
@remote.tokengetter  
def get\_oauth\_token():  
    return session.get('remote\_oauth')  
  
  
if \_\_name\_\_ == '\_\_main\_\_':  
    import os  
    os.environ\['DEBUG'\] = 'true'  
   os.environ\['OAUTHLIB\_INSECURE\_TRANSPORT'\] = 'true'  
    app.run(host='localhost', port=8000)

\# app.py

*\# coding: utf-8*  
  
**from** datetime **import** datetime,timedelta  
  
**from** flask **import**Flask  
**from** flask **import**session, request  
**from** flask **import**render\_template, redirect, jsonify  
**from** flask\_sqlalchemy **importSQLAlchemy  
from** werkzeug.security **import**gen\_salt  
**from** flask\_**oauthlib**.provider**import** OAuth2Provider  
**from** flask.ext **import**restful  
**from** flask.ext.restful **import**fields,marshal\_with  
  
app = Flask(\_\_name\_\_, template\_folder=**'templates'**)  
app.debug = True  
app.secret\_key = **'secret'**  
app.config.update(\{  
    **'SQLALCHEMY\_DATABASE\_URI'**: **'sqlite:///db.sqlite'**,  
\})  
db = SQLAlchemy(app)  
oauth = OAuth2Provider(app)  
  
*\# 相当于/*  
@app.route(**'/'**,methods=(**'GET'**,**'POST'**))  
**def** home():  
    **if** request.method == **'POST'**:  
        username = request.form.get(**'username'**)  
        user = User.query.filter\_by(username=username).first()  
        **if not** user:  
            user = User(username=username)  
            db.session.add(user)  
            db.session.commit()  
        session\[**'id'**\] = user.id  
        **return** redirect(**'/'**)  
    user = current\_user()  
    **return** render\_template(**'home.html'**,user=user)  
  
  
**@app.route('/client')**  // 生成client\_id 和 client\_secret  
**def** client():  
    *'''  
    为登录用户注册一个新的客户端  
   * **:return***:  
    '''  
   * user =current\_user()  
    **if not** user:  
        **return** redirect(**'/'**)  
    item = Client(  
        client\_id=gen\_salt(40),  
        client\_secret=gen\_salt(50),  
        \_redirect\_uris=**' '**.join(\[  
            **'http://localhost:8000/authorized'**,  
            **'http://127.0.0.1:8000/authorized'**,  
            **'http://127.0.1:8000/authorized'**,  
            **'http://127.1:8000/authorized'**,  
            \]),  
        \_default\_scopes=**'email'**,  
        user\_id=user.id,  
    )  
    db.session.add(item)  
    db.session.commit()  
    **return** jsonify(  
        client\_id=item.client\_id,  
        client\_secret=item.client\_secret,  
    )

*\# 相当于oauth*  
@app.route(**'/oauth/token'**,methods=\[**'GET'**,**'POST'**\])  
@oauth.token\_handler  
**def** access\_token():  
    **return** None  
  
*\# 相当于login*  
@app.route(**'/oauth/authorize'**,methods=\[**'GET'**,**'POST'**\])  
@oauth.authorize\_handler  
**def** authorize(\*args, \*\*kwargs):  
    user = current\_user()  
    **if not** user:  
        **return** redirect(**'/'**)  
    **if** request.method == **'GET'**:  
        client\_id = kwargs.get(**'client\_id'**)  
        client = Client.query.filter\_by(client\_id=client\_id).first()  
        kwargs\[**'client'**\] = client  
        kwargs\[**'user'**\] = user  
        **return** render\_template(**'authorize.html'**, \*\*kwargs)  
  
    confirm = request.form.get(**'confirm'**,**'no'**)  
    **return** confirm == **'yes'**

\# 老式的资源服务器的写法

@app.route(‘/api/me’)

@oauth.require\_oauth()

def me():

user = request.oauth.user

return jsonify(username=user.username)

**if** \_\_name\_\_ == **'\_\_main\_\_'**:  
    db.create\_all()  
    app.run()

3 使用Flask-OAuthlib提供的装饰器保护资源服务器

ORM - 对象关系映射 – 其实是创建了一个可在编程语言使用的“虚拟对象数据库”

属性装饰器

**@property**

代码：

1）

*\# 存储用户信息的ORM*  
**class** User(db.Model):  
    id = db.Column(db.Integer, primary\_key=True)  
    username = db.Column(db.String(40),  unique=True)  
  
*\# 存储客户端信息的ORM*  
**class** Client(db.Model):  
    client\_id = db.Column(db.String(40),  primary\_key=True)  
    client\_secret = db.Column(db.String(55),  nullable=False)  
  
   ** user\_id = db.Column(db.ForeignKey('user.id'))**  
    user = db.relationship(**'User'**)  
  
    \_redirect\_uris = db.Column(db.Text)  
    \_default\_scopes = db.Column(db.Text)  
  
    **@property**  
    **def** client\_type(self):  
        **return 'pub lic'  
  
   ** @property  
    **def** redirect\_uris(self):  
        **if** self.\_redirect\_uris:  
            **return** self.\_redirect\_uris.split()  
        **return** \[\]  
  
    @property  
    **def** default\_redirect\_uri(self):  
        **return** self.redirect\_uris\[0\]  
  
    @property  
    **def** default\_scopes(self):  
        **if** self.\_default\_scopes:  
            **return** self.\_default\_scopes.split()  
        **return** \[\]  
  
*\# 存储授权码信息的ORM*  
**class** Grant(db.Model):  
    id = db.Column(db.Integer, primary\_key=True)  
  
    user\_id = db.Column(  
        db.Integer, db.ForeignKey(**'user.id'**,ondelete=**'CASCADE'**)  
    )  
    user = db.relationship(**'User'**)  
  
    client\_id = db.Column(  
        db.String(40), db.ForeignKey(**'client.client\_id'**),  
        nullable=False,  
    )  
    client = db.relationship(**'Client'**)  
  
    code = db.Column(db.String(255),  index=True, nullable=False)  
  
    redirect\_uri = db.Column(db.String(255))  
    expires = db.Column(db.DateTime)  
  
    \_scopes = db.Column(db.Text)  
  
    **def** delete(self):  
        db.session.delete(self)  
        db.session.commit()  
        **return** self  
  
    @property  
    **def** scopes(self):  
        **if** self.\_scopes:  
            **return** self.\_scopes.split()  
        **return** \[\]  
  
*\# 存储token信息的ORM*  
**class** Token(db.Model):  
    id = db.Column(db.Integer, primary\_key=True)  
    client\_id = db.Column(  
        db.String(40), db.ForeignKey(**'client.client\_id'**),  
        nullable=False,  
    )  
    client = db.relationship(**'Client'**)  
  
    user\_id = db.Column(  
        db.Integer, db.ForeignKey(**'user.id'**)  
    )  
    user = db.relationship(**'User'**)  
  
    *\# currently onlybearer is supported  
   * token\_type =db.Column(db.String(40))  
  
    access\_token = db.Column(db.String(255),  unique=True)  
    refresh\_token = db.Column(db.String(255),  unique=True)  
    expires = db.Column(db.DateTime)  
    \_scopes = db.Column(db.Text)  
  
    @property  
    **def** scopes(self):  
        **if** self.\_scopes:  
            **return** self.\_scopes.split()  
        **return** \[\]

2）

@oauth.clientgetter  
**def** load\_client(client\_id):  
    **return** Client.query.filter\_by(client\_id=client\_id).first()  
  
  
@oauth.grantgetter  
**def** load\_grant(client\_id, code):  
    **return** Grant.query.filter\_by(client\_id=client\_id,code=code).first()  
  
  
@oauth.grantsetter  
**def** save\_grant(client\_id, code, request, \*args, \*\*kwargs):  
    *\# decide theexpires time yourself  
   * expires =datetime.utcnow() + timedelta(seconds=100)  
    grant = Grant(  
        client\_id=client\_id,  
        code=code\[**'code'**\],  
        redirect\_uri=request.redirect\_uri,  
        \_scopes=**' '**.join(request.scopes),  
        user=current\_user(),  
        expires=expires  
    )  
    db.session.add(grant)  
    db.session.commit()  
    **return** grant  
  
  
@oauth.tokengetter  
**def** load\_token(access\_token=None, refresh\_token=None):  
    **if** access\_token:  
        **return** Token.query.filter\_by(access\_token=access\_token).first()  
    **elif** refresh\_token:  
        **return** Token.query.filter\_by(refresh\_token=refresh\_token).first()  
  
  
@oauth.tokensetter  
**def** save\_token(token, request, \*args, \*\*kwargs):  
    toks = Token.query.filter\_by(  
        client\_id=request.client.client\_id,  
        user\_id=request.user.id  
    )  
    *\# make sure thatevery client has only one token connected to a user  
   * **for** t **in**toks:  
        db.session.delete(t)  
  
    expires\_in = token.pop(**'expires\_in'**)  
    expires = datetime.utcnow() +timedelta(seconds=expires\_in)  
  
    tok = Token(  
        access\_token=token\[**'access\_token'**\],  
        refresh\_token=token\[**'refresh\_token'**\],  
        token\_type=token\[**'token\_type'**\],  
        \_scopes=token\[**'scope'**\],  
        expires=expires,  
        client\_id=request.client.client\_id,  
        user\_id=request.user.id,  
    )  
    db.session.add(tok)  
    db.session.commit()  
    **return** tok

3）

**def** current\_user():  
    **if 'id'  in** session:  
        uid = session\[**'id'**\]  
        **return** User.query.get(uid)  
    **return** None

4 重构程序

代码：

*\# 新的资源服务器*  
api = restful.Api(app, decorators=\[oauth.require\_oauth(**'email'**)\])  
  
resource\_fields = \{  
    **'username'**: fields.String(),  
    **'date'**: fields.DateTime(default=str(datetime.now())),  
    **'id'**: fields.Integer()  
\}  
  
**class** ApiMe(restful.Resource):  
    @marshal\_with(resource\_fields)  
    **def** get(self):  
        user = request.oauth.user  
        **return** user  
  
api.add\_resource(ApiMe, **'/api/me'**)

总结：

掌握HTTPs的基本概念以及搭建Flask开发环境

掌握使用Flask-OAuthlib搭建OAuth2 Server

掌握使用Flask-OAuthlib保护资源服务器

[http_requests-docs-cn.readthedocs.org_zh_CN_latest_user_quickstart.html]: http://requests-docs-cn.readthedocs.org/zh_CN/latest/user/quickstart.html
[http_dormousehole.readthedocs.org_en_latest_quickstart.html_sessions]: http://dormousehole.readthedocs.org/en/latest/quickstart.html#sessions
[http_dormousehole.readthedocs.org_en_latest_views.html]: http://dormousehole.readthedocs.org/en/latest/views.html
[HTTP]: http://baike.baidu.com/view/9472.htm
[Link 1]: http://baike.baidu.com/view/478896.htm
[Link 2]: http://baike.baidu.com/view/20429.htm
[https_flask-oauthlib.readthedocs.org_en_latest]: https://flask-oauthlib.readthedocs.org/en/latest/