python︱微服务Sanic制作一个简易本地restful API 方案一

2021-05-07 10:00:49 阅读：341 来源： 互联网

标签：python app request json API sanic import restful route

文章目录

一、Snaic基本功能
1.Routing路由
1.1 传入参数且参数格式规定
1.2 路由的第二种写法
2.Request 请求
3.Response
3.1 文本格式
3.2 HTML
3.3 JSON
3.4 File
案例一：回传图片案例
二、Snaic其他信息
1.app.run参数
after_start与before_stop
命令行格式运行
2.报错信息的返回
3.蓝本 Blueprint
延伸一：路由为post，如何写请求？
延伸二：设置sanic 的HTTPS服务
延伸三：压力测试
延伸四：如何回传整张图片
报错：sanic.json 对数字非常严苛
跨域问题 Sanic-CORS
py3.5/py3.6版本问题
sanic返回HTML
参考文献

Sanic是一个支持 async/await 语法的异步无阻塞框架，Flask的升级版，效率更高，性能会提升不少，我将同一服务分别用Flask和Sanic编写，再将压测的结果进行对比，发现Sanic编写的服务大概是Falsk的1.5倍。
不过Sanic对环境的要求比较苛刻：linux /Mac + python3.5+
window不支持uvloop

先上一个简单案例：

#!/usr/bin/env python
from sanic import Sanic
from sanic.response import text

app = Sanic()


@app.route("/")
async def test(request):
    return text('Hello World!')


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

来对比一下flask的code:

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)

从两者对比，可以看到相似性非常高，可以作为flask是完全没有问题的。

一、Snaic基本功能
这里笔者只解读了Snaic的三个方面：Request 、Routing、Response。

1.Routing路由
一个简单的例子：

@app.route('/post7/<param>', methods=['POST','GET'], host='example.com')

'/post7’代表接下来post时候，需要在url后面加上这个后缀：‘http://127.0.0.1:8000/post7’
methods是指request的方式接受那些方式，常见的有post/ get（大写）1.1 传入参数且参数格式规定

from sanic.response import text

@app.route('/tag/<tag>')
async def tag_handler(request, tag):
    return text('Tag - {}'.format(tag))

这边在URL中会写入一个参数，‘http://127.0.0.1:8000/tag/tag01’，async def tag_handler之中需要写入tag参数
然后该参数即可在函数中任意使用。
相似写法：


@app.route('/number/<integer_arg:int>')
async def integer_handler(request, integer_arg):
    return text('Integer - {}'.format(integer_arg))

@app.route('/number/<number_arg:number>')
async def number_handler(request, number_arg):
    return text('Number - {}'.format(number_arg))

@app.route('/person/<name:[A-z]+>')
async def person_handler(request, name):
    return text('Person - {}'.format(name))

@app.route('/folder/<folder_id:[A-z0-9]{0,4}>')
async def folder_handler(request, folder_id):
    return text('Folder - {}'.format(folder_id))

1.2 路由的第二种写法
之前看到的路由的写法都是以装饰器的形式出现：

@app.route()
async def function():

其实也可以不用装饰器，简单的写法：

from sanic.response import text

# Define the handler functions
async def person_handler2(request, name):
    return text('Person - {}'.format(name))

# Add each handler function as a route
app.add_route(person_handler2, '/person/<name:[A-z]>', methods=['GET'])

通过app.add_route的方式去加路由。
.

2.Request 请求
来看一个比较完整的例子。

# 加载
import aiohttp
import random
from sanic import Sanic
from sanic.exceptions import NotFound
from sanic.response import html, json, redirect
app = Sanic()

#定义
@app.route('/matt01')
async def index_json(request):
    # 用户定义一些传入参数
    content = request.args.get('titles')
    content _list = request.args.getlist('titles')
    # 获取数据
    return json({'titles':content ,'title_list':content _list,'args1':request.args['titles'],
                 "args2": request.args, "url": request.url, "query_string": request.query_string })
# 启动
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

