Sanic教程: 6.常用的技巧

结合前面讲的配置、项目结构、页面渲染、数据库连接，构造一个优雅的Sanic应用对你来说估计没什么大问题了，但是在实际使用过程中，可能你会碰到各种各样的需求，与之对应，你也会遇到千奇百怪的问题，除了在官方pro提issue，你大部分问题都需要自己去面对，看官方的介绍：

Async Python 3.5+ web server that's written to go fast

大概就可以明白 Sanic框架的重心不会放在诸如 session cache reload authorized这些问题上。

此篇我会将我遇到的一些问题以及解决方案一一记录下来，估计会持续更新，因为问题是不断的哈哈，可能有些问题与前面讲的有些重复，你大可略过，我将其总结成一些小技巧，供大家参考，具体如下：

• api请求json参数以及api接口验证

• gRPC的异步调用方式

• Blueprint

• html&templates编写

• cache

• 热加载

• session

对于一些问题，我将编写一个小服务来演示这些技巧，具体见demo06，依旧使用前面rss的那个例子，经过修改一番后的rss例子现在目录变成这样子了，里面加了我遇到的各种问题的解决方案，或许你只需要关注你想要了解的就好，如果你有其他的问题，欢迎issue提问，目录大概如下所示：

1. src

2. ├── config

3. │   ├── __init__.py

4. │   ├── config.py

5. │   ├── dev_config.py

6. │   └── pro_config.py

7. ├── database

8. │   ├── __init__.py

9. │   └── redis_base

10. ├── grpc_service

11. │   ├── __init__.py

12. │   ├── grpc_asyncio_client.py

13. │   ├── grpc_client.py

14. │   ├── grpc_server.py

15. │   ├── hello_grpc.py

16. │   ├── hello_pb2.py

17. │   ├── hello_pb2_grpc.py

18. │   └── proto

19. ├── statics

20. │   ├── rss_html

21. │   │   ├── css

22. │   │   └── js

23. │   └── rss_json

24. │       ├── css

25. │       └── js

26. ├── templates

27. │   ├── rss_html

28. │   └── rss_json

29. ├── tools

30. │   ├── __init__.py

31. │   └── mid_decorator.py

32. ├── views

33. │   ├── __init__.py

34. │   ├── rss_api.py

35. │   ├── rss_html.py

36. │   └── rss_json.py

37. └── run.py

和前面相比，可以看到目录里面增加了几个目录和文件，一个是关于数据库的 database，以及关于gRPC的 grpc_service，其实主要是为了演示而已，如果想了解，就可以看关于gRPC的部分，不喜欢就看其他的。

增加的文件是 rss_json.py，假设这是个接口文件吧，将会根据你的请求参数返回一串json，具体还是建议直接去看代码吧，点这里sample。

验证问题

假设你正在编写一个api服务，比如根据传的blog名字返回其rss数据，比如 rss_json.py：

1. #!/usr/bin/env python

2. from feedparser import parse

3. from sanic import Blueprint

4. from sanic.response import json

5. 
   

6. api_bp = Blueprint('rss_api', url_prefix='v1')

7. 
   

8. 
   

9. @api_bp.route("/get/rss/<param>")

10. async def get_rss_json(request, param):

11.    if param == 'howie6879':

12.        url = "http://blog.howie6879.cn/atom.xml"

13.        feed = parse(url)

14.        articles = feed['entries']

15.        data = []

16.        for article in articles:

17.            data.append({"title": article["title_detail"]["value"], "link": article["link"]})

18.        return json(data)

19.    else:

20.        return json({'info': '请访问 http://0.0.0.0:8000/v1/get/rss/howie6879'})

启动服务后，此时用 GET方式访问 http://0.0.0.0:8000/v1/get/rss/howie6879，就会返回一串你需要的json，这样使用，没什么问题，好，下面改成post请求，其中请求的参数如下：

1. {

2.    "name": "howie6879"

3. }

在 rss_json.py中加入一个新的视图函数：

1. @api_bp.route("/post/rss/", methods=['POST'])

2. async def post_rss_json(request, **kwargs):

3.    post_data = json_loads(str(request.body, encoding='utf-8'))

4.    name = post_data.get('name')

5.    if name == 'howie6879':

6.        url = "http://blog.howie6879.cn/atom.xml"

7.        feed = parse(url)

8.        articles = feed['entries']

9.        data = []

