手把手教你用Python和Flask创建REST API

在我们之前的一篇文章中:使用 Flask API 将机器学习模型部署到生产环境中，我们学习了 Flask 的基础知识以及如何设置它。我们用 Flask 做了一个完整的端端机器学习项目。今天云朵君和大家一起学习使用更好的 REST API 创建 Flask API。本文目的是为了在实际工作中，可能存在要求数据挖掘工程师或算法工程师将你的机器学习模型封装，并提供API供其他软件工程师调用。

REST API 简介

注意，这里只是对REST API 的简单介绍，起到抛砖引玉作用，更加深入内容不在本次学习范围内，感兴趣的小伙伴可以查看相关资料深入学习。此外本号接受该领域的投稿，欢迎联系云朵君！

API

API，全名Application Programming Interface (应用程式介面)，简单来说，是品牌开发出的一种接口，让第三方可以额外开发、应用在自身的产品上的系统沟通介面。

REST API

REST 是一种通过 HTTP 协议设计 API 的架构风格。它的主要优点是其极大的灵活性。只要需要直接从服务器向 Web 应用程序或站点的用户提供数据，开发人员就会使用 REST API。

REST API 的主要组件：

客户 — 在用户端（在他的设备上）启动的客户端或程序启动通信。
服务器 — 使用 API 访问其功能和数据的服务器。
资源 — 服务器传输给客户端的任何内容（视频、文本、图片）。

REST API 通过 HTTP 请求进行通信，完成以下功能——创建、读取、更新和删除数据。它们也称为 CRUD 操作。REST 提供有关请求资源的信息，并使用四种方法来描述如何处理资源：

POST — 创建资源；
GET — 获取资源；
PUT — 更新资源；
DELETE — 删除资源。

RESTful API

REST，全名Representational State Transfer( 表现层状态转移)，他是一种设计风格，RESTful 只是转为形容词，像是 peace 和平这名词，转成形容词是peaceful，RESTful 则形容以此规范设计的API，称为RESTful API。

RESTful API 主要由三种元件组成：

Nouns 名词：定义资源位置的URL，每个资源在网路上都会有唯一的位置，就如每户人家都有唯一的地址一样。
Verbs 动词：对资源要做的动作。
Content Types 资源呈现方式：API 资源可以以多种方式表现，最常用的是JSON，轻量易处理。

所以使用 RESTful 风格设计的API，就有了以下几种优点及限制：

有唯一的URL表示资源位置，统一的API 接口。(Uniform Interface)
无状态。(Stateless)

Restful API 它允许集成应用程序 app 或与 Restful Web 服务交互。它现在正在成长为连接微服务架构中组件的最常用方法。我们借助 API，能够获取或发送数据到网站并执行一些操作，目的是通过 Web 服务完成我们的任务。每个网站都使用不同类型的 API，例如股票市场交易网站，借助 API 获取当前价格和上下波动。

创建第一个 REST API

同样，我们创建 Hello world API，它表示如果你对其发出 get 请求，将获得 JSON 响应，一般情况下， API 给出 JSON 类型的响应。接下来，使用 pip 包管理器安装 Flask：

pip install flask

pip install flask-restful



from flask import Flask

from flask_restful import Resource, Api

app = Flask(__name__)

api = Api(app)



class Helloworld(Resource):

  def __init__(self):

    pass

  def get(self):

    return {

      "Hello": "World"

        }

api.add_resource(Helloworld, '/')



if __name__ == '__main__':

 app.run(debug=True)

ok，到现在已经创建了第一个Rest api，看起来挺简单的，那么，什么是Flask-Restful呢？

Flask restful 定义了资源Resource类，其中包含每个 HTTP 方法的方法。方法名称应与其对应的 HTTP 方法相同，并以小写形式书写，如上代码中所示。我们发现这些方法没有路由装饰器，这是它们是基于资源路由的。无论定义什么类，我们都可以使用添加资源add_resource方法定义到它的路由以及在对应路由上调用该类。

说明： 在上面的代码中，我们首先加载需要的父类，然后初始化我们的app和API。之后，我们创建了一个程序，并且我们正在发出一个 GET 请求，说明如果有人点击了这个程序，那么他将得到 Hello world 作为 JSON 格式的响应。要打开特定 URL，我们使用 add resource 方法并将其路由到默认斜杠。要运行此文件，可以使用 POSTMAN 工具（一种 API 维护工具）来创建、测试和管理 API。还可以使用requests请求模块使用以下代码测试此 API。首先，运行上面的文件，它会给你 localhost URL，然后在另一个命令提示符下，运行下面的代码文件：

