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

概述文章目录一、Snaic基本功能1.Routing路由1.1传入参数且参数格式规定1.2路由的第二种写法2.Request请求3.Response3.1文本格式3.2HTML3.3JSON3.4File案例一：回传图片案例二、Snaic其他信息1.app.run参数after_start与before_stop命令行格式运行2.报错信息的返回3.蓝本Blu 文章目录一、Snaic基本功能1.Routing路由1.1 传入参数且参数格式规定1.2 路由的第二种写法2.Request 请求3.Response3.1 文本格式3.2 HTML3.3 JsON3.4 file案例一：回传图片案例二、Snaic其他信息1.app.run参数after_start与before_stop命令行格式运行2.报错信息的返回3.蓝本 Blueprint延伸一：路由为post，如何写请求？延伸二：设置sanic 的httpS服务延伸三：压力测试延伸四：如何回传整张图片报错：sanic.Json 对数字非常严苛跨域问题 Sanic-CORSpy3.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 pythonfrom sanic import Sanicfrom sanic.response import textapp = 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 Flaskfrom flask.ext import restfulapp = 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 [email protected]('/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 functionsasync def person_handler2(request, name):    return text('Person - {}'.format(name))# Add each handler function as a routeapp.add_route(person_handler2, '/person/<name:[A-z]>', methods=['GET'])

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

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

# 加载import aiohttpimport randomfrom sanic import Sanicfrom sanic.exceptions import NotFoundfrom sanic.response import HTML, Json, redirectapp = 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 [email protected]('/text')def handle_request(request):    return response.text('Hello World!')

3.2 HTML

from sanic import [email protected]('/HTML')def handle_request(request):    return response.HTML('<p>Hello World!</p>')

3.3 JsON

from sanic import [email protected]('/Json')def handle_request(request):    return response.Json({'message': 'Hello World!'})

3.4 file

from sanic import [email protected]('/file')async def handle_request(request):    return await response.file('/srv/www/whatever.png')

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

# 实验8：传回一张图from sanic import [email protected]('/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_requestdef before_request():    ip = request.remote_addr    url = request.url    print ip,    print url

 

@app.app.before_requestdef 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 textfrom sanic.exceptions import [email protected](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 Blueprintfrom sanic.response import HTML, Json, redirectapp = 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 Blueprintfrom sanic.response import HTML, Json, redirectapp = 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 pythonimport sysfrom sanic import Sanicsys.path.append('../')from code1 import blueprint_v1app = 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 postimport Jsonsentence ='一直用'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 postsentence ='一直用'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.pemCountry name (2 letter code) [AU]:cn State or Province name (full name) [Some-State]:guangdong Locality name (eg, city) []:guangzhouOrganization name (eg, company) [Internet WIDgits Pty Ltd]:testOrganizational 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,getfrom locust import TaskSet, taskfrom locust import httpLocustimport 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)# locustclass 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 Sanicfrom sanic.response import textfrom sanic_cors import CORS, cross_originapp = 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︱微服务Sanic制作一个简易本地restful API 方案一全部内容，希望文章能够帮你解决python︱微服务Sanic制作一个简易本地restful API 方案一所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。