request.args.get---->得到{‘titles’: [‘yes! hello’,‘no!’]}中的第一个’yes! hello’，
request.args.getlist---->得到list所有内容[‘yes! hello’, ‘no!’]，
request.args[‘titles’],---->得到[‘yes! hello’,‘no!’]，
request.args---->得到{‘titles’: [‘yes! hello’, ‘no!’]}，
request.url---->传入URL的所有内容，
request.query_string---->IP+端口+Routing之后的所有内容，

有两种获取结果的写法：
'http://127.0.0.1:8000/matt01’后面可以直接接dict，也可以用?后面接上。

get('http://127.0.0.1:8000/matt01',{'titles': ['yes! hello','no!']}).json()
	>>> {'args1': ['yes! hello', 'no!'],
	>>>  'args2': {'titles': ['yes! hello', 'no!']},
	>>>  'query_string': 'titles=yes%21+hello&titles=no%21',
	>>>  'title_list': ['yes! hello', 'no!'],
	>>>  'titles': 'yes! hello',
	>>>  'url': 'http://127.0.0.1:8000/matt01?titles=yes%21+hello&titles=no%21'}

get('http://127.0.0.1:8000/matt01?titles=value1&titles=value2').json()
	>>> {'args1': ['value1', 'value2'],
	>>>  'args2': {'titles': ['value1', 'value2']},
	>>>  'query_string': 'titles=value1&titles=value2',
	>>>  'title_list': ['value1', 'value2'],
	>>>  'titles': 'value1',
	>>>  'url': 'http://127.0.0.1:8000/matt01?titles=value1&titles=value2'}

3.Response
在Request 之中，较多的都是以json格式，也可以是很多其他的格式：text、HTML、file、Streaming等。3.1 文本格式

from sanic import response


@app.route('/text')
def handle_request(request):
    return response.text('Hello world!')

3.2 HTML

from sanic import response


@app.route('/html')
def handle_request(request):
    return response.html('<p>Hello world!</p>')

3.3 JSON

from sanic import response


@app.route('/json')
def handle_request(request):
    return response.json({'message': 'Hello world!'})

3.4 File

from sanic import response


@app.route('/file')
async def handle_request(request):
    return await response.file('/srv/www/whatever.png')

案例一：回传图片案例
回传图片用的是response.file

# 实验8：传回一张图

from sanic import response
@app.route('/file8')
async def handle_request(request):
    return await response.file('/mnt/flask/out3.jpg')

# 实验8 返回图片

# 实验8 返回图片
get('http://127.0.0.1:8000/file8')
>>> <Response [200]>

返回的是bytes格式的图片。
.

二、Snaic其他信息
1.app.run参数
来源于：Sanic框架

try:
    serve(
        host=host,
        port=port,
        debug=debug,
        # 服务开始后启动的函数
        after_start=after_start,
        # 在服务关闭前启动的函数
        before_stop=before_stop,
        # Sanic(__name__).handle_request()
        request_handler=self.handle_request,
        # 默认读取Config
        request_timeout=self.config.REQUEST_TIMEOUT,
        request_max_size=self.config.REQUEST_MAX_SIZE,
    )
except:
    pass

host(默认“127.0.0.1”): 服务器主机的地址。
port(默认8000): 服务器的端口。
debug(默认False): 启用调试（减慢服务器速度）。
ssl(默认None): 用于工作者SSL加密的SSLContext。
sock（默认None）：服务器接受连接的Socket。
worker（默认值1）：生成的工作进程数。
loop（默认None）: asyncio兼容的事件循环。如果没有指定，Sanic会创建自己的事件循环。
protocol（默认HttpProtocol）：asyncio.protocol的子类。
默认情况下，Sanic在主进程中只侦听一个CPU内核。要启动其它核心，只需指定run参数中进程的数量。

app.run(host='0.0.0.0', port=1337, workers=4)

Sanic将自动启动多个进程并在它们之间建立路由路径。建议进程数和CPU核心数一样。

after_start与before_stop
相当于：

请求 —> before_stop()函数（适合检验权限）—> 执行函数reponse —> after_start()函数（适合检验输出结果）