10.        for article in articles:

11.            data.append({"title": article["title_detail"]["value"], "link": article["link"]})

12.        return json(data)

13.    else:

14.        return json({'info': '参数错误'})

发送post请求到 http://0.0.0.0:8000/v1/post/rss/，依旧和上面的结果一样，代码里面有个问题，我们需要判断 name参数是不是存在于post过来的data中，解决方案很简单，加一个判断就好了，可这个是最佳方案么？

显然并不是的，如果有十个参数呢？然后再增加十几个路由呢？每个路由里有十个参数需要判断，想象一下，这样实施下来，难道要在代码里面堆积满屏幕的判断语句么？这样实在是太可怕了，我们需要更好更通用的解决方案，是时候写个装饰器来验证参数了，打开 mid_decorator.py文件，添加一个验证的装饰器函数：

1. def auth_params(*keys):

2.    """

3.    api请求参数验证

4.    :param keys: params

5.    :return:

6.    """

7. 
   

8.    def wrapper(func):

9.        @wraps(func)

10.        async def auth_param(request=None, rpc_data=None, *args, **kwargs):

11.            request_params, params = {}, []

12.            if isinstance(request, Request):

13.                # sanic request

14.                if request.method == 'POST':

15.                    try:

16.                        post_data = json_loads(str(request.body, encoding='utf-8'))

17.                    except Exception as e:

18.                        return response_handle(request, {'info': 'error'})

19.                    else:

20.                        request_params.update(post_data)

21.                        params = [key for key, value in post_data.items() if value]

22.                elif request.method == 'GET':

23.                    request_params.update(request.args)

24.                    params = [key for key, value in request.args.items() if value]

25.                else:

26.                    return response_handle(request, {'info': 'error'})

27.            else:

28.                pass

29. 
    

30.            if set(keys).issubset(set(params)):

31.                kwargs['request_params'] = request_params

32.                return await dec_func(func, request, *args, **kwargs)

33.            else:

34.                return response_handle(request, {'info': 'error'})

35. 
    

36.        return auth_param

37. 
    

38.    return wrapper

39. 
    

40. 
    

41. async def dec_func(func, request, *args, **kwargs):

42.    try:

43.        response = await func(request, *args, **kwargs)

44.        return response

45.    except Exception as e:

46.        return response_handle(request, {'info': 'error'})

注意，上面增加的路由函数改为这样：

1. @api_bp.route("/post/rss/", methods=['POST'])

2. @auth_params('name')

3. async def post_rss_json(request, **kwargs):

这样一个请求进来，就会验证参数 name是否存在，而在视图函数里面，就可以放心大胆地使用传进来的参数了，而且对于其他不同的参数验证，只要按照这个写法，直接增加验证参数就好，十分灵活方便。

对于请求验证的问题，解决方法也是类似，就是利用装饰器来实现，我自己也实现过，在上面的代码链接里面可以找到，不过现在的 Sanic官方已经提供了 demo，见这里

gRPC的异步调用方式

在编写微服务的时候，除了需要支持 http请求外，一般还需要支持 gRPC请求，我在使用 Sanic编写微服务的时候，遇到关于异步请求 RPC的需求，当时确实困扰了我，意外发现了这个库grpclib，进入 src/grpc_service目录，里面就是解决方案，很简单，这里就不多说了，直接看代码就好。

Blueprint

Blueprint前面的章节有仔细聊过，这里不多说，借用官方文档的例子，一个简单的sanic服务就搭好了：

1. # main.py

2. from sanic import Sanic

3. from sanic.response import json

4. 
   

5. app = Sanic()

6. 
   

7. @app.route("/")

8. async def test(request):

9.    return json({"hello": "world"})

10. 
    

11. #访问http://0.0.0.0:8000/即可

12. if __name__ == "__main__":

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

上面的例子可以当做一个完整的小应用，关于Blueprint的概念，可以这么理解，一个蓝图可以独立完成某一个任务，包括模板文件，静态文件，路由都是独立的，而一个应用可以通过注册许多蓝图来进行构建。 比如我现在编写的项目，我使用的是功能式架构，具体如下：

1. ├── server.py

2. ├── static

3. │   └── novels

4. │       ├── css

5. │       │   └── result.css

6. │       ├── img

7. │       │   └── read_content.png

8. │       └── js

9. │           └── main.js

10. ├── template

11. │   └── novels