import requests

url = "http://127.0.0.1:5000/"

response = requests.get(url=url)

print(response.text)

{

"Hello": "World"

}

通过 Flask RESTApi 理解 HTTP 请求

通过以上内容的学习，相比大家已经对 REST API 有个初步印象。接下来我们将继续探索使用 REST API 的不同 HTTP 方法，其中我们定义一个列表，该列表将以字典（JSON 对象）的形式存储从服务器获取的所有数据。这是很重要的，因为我们在项目中有不同的api来获取数据，而不是其他地方的数据。

首先创建一个 API，在其中创建 3 个名为 GET、POST 和 DELETE 的 HTTP 方法，并且在其中创建一个自定义 URL，当请求 POST 方法时，它将以 Name 作为输入；在请求 GET 方法时，将名称返回；在DELETE时，如果该名称存在，我们将删除该名称，再次访问它会给我们 NULL。

创建一个文件并编写以下代码：

from flask import Flask

from flask_restful import Resource, Api

app = Flask(__name__)

api = Api(app)

data = []

class People(Resource):

   def get(self):

      for x in data:

          if x['Data'] == name:

              return x

      return {'Data': None}

 def post(self, name):

    temp = {'Data': name}

    data.append(temp)

    return temp

 def delete(self):

    for ind, x in enumerate(data):

       if x['Data'] == name:

           temp = data.pop(ind)

       return {'Note': 'Deleted'}

api.add_resource(People, '/Name/')

if __name__ == '__main__':

 app.run(debug=True)

打开 POSTMAN API 工具并点击每个 HTTP 方法请求。首先，当我们使用 post 请求Name时，它给了我们一个name。在获取请求时，我们将返回name。它在删除时被删除，当再次将其取回时，它会给你返回 NULL。

结果如下

如何在 Flask REST API 中使用装饰器

我们使用带有 API 的装饰器来监控 IP 地址、cookie 等。我们将继续学习如何使用带有装饰器的Flask API。装饰器是一个将另一个函数作为参数并返回另一个函数的函数。也可以将其理解为在不改变或修改当前功能的情况下，为现有功能提供一些附加功能的功能。

这里我们创建一个新文件，我将通过创建两个装饰器来展示。在第一个文件中，编写返回代码执行时间的外部时间函数。我们从 functools 模块（用于高阶 python 函数的标准模块）导入应用于 wrapper 函数的 wrap 装饰器。它通过复制所有参数来更新包装函数。

from flask import Flask

from flask_restful import Resource, Api

import datetime

from flask import request

from functools import wraps

app = Flask(__name__)

api = Api(app)



def time(function=None):

    @wraps(function)

    def wrapper(*args, **kwargs):

        s = datetime.datetime.now()

        _ = function(*args, **kwargs)

        e = datetime.datetime.now()

        print("Execution Time : {} ".format(e-s))

        return _

    return wrapper

class HelloWorld(Resource):

    @monitor

    def get(self):

        return {"hello": "world"}



api.add_resource(HelloWorld, "/")

if __name__ == "__main__":

    app.run(debug=True)

我们创建第二个装饰器来监视cookie和IP地址，因此创建下面的函数。不是向hello world函数添加时间装饰器，而是添加监视器装饰器并运行代码。

def monitor(function=None):

    @wraps(function)

    def wrapper(*args, **kwargs):

        _ = function(*args, **kwargs)

        print("Ip Address  : {} ".format(request.remote_user))

        print("Cookies : {} ".format(request.cookies))

        print(request.user_agent)

        return _

    return wrapper

如何使 Flask API 更安全

当我们设计API时，我们也应该注意安全性，因为很多人会访问它。因为API可能包含一些双方之间的机密数据，因此我们可以指定只有授权的人可以访问API，那该怎么办？此时可以使用Flask基本身份验证。当然，此时需要使用pip命令安装这个flask模块。

pip install flask-httpauth

我们正在构建一个API并定义User数据字典，其中包含用户名和密码。当在实时用例中工作时，可以通过配置文件或从数据库中接受用户名和密码。首先，我们创建一个主要函数来匹配用户名和密码，并创建一个GET方法，该方法表示任何点击此API的人，如果没有登录，我们就无法访问数据。

from flask import Flask

from flask_restful import Resource, Api

from flask_httpauth import HTTPBasicAuth

app = Flask(__name__)

api = Api(app, prefix="/api/v1")

auth = HTTPBasicAuth()

USER_DATA = {

 "admin": "SuperSecretPwd"

}

#route to verify the password

@auth.verify_password

def verify(username, password):

   if not(username and password):

      return False

   return USER_DATA.get(username) == password

class PrivateResource(Resource):

   @auth.login_required

   def get(self):

      return {"How's the Josh": "High"}

api.add_resource(PrivateResource, '/private')

if __name__ == '__main__':

   app.run(debug=True)

当我们使用POSTMAN运行上述文件时，我们会尝试在没有登录的情况下获取数据，以便给你演示未经授权的访问权限。

现在转到授权，并单击Basic authorization。输入用户名和密码，然后点击GET请求以获得所需的结果。

这是保护 Flask API 的方法，也是最基本的方法，当然还有更多更高级的方法，这里不做过多的介绍。

如何在 Flask API 上启用跟踪

至此我们已经了解了如何保护我们的 API，在未经授权的登录禁止访问，但是如果我们还想知道访问者的位置（纬度和经度点）、IP 地址、服务器名称（例如访问API 的人的详细信息），我们还可以继续配置，使用 REST API 的基本flask跟踪应用程序。首先，使用 PIP 命令安装flask跟踪包。

pip install flask-track-usage

接下来看下该程序：

from flask import Flask, g

app = Flask(__name__)

app.config['TRACK_USAGE_USE_FREEGEOIP'] = False

app.config['TRACK_USAGE_INCLUDE_OR_EXCLUDE_VIEWS'] = 'include'

from flask_track_usage import TrackUsage

from flask_track_usage.storage.printer import PrintWriter

from flask_track_usage.storage.output import OutputWriter

t = TrackUsage(app, [

  PrintWriter(),

  OutputWriter(transform=lambda s: "OUTPUT: " + str(s))

 ])

@t.include

@app.route('/')

def index():

    g.track_var["optional"] = "Write_Something"

    return "Hello"

#Run the application

if __name__ == "__main__":

   app.run(debug=True)

该程序通过导入 Track Usage、Input writer 和 output writer 来创建一个跟踪应用程序。将flask app传递给 Track 包并使用输出编写器，并使用 lambda 函数以字符串格式写入输出。之后在 slash 上创建一个基本路由，并将跟踪应用程序作为装饰器包含在内。g 代表全局，表示数据在上下文中是全局的。因此，创建一个基本 API，它在浏览器返回"Hello"，同时在后端获得所有人员的信息。

如何为REST API编写单元测试代码

现在已经为案例创建了一个不错的REST API。尽管如此，我们还需要为REST API编写单元测试代码，因为从API中识别常见错误，并确保生产安全是至关重要的。

如下是创建一个名为run的新文件并开发以下简单API。

from flask import Flask

from flask_restful import Resource, Api

import json

app = Flask(__name__)

api = Api(app)

class Helloworld(Resource):

   def __init__(self):

      pass

   def get(self):

      return json.dumps({"Message": "Fine"})

api.add_resource(Helloworld, '/')

if __name__ == '__main__':

   app.run(debug=True)

现在创建另一个名为test的文件，在其中编写用于对API进行单元测试的代码。最常见的情况是执行以下三个基本单元测试。

检查响应码是否为200
检查从API编写的内容是否为应用程序JSON格式
检查我们正在访问的所有键是否都存在于API数据处理中

from run import app

import unittest

class FlaskTest(unittest.TestCase):

 #Check for response 200

 def test_inde(self):

    tester = app.test_client(self) #tester object

    response = tester.get("/")

    statuscode = response.status_code

    self.assertEqual(statuscode, 200)

 #check if the content return is application JSON

 def test_index_content(self):

     tester = app.test_client(self)

     response = tester.get("/")

     self.assertEqual(response.content_type, "application/json")

 #check the Data returned

 def test_index_data(self):

     tester = app.test_client(self)

     response = tester.get("/")

     self.assertTrue(b'Message' in response.data)

if __name__ == '__main__':

   unittest.main()