before_request()函数被app.before_request修饰以后，每一次请求到来后，都会先进入函数before_request()中，如上代码，获取请求的ip以及url，并打印出来，执行完毕后请求才会正常进入到app.route修饰的函数中响应，如果有多个函数被app.before_request修饰了，那么这些函数会被依次执行。
app.before_request修饰器在开发中用处非常大，比如判断某个ip是否有恶意访问行为，从而进行拦截等操作。

@app.before_request
def before_request():
    ip = request.remote_addr
    url = request.url
    print ip,
    print url

@app.app.before_request
def before_request():
    #可在此处检查jwt等auth_key是否合法，
    #abort(401)
    #然后根据endpoint，检查此api是否有权限，需要自行处理
    #print(["endpoint",connexion.request.url_rule.endpoint])
    #abort(401)
    #也可做ip检查，以阻挡受限制的ip等

因为使用restful方式，因此每次用户访问都会上传带入auth_key,如jwt等，因此可在@app.before_request中做权限的检查。

命令行格式运行
如果你将Sanic应用程序在名为server.py的文件中初始化，那么可以像这样运行服务：

python -m sanic server.app --host=0.0.0.0 --port=1337 --workers=4

2.报错信息的返回

from sanic.response import text
from sanic.exceptions import NotFound

@app.exception(NotFound)
def ignore_404s(request, exception):
    return text("Yep, I totally found the page: {}".format(request.url))

如果出现错误就会显示出以下的内容：

b'Yep, I totally found the page: http://127.0.0.1:8000/users/matt'

3.蓝本 Blueprint
把一些小功能包装在一个小集合Blueprint里面。参考

from sanic import Blueprint
from sanic.response import html, json, redirect

app = Sanic()
blueprint = Blueprint('name', url_prefix='/my_blueprint')
blueprint2 = Blueprint('name2', url_prefix='/my_blueprint2')


@blueprint.route('/foo')
async def foo(request):
    return json({'msg': 'hi from blueprint'})


@blueprint2.route('/foo')
async def foo2(request):
    return json({'msg': 'hi from blueprint2'})


app.register_blueprint(blueprint)
app.register_blueprint(blueprint2)

app.run(host="0.0.0.0", port=8000, debug=True)

Blueprint(‘name’, url_prefix=’/my_blueprint’)中为，该小蓝本的名字为‘name’(好像没啥用，后面都用不着)，前缀为：’/my_blueprint’;
定义蓝本之后，就要定义蓝本内容，@blueprint.route('/foo')还有@blueprint2.route('/foo')，即为相关内容；
app.register_blueprint(blueprint)和app.register_blueprint(blueprint2)把蓝本合并到app这个大服务中，然后一起打包给出。

因为每个蓝本有不同的前缀，所以需要在URL之后加入自己的前缀内容以示区分：

get('http://0.0.0.0:8000/my_blueprint2/foo').content
>>> b'{"msg":"hi from blueprint2"}'
get('http://0.0.0.0:8000/my_blueprint/foo').content
>>> b'{"msg":"hi from blueprint"}'

来用小蓝本进行代码简化：

code1.py

from sanic import Blueprint
from sanic.response import html, json, redirect

app = Sanic()
blueprint_v1 = Blueprint('name', url_prefix='/my_blueprint')

@blueprint_v1.route('/foo')
async def foo(request):
    return json({'msg': 'hi from blueprint'})

sanic.py

# !/usr/bin/env python
import sys

from sanic import Sanic

sys.path.append('../')
from code1 import blueprint_v1

app = Sanic(__name__)
app.blueprint(blueprint_v1)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

蓝本请求：

ip:0000/my_blueprint/foo

参考：
https://github.com/howie6879/Sanic-For-Pythoneer/blob/master/examples/demo04/sample01/src/run.py
https://github.com/howie6879/Sanic-For-Pythoneer/blob/master/examples/demo04/sample01/src/views/rss_html.py

延伸一：路由为post，如何写请求？
对于小白来说，post的方式太难懂了。

@app.route('/post12', methods=['POST'])
async def get_handler(request):
    return json('POST request - {}'.format(request.body))

# 启动
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

这边用request.get 或者request.args好像在post都不太好用，只能在get请求上可以使用。这边只有request.body比较使用。
那么Post以及Post之后的结果可见：