12. │       └── index.html

13. └── views

14.    └── novels_blueprint.py

可以看到，总的templates以及静态文件还是放在一起，但是不同的blueprint则放在对应的文件夹中，还有一种分区式架构，则是将不同的templats以及static等文件夹全都放在不同的的blueprint中。 最后只要将每个单独的blueprint在主启动文件进行注册就好，上面的目录树是我以前的一个项目owllook的，也是用 Sanic编写的，有兴趣可以看看，不过现在结构改变挺大。

html&templates编写

编写web服务，自然会涉及到html，sanic自带有html函数，但这并不能满足有些需求，故引入jinja2迫在眉睫，使用方法也很简单：

1. # 适用python3.5+

2. # 代码片段，明白意思就好

3. from sanic import Blueprint

4. from jinja2 import Environment, PackageLoader, select_autoescape

5. 
   

6. # 初始化blueprint并定义静态文件夹路径

7. bp = Blueprint('novels_blueprint')

8. bp.static('/static', './static/novels')

9. 
   

10. # jinjia2 config

11. env = Environment(

12.    loader=PackageLoader('views.novels_blueprint', '../templates/novels'),

13.    autoescape=select_autoescape(['html', 'xml', 'tpl']))

14. 
    

15. def template(tpl, **kwargs):

16.    template = env.get_template(tpl)

17.    return html(template.render(kwargs))

18. 
    

19. @bp.route("/")

20. async def index(request):

21.    return template('index.html', title='index')

如果是 python3.6，可以试试下面的写法：

1. # 适用python3.5+

2. # 代码片段，明白意思就好

3. #!/usr/bin/env python

4. import sys

5. 
   

6. from feedparser import parse

7. from jinja2 import Environment, PackageLoader, select_autoescape

8. from sanic import Blueprint

9. from sanic.response import html

10. 
    

11. from src.config import CONFIG

12. 
    

13. # https://github.com/channelcat/sanic/blob/5bb640ca1706a42a012109dc3d811925d7453217/examples/jinja_example/jinja_example.py

14. # 开启异步特性  要求3.6+

15. enable_async = sys.version_info >= (3, 6)

16. 
    

17. html_bp = Blueprint('rss_html', url_prefix='html')

18. html_bp.static('/statics/rss_html', CONFIG.BASE_DIR + '/statics/rss_html')

19. 
    

20. # jinjia2 config

21. env = Environment(

22.    loader=PackageLoader('views.rss_html', '../templates/rss_html'),

23.    autoescape=select_autoescape(['html', 'xml', 'tpl']),

24.    enable_async=enable_async)

25. 
    

26. 
    

27. async def template(tpl, **kwargs):

28.    template = env.get_template(tpl)

29.    rendered_template = await template.render_async(**kwargs)

30.    return html(rendered_template)

cache

我在项目中主要使用redis作为缓存，使用aiocache很方便就完成了我需要的功能，当然自己利用aioredis编写也不会复杂到哪里去。

比如上面的例子，每次访问 http://0.0.0.0:8000/v1/get/rss/howie6879，都要请求一次对应的rss资源，如果做个缓存那岂不是简单很多？

改写成这样：

1. @cached(ttl=1000, cache=RedisCache, key="rss", serializer=PickleSerializer(), port=6379, namespace="main")

2. async def get_rss():

3.    print("第一次请求休眠3秒...")

4.    await asyncio.sleep(3)

5.    url = "http://blog.howie6879.cn/atom.xml"

6.    feed = parse(url)

7.    articles = feed['entries']

8.    data = []

9.    for article in articles:

10.        data.append({"title": article["title_detail"]["value"], "link": article["link"]})

11.    return data

12. 
    

13. 
    

14. @api_bp.route("/get/rss/<name>")

15. async def get_rss_json(request, name):

16.    if name == 'howie6879':

17.        data = await get_rss()

18.        return json(data)

19.    else:

20.        return json({'info': '请访问 http://0.0.0.0:8000/v1/get/rss/howie6879'})

为了体现缓存的速度，首次请求休眠3秒，请求过后，redis中就会将此次json数据缓存进去了，下次去请求就会直接冲redis读取数据。

带上装饰器，什么都解决了。

热加载

我发现有个pr实现了这个需求，如果你有兴趣，可以看这里

session

sanic对此有一个第三方插件 sanic_session，用法非常简单，有兴趣可以看看