Sanic 基础应用 -2

时间 2019-11-08

标签 sanic 基础应用繁體版

原文原文链接

Sanic 基础 - 2

1. 异常的处理

NotFound: 找不到合适的路由请求。
ServerError: 服务器内部出现问题时调用。一般发生在用户代码出现错误的状况。

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

app = Sanic(__name__)


@app.exception(NotFound)
def ignore_404s(request, exception):
    # 捕获页面找不到异常  404
    return text("Yep, I totally found the page: {}".format(request.url))


@app.route('/error')
def i_am_ready_to_die(request):
    raise ServerError("抛出一个异常!!", status_code=500)


if __name__ == '__main__':
    app.run()

2. 中间件和监听器

中间件:

这里有两种不一样类型的中间件：请求request和响应response。都是使用@app.middleware装饰器进行声明的，利用'request'或'response'字符串来表示其参数类型。

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

app = Sanic(__name__)


@app.middleware('request')
async def print_on_request(request):
    print("当服务器接收到请求时，我打印")
    return text('I halted the request')  # 修改响应内容


@app.middleware('response')
async def add_headers(request, response):
    response.headers["Server"] = "Fake-Server"


@app.middleware('response')
async def add_headers(request, response):
    print("当服务器返回响应时打印")
    # 添加响应头
    response.headers["x-xss-protection"] = "1; mode=block"  # 将添加HTTP头以防止跨站点脚本（XSS）攻击。
    # response.headers["Server"] = "Fake-Server"
    return text('I halted the response')  # 修改响应内容  就不会设置响应头了


@app.route('/')
async def index(request):
    return text('OK')


if __name__ == '__main__':
    app.run()

监听器:

from sanic import Sanic
from sanic.response import text

app = Sanic(__name__)


@app.listener('before_server_start')
async def setup_db(app, loop):
    # app.db = await db_setup()
    print("服务器开启时")


@app.listener('after_server_start')
async def notify_server_started(app, loop):
    print("服务器成功启动时")


@app.listener('before_server_stop')
async def notify_server_stopping(app, loop):
    print('服务器关闭以前')


@app.listener('after_server_stop')
async def close_db(app, loop):
    print('服务器关闭以后')
    await app.db.close()

    
async def notify_server_started_after_five_seconds():
    await asyncio.sleep(5)
    print('Server successfully started!')

app.add_task(notify_server_started_after_five_seconds())

@app.route('/')
async def index(request):
    return text('OK')


if __name__ == '__main__':
    app.run()

3. 蓝图

3.1 实现并注册

蓝图python

from sanic.response import json
from sanic import Blueprint, Sanic

# bp = Blueprint('my_blueprint')
bp = Blueprint(__name__) # 名字随意

@bp.route('/')
async def bp_root(request):
    return json({'my': 'blueprint'})

注册mysql

from sanic import Sanic
from my_blueprint import bp

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

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

可全局注册中间件web

@bp.middleware
async def print_on_request(request):
    print("I am a spy")


@bp.middleware('request')
async def halt_request(request):
    return text('I halted the request')


@bp.middleware('response')
async def halt_response(request, response):
    return text('I halted the response')

全局处理异常sql

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

静态文件json

bp.static('/folder/to/serve', '/web/path')

路由反响解析api

from sanic.response import text, redirect
from sanic import Blueprint, Sanic

app = Sanic()

blueprint_v1 = Blueprint('v1', url_prefix='/v1')

app.blueprint(blueprint_v1)


@blueprint_v1.route('/post/<arg>', name='post_handler')
async def post_handler(request, arg):
    return text('Post {} in Blueprint V1'.format(arg))


print(app.url_for('v1.post_handler', arg=123))  # /v1/post/123

监听浏览器

@bp.listener('before_server_start')
async def setup_connection(app, loop):
    # global database
    # database = mysql.connect(host='127.0.0.1'...)
    print('开始')


@bp.listener('after_server_stop')
async def close_connection(app, loop):
    print('结束')
    # await database.close()

3.2 API版本控制

lueprints对于API版本控制很是有用，其中一个蓝图可能指向/v1/<routes>，另外一个指向/v2/<routes>。服务器

当蓝图被初始化时，它可使用一个可选的url_prefix参数，这个参数将被添加到蓝图上定义的全部路由上。此功能可用于实现API版本控制。cookie

from sanic.response import text
from sanic import Blueprint

blueprint_v1 = Blueprint('v1', url_prefix='/v1')
blueprint_v2 = Blueprint('v2', url_prefix='/v2')


@blueprint_v1.route('/')
async def api_v1_root(request):
    return text('版本一')


@blueprint_v2.route('/')
async def api_v2_root(request):
    return text('版本二')

from sanic import Sanic
from blueprints import blueprint_v1, blueprint_v2

app = Sanic(__name__)
app.blueprint(blueprint_v1, url_prefix='/v1')
app.blueprint(blueprint_v2, url_prefix='/v2')

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

4. 配置

#  第一种方式
app = Sanic('myapp')
app.config.DB_NAME = 'appdb'
app.config.DB_USER = 'appuser'

# 第二种方式
db_settings = {
    'DB_HOST': 'localhost',
    'DB_NAME': 'appdb',
    'DB_USER': 'appuser'
}
app.config.update(db_settings)

配置加载方式

从环境变量加载

# 任何由SANIC_定义的变量都将应用于sanic配置。例如，设置SANIC_REQUEST_TIMEOUT自动加载应用程序。你可使用load_cars将布尔值传递给Sanic构造函数来进行覆盖。

app = Sanic(load_vars=False)

从对象加载

# 若是有不少配置参数而且它们有合理的默认值，将它们放置于模块是有帮助的。

import myapp.default_settings

app = Sanic('myapp')
app.config.from_object(myapp.default_settings)

从文件加载

# 一般状况下，你想要从文件中加载配置参数。你能够从from_file(/path/to/config_file)来加载配置参数。然而，这须要程序知道配置文件的位置，因此你能够在环境变量中指定配置文件的路径，并让Sanic寻找配置文件并使用配置文件。
app = Sanic('myapp')
app.config.from_envvar('MYAPP_SETTINGS')

# 而后你能够在MYAPP_SETTINGS环境设置下运行你的应用程序：
$ MYAPP_SETTINGS=/path/to/config_file python3 myapp.py
INFO: Goin' Fast @ http://0.0.0.0:8000

# 配置文件是常规的Python文件，运行它们只是为了加载配置。这容许你使用任何正确的逻辑进行正确的配置。只要uppercase变量被添加到配置中，最多见的配置包括简单的键值对：

# config_file
DB_HOST = 'localhost'
DB_NAME = 'appdb'
DB_USER = 'appuser'

5. COOKIES

响应的cookies能够设置为字典值，同时也有如下参数可用：app

expires(时间): cookie最后在客户端浏览器上存在时间。
path(字符串): Cookie的URL子集。默认为/。
comment(字符串): 注释（元数据）。
domain(字符串): 指定cookie有效的域。显式指定的域必须始终以点开头。
max-age(数字): cookie应该存在的秒数。
secure(布尔值): 指定cookie是否只能经过HTTPS发送。
httponly(布尔值): 指定cookie是否能被Javascript读取。

from sanic import Sanic
from sanic.response import text

app = Sanic()


@app.route("/get_cookie")
async def test(request):
    test_cookie = request.cookies.get('test')
    return text("Test cookie set to: {}".format(test_cookie))


@app.route("/set_cookie")
async def test(request):
    response = text("There's a cookie up in this response")
    response.cookies['id'] = 'It worked!'
    # response.cookies['test']['domain'] = '.gotta-go-fast.com'
    # response.cookies['test']['httponly'] = True
    return response


@app.route("/del_cookie")
async def test(request):
    response = text("删除cookies")

    # 此cookie将被设置为0秒后过时
    del response.cookies['kill_me']

    # 这个cookie将在5秒内自动销毁
    response.cookies['short_life'] = 'Glad to be here'
    response.cookies['short_life']['max-age'] = 5
    del response.cookies['favorite_color']

    # This cookie will remain unchanged
    response.cookies['favorite_color'] = 'blue'
    response.cookies['favorite_color'] = 'pink'
    del response.cookies['favorite_color']

    return response


if __name__ == '__main__':
    app.run()

6.装饰器添加

from functools import wraps

from sanic import Sanic
from sanic.response import json

app = Sanic()


def auth():
    def decorator(f):
        @wraps(f)
        async def decorated_function(request, *args, **kwargs):
            # 判断是否经过 认证
            is_authorized = True
            if is_authorized:
                # 经过认证
                response = await f(request, *args, **kwargs)
                return response
            else:
                # 未经过认证
                return json({'status': 'not_authorized'}, 403)
        return decorated_function

    return decorator


@app.route("/")
@auth()
async def test(request):
    return json({"status": 'authorized'})


if __name__ == '__main__':
    app.run()

7. CBV 试图写法

from functools import wraps

from sanic.response import text, json, redirect
from sanic.views import HTTPMethodView, CompositionView
from sanic import Sanic

app = Sanic(__name__)


def auth():
    def decorator(f):
        @wraps(f)
        async def decorated_function(request, *args, **kwargs):
            # 判断是否经过 认证
            is_authorized = True
            print("认证装饰器")
            if is_authorized:
                # 经过认证
                response = await f(request, *args, **kwargs)
                return response
            else:
                # 未经过认证
                return json({'status': 'not_authorized'}, 403)

        return decorated_function

    return decorator


# @app.route('/')
async def index(request):
    url = app.url_for('Users', id=10)  # 反向解析
    print(url)  # /user/10
    return redirect(url)  # 302

# 使用组成视图 处理错误请求
view = CompositionView()
view.add(['GET'], index)
view.add(['POST', 'PUT'], lambda request: text('我没有 post / put 方法'))
app.add_route(view, '/')


class Users(HTTPMethodView):
    decorators = [auth(), ]  # 加装饰器

    # 能够不使用异步
    async def get(self, request, id):
        print(id)
        return text('{}---OK'.format(id))

    async def post(self, request,id):
        return text("{}".format(request.form))
    
    ...


app.add_route(Users.as_view(), '/user/<id>')

if __name__ == '__main__':
    app.run()

8.自定义协议

from sanic import Sanic
from sanic.response import text
from sanic.server import HttpProtocol

app = Sanic()


# 自定义 http 协议
class CustomHttpProtocol(HttpProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def write_response(self, response):
        if isinstance(response, str):
            response = text(response)
        self.transport.write(
            response.output(self.request.version)
        )
        self.transport.close()

    
@app.route('/')
async def string(request):
    return 'string'


@app.route('/1')
async def response(request):
    return text('response')


app.run(protocol=CustomHttpProtocol)  # 指定协议

9.SSL 证书

import ssl
ctx = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
ctx.load_cert_chain("/path/to/cert", keyfile="/path/to/keyfile")

app.run(host="0.0.0.0", port=8443, ssl=ctx)

#  你还能够将证书和密钥的位置作为自检进行传递。
ssl = {'cert': "/path/to/cert", 'key': "/path/to/keyfile"}
app.run(host="0.0.0.0", port=5000, ssl=ssl)

10.命令行启动

须要指定 server.py 文件为启动文件

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