sanic自定义token认证 session cache

      

1.前言

在使用过程中，我尽量将程序编写成异步的，首先进行安装：

python -m pip install sanic

sanic的文档写得很详细，但是在使用过程中我还是有些问题。
下面记录的都是我在使用sanic过程中遇到的问题，后续有新问题会继续补充：

• 1.blueprint
• 2.html templates编写（引入jinja2）
• 3.session（引入sanic_session）
• 4.缓存（引入aiocache）
• 5.api接口验证（自定义一个装饰器）
• 6.恶意将其他域名绑定到你的网站独立ip

2.问题记录

2.1.blueprint

借用官方文档的例子，一个简单的sanic服务就搭好了：

# main.py
from sanic import Sanic
from sanic.response import json

app = Sanic()

@app.route("/")
async def test(request):
    return json({"hello": "world"})

#访问http://0.0.0.0:8000/即可
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

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

├── server.py
├── static
│   └── novels
│       ├── css
│       │   ├── chapter.css
│       │   ├── content.css
│       │   ├── index.css
│       │   ├── main.css
│       │   └── result.css
│       ├── img
│       │   ├── bookMark.png
│       │   ├── favicon.ico
│       │   ├── read_bg.png
│       │   └── read_content.png
│       └── js
│           ├── list.js
│           └── main.js
├── template
│   └── novels
│       ├── chapter.html
│       ├── content.html
│       ├── donate.html
│       ├── feedback.html
│       ├── index.html
│       ├── main.html
│       └── result.html
└── views
    └── novels_blueprint.py

可以看到，总的templates以及静态文件还是放在一起，但是不同的blueprint则放在对应的文件夹中，还有一种分区式架构，则是将不同的templats以及static等文件夹全都放在不同的的blueprint中。
最后只要将每个单独的blueprint在主启动文件进行注册就好，比如我的这个项目：

# server.py
#!/usr/bin/env python
from sanic import Sanic
from novels_search.views.novels_blueprint import bp

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

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

具体的novels_blueprint.py请点击这里

2.2.html templates编写

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

# novels_blueprint.py片段
from sanic import Blueprint
from jinja2 import Environment, PackageLoader, select_autoescape

# 初始化blueprint并定义静态文件夹路径
bp = Blueprint('novels_blueprint')
bp.static('/static', './static/novels')

# jinjia2 config
env = Environment(
    loader=PackageLoader('views.novels_blueprint', '../templates/novels'),
    autoescape=select_autoescape(['html', 'xml', 'tpl']))

def template(tpl, **kwargs):
    template = env.get_template(tpl)
    return html(template.render(kwargs))

@bp.route("/")
async def index(request):
    return template('index.html', title='index')

这样，就实现了jinja2 的引入。

2.3.session

sanic对此有一个第三方插件sanic_session，用法非常简单，见官方例子如下：

import asyncio_redis

from sanic import Sanic
from sanic.response import text
from sanic_session import RedisSessionInterface

app = Sanic()


# Token from https://github.com/subyraman/sanic_session

class Redis:
    """
    A simple wrapper class that allows you to share a connection
    pool across your application.
    """
    _pool = None

    async def get_redis_pool(self):
        if not self._pool:
            self._pool = await asyncio_redis.Pool.create(
                host='localhost', port=6379, poolsize=10
            )

        return self._pool


redis = Redis()

# pass the getter method for the connection pool into the session
session_interface = RedisSessionInterface(redis.get_redis_pool, expiry=604800)


@app.middleware('request')
async def add_session_to_request(request):
    # before each request initialize a session
    # using the client's request
    await session_interface.open(request)


@app.middleware('response')
async def save_session(request, response):
    # after each request save the session,
    # pass the response to set client cookies
    await session_interface.save(request, response)


@app.route("/")
async def test(request):
    # interact with the session like a normal dict
    if not request['session'].get('foo'):
        request['session']['foo'] = 0

    request['session']['foo'] += 1

    response = text(request['session']['foo'])

    return response


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8888, debug=True)

2.4.缓存

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

@cached(ttl=1800, cache=RedisCache, key_from_attr='url', serializer=PickleSerializer(),
        endpoint=REDIS_DICT.get('REDIS_ENDPOINT', None), port=REDIS_DICT.get('REDIS_PORT', None), namespace="main")
async def cache_owllook_novels_chapter(url, netloc):
    async with aiohttp.ClientSession() as client:
        html = await target_fetch(client=client, url=url)
        if html:
            soup = BeautifulSoup(html, 'html5lib')
            selector = RULES[netloc].chapter_selector
            if selector.get('id', None):
                content = soup.find_all(id=selector['id'])
            elif selector.get('class', None):
                content = soup.find_all(class_=selector['class'])
            else:
                content = soup.find_all(selector.get('tag'))
            return str(content) if content else None
        return None

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

2.5.api接口验证

在编写接口的时候，有些接口可能需要在headers中加个验证头或者token，sanic官方貌似并没有提供，解决起来很简单，可以在request时候进行验证，代码如下：

# 比如想验证一个请求的 Owllook-Api-Key值是否跟自己约定的值一样，这里写个装饰器就好
from functools import wraps
from sanic.response import json
from novels_search.config import AUTH

def authenticator(key):
    """

    :param keys: 验证方式 Owllook-Api-Key : Maginc Key,  Authorization : Token
    :return: 返回值
    """

    def wrapper(func):
        @wraps(func)
        async def authenticate(request, *args, **kwargs):
            value = request.headers.get(key, None)
            if value and AUTH[key] == value:
                response = await func(request, *args, **kwargs)
                return response
            else:
                return json({'msg': 'not_authorized', 'status': 403})

        return authenticate

    return wrapper

至于使用

@api_bp.route("/bd_novels/<name>")
@authenticator('Owllook-Api-Key')
async def bd_novels(request, name):
    pass

具体见例子

2.6.恶意将其他域名绑定到你的网站独立ip

前两天突然发现别人没经过我的同意就将自己的域名绑定到我的网站，不能忍，解决办法有不少，web服务器写配置解决，或者代码层面解决，我说下第二种。

首先，定义你允许访问的HOST，再获取访问时获取的host，比如：

# config.py
HOST = ['127.0.0.1:8001']

# sanic获取host
host = request.headers.get('host', None)

那么就可以在sanic中间件中定义：

# 当host不满足条件就调到baidu
@app.middleware('request')
async def add_session_to_request(request):
    host = request.headers.get('host', None)
    if not host or host not in HOST:
        return redirect('http://www.baidu.com')