post('http://127.0.0.1:8000/post12',{'titles': ['value1', 'value2']}).json()
>>> "POST request - b'titles=value1&titles=value2'"
post('http://127.0.0.1:8000/post12','value1').json()
>>> "POST request - b'value1'"
# 实践不出来的几种方式
post('http://127.0.0.1:8000/post12?titles=value1&titles=value2').json()
>>> 'POST request - None'

第二个再Post之下，比较适合使用的是request.json：
这个比request.body好的一点是，中文就不用做encode(‘utf-8编码’)，就是需要变为json格式。

from requests import post
import json
sentence ='一直用'
post_data = {'sentence':sentence,'len' :3}
post('http://xxxxxx:2222/test', json.dumps(post_data)).json()

#在后台自然会返回：
request.json
>>> {'right_exclude_len': 3, 'sentence': '一直用的这款，很好，这次价格做的也比较便宜，第一次购买，比想象中要少得多，还有满减活动，很划算啊，好评！', 'left_exclude_len': 3, 'WordSpacing': 100}

如果是request.body：

from requests import post
sentence ='一直用'
post('http://xxxxxx:2222/test', sentence .encode('utf-8')).json()

延伸二：设置sanic 的HTTPS服务
网上教程很多，利用 stunnel 建立一个 SSL 加密通道绑定到 Django/Flask 内建服务器，好像不好用。
HTTPS相当于需要设置SSL证书，sanic官方文档写的是：

ssl = {'cert': "/path/to/cert", 'key': "/path/to/keyfile"}
app.run(host="0.0.0.0", port=8443, ssl=ssl)

那么这里可以使用openssl，就需要设置一步：

openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365

然后依次输入：

> openssl req -new -out ca/ca-req.csr -key ca/ca-key.pem

Country Name (2 letter code) [AU]:cn 
State or Province Name (full name) [Some-State]:guangdong 
Locality Name (eg, city) []:guangzhou
Organization Name (eg, company) [Internet Widgits Pty Ltd]:test
Organizational Unit Name (eg, section) []:test 
Common Name (eg, YOUR name) []:root 
Email Address []:test

一些个人信息都可以随意填时该目就会生成： key.pem,key.pem

ssl = {'cert': "./cert.pem", 'key': "./key.pem"}
app.run(host="0.0.0.0", port=xxxx,ssl = ssl)

延伸三：压力测试
测试环节有两个部分：单元测试与压力测试。单元测试这边其实可简单可复杂，sanic有自己的测试环节，插件：pytest，这边提一下压力测试。使用的是：locust，压力测试最好在内外网都进行测试下，当然服务器配置是你定。（主要参考：9. 测试与部署）

from requests import post,get
from locust import TaskSet, task
from locust import HttpLocust
import sys
# 端口调用环节
def action_rss(client):
    data = '心心念念的要来吃花和，好不容易找到个周日的中午有时间，气温刚好不热，组个小团来吃顿棒棒的海鲜,味道超级好。'.encode("utf-8")       # matt
    route = '/seg'         # matt
    headers = {'content-type': 'application/json'}
    return client.post(route, data =data,headers=headers).json()

# 压力测试
class RssBehavior(TaskSet):
    @task(1)
    def interface_rss(self):
        action_rss(self.client)

# locust
class Rss(HttpLocust):
    host = 'http://0.0.0.0:7777'     # matt
    task_set = RssBehavior
    min_wait = 20
    max_wait = 3000

执行的时候，在终端输入：locust -f locustfile.py --no-web -c 1，这个结果直接在终端显示，也可以使用：locust -f locustfile.py，可以通过访问IP：http://0.0.0.0:8089/，在web端口看到压力测试的结果。

这里写图片描述
再来说一下，里面的逻辑，

class Rss(HttpLocust):为主函数
min_wait/max_wait为最小、最大重复测试次数，host为主要请求的端口，此时不能加路由route class

RssBehavior(TaskSet)执行函数
笔者之前没有接触这样的模块，这段代码这段很诡异的是，一开始笔者很难理解，self.client在哪定义了？？
不过，原来在这段函数中，只要执行了Rss，那么隐藏了这么一条信息：self.client = 'http://0.0.0.0:7777'

action_rss(client) 比较奇怪的就是： client.pos环节，那么这边隐藏的条件为：还有路由'/seg'和数据data其实都可以及时变更。

client.post(route, data =data,headers=headers).json() = 'http://0.0.0.0:7777'.post('/seg',data = data)

延伸四：如何回传整张图片
之前以为把图片压缩成字节型，回传就会没有问题，但是仍然报错：

TypeError: a bytes-like object is required, not 'str'

那么怎么样的才算是合理的？
借鉴百度API的写法，用base64编码。

    import base64
    # 读入
    def get_file_content_base64(filePath):
        with open(filePath,"rb") as f:
            base64_data = base64.b64encode(f.read())
            return base64_data.decode('utf-8')
# 保存
bytespic = get_file_content_base64(filePath)
        with open(name, 'wb') as file:
            file.write(base64.b64decode(bytespic))

报错：sanic.json 对数字非常严苛

 File "/usr/local/lib/python3.5/dist-packages/sanic/app.py", line 556, in handle_request
    response = await response
  File "<ipython-input-263-38c8d86e5e18>", line 283, in text_segmentation
    return json(result)
  File "/usr/local/lib/python3.5/dist-packages/sanic/response.py", line 242, in json
    return HTTPResponse(dumps(body, **kwargs), headers=headers,
OverflowError: Maximum recursion level reached

这个报错，基本是因为json体中有一些数字格式有错误，需要调整为：

n = 1
{'version':n}
# 调整为：
{'version':int(n)}

还有遇到另外一种情况也会出现问题：

{[[...], [...]]}

跨域问题 Sanic-CORS
类似flask-cor，GitHub:https://github.com/ashleysommer/sanic-cors
最简单的使用：

from sanic import Sanic
from sanic.response import text
from sanic_cors import CORS, cross_origin

app = Sanic(__name__)
CORS(app)

@app.route("/", methods=['GET', 'OPTIONS'])
def hello_world(request):
  return text("Hello, cross-origin-world!")

py3.5/py3.6版本问题
由于Sanic不支持19.6版和更高版本的python 3.5。然而，到2020年12月，18.12LTS版本得到了支持。3.5版的官方python支持将于2020年9月到期。
参考：https://www.osgeo.cn/sanic/index.html

那么，只有19.3.1之前的版本才支持py3.5，譬如：0.6.0.2
目前支持py3.5的有：

Could not find a version that satisfies the requirement sanic==19.3 (from versions: 0.1.0, 0.1.1, 0.1.3, 0.1.4, 0.1.5, 0.1.6, 0.1.7, 0.1.8, 0.1.9, 0.2.0, 0.3.0, 0.3.1, 0.4.0, 0.4.1, 0.5.0, 0.5.1, 0.5.2, 0.5.4, 0.6.0, 0.7.0, 0.8.0, 0.8.1, 0.8.2, 0.8.3, 18.12.0, 19.3.1, 19.6.0, 19.6.2)
No matching distribution found for sanic==19.3

不然这样的代码，在py3.5中是会报错的：

(f"{host}:{port}").encode())

sanic返回HTML

@app.route('/html', methods=['POST','GET', 'OPTIONS'])
async def handle_request(request):
    htmlf=open('index.html','r',encoding="utf-8")
    htmlcont=htmlf.read()
    return html(htmlcont)

后面是可以直接通过:ip:port/html进行访问得到的。

参考文献
Running Your Flask Application Over HTTPS
【Flask】在Flask中使用HTTPS
【网络安全】在局域网里创建个人CA证书
 python Flask 使用https 安全协议

1、Sanic教程：快速开始其howie6879/Sanic-For-Pythoneer
2、官网地址
3、howie6879/Sanic-For-Pythoneer技术文档地址
4、Sanic框架
原文链接：https://blog.csdn.net/sinat_26917383/article/details/79292506

标签：python,app,request,json,API,sanic,import,restful,route
来源： https://blog.csdn.net/sirobot/article/details/116480618

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

python︱微服务Sanic制作一个简易本地restful API 方案一

文章目录