精华内容
下载资源
问答
  • Sanic 中文文档 拉取了 Sanic 官方文档,翻译成了中文文档,仅供参考。 官方介绍 Sanic 是一款类似 Flask 的 Python 3.5+ Web服务器,它的写入速度非常快。它基于 magicstack 那群令人惊叹的人所完成的工作,并且受...
  • pytest-sanic 建造 文件 包裹 一个pytest插件。 它可以帮助您异步测试代码。 该插件提供: 使用异步协程非常简单的测试 常用和有用的装置 异步夹具支持 用于Sanic应用程序的test_client / sanic_client 用于...
  • Support Sanic

    2021-01-09 04:31:26
    <div><p>Fixes #496 . ...<ul><li>support Sanic</li><li>creating a sanic_api.py and sanic_app.py modules</li><li>add relevant tests</li></ul>该提问来源于开源项目:zalando/connexion</p></div>
  • Sanic-Sentry Sanic-Sentry -- Sentry 集成到 sanic web 服务器。 要求 Python>= 3.6 安装 Sanic-Sentry应该使用 pip 安装: pip install sanic-sentry 用法 要初始化插件,您可以将“app”传递给 __init__: >...
  • Sanic integration

    2020-12-08 19:01:43
    <div><p>I assume there are <em>no</em> plans whatsoever to integrate with <a href="https://github.com/channelcat/sanic">Sanic, so I'll be handling it manually, but I'll be more than interested...
  • sanic-openapi and sanic-jwt

    2020-12-25 21:04:40
    <div><p>I use sanic-openapi and sanic-jwt How to set <code>.consumes</code> and <code>.produces</code> for below API's of sanic-jwt <p>POST/auth/ GET/auth/me/ GET/auth/verify/ 该提问来源于开源...
  • Sanic指南 即将到
  • sanic入门

    2020-11-28 15:27:25
    都是抄的api:https://sanic.readthedocs.io/en/latest/index.html 安装: pip3 install sanic conda config --add ...SANIC_NO_UVLOOP=true SANIC_NO_UJSON=true pip3 install --no-binary :all: sanic ...

    都是抄的api:https://sanic.readthedocs.io/en/latest/index.html

     

    安装:

    • pip3 install sanic
    • conda config --add channels conda-forge
      conda install sanic

    如果不想安装uvloop或者ujson,可以:

    SANIC_NO_UVLOOP=true SANIC_NO_UJSON=true pip3 install --no-binary :all: sanic

      

    Demo:

     

    from sanic import Sanic
    from sanic.response import json
    
    app = Sanic("hello_example")
    
    @app.route("/")
    async def test(request):
      return json({"hello": "world"})
    
    if __name__ == "__main__":
      app.run(host="0.0.0.0", port=8000)

     直接run即可

     

    配置文件:

    Sanic在应用程序对象的config属性中保存配置。配置对象只是一个对象,可以修改,使用点符号或像一个字典(一般参数是大写的):

    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的设置将被应用程序自动加载,并被输入到REQUEST_TIMEOUT配置变量中。也可以传递不同的前缀给Sanic:

    app = Sanic(__name__, load_env='MYAPP_'),这样就使用MYAPP_这个前缀了

    相应的不想从环境变量读app = Sanic(__name__, load_env=False)

    从文件、字典或者其他有__dict__属性的对象读取配置:

    • 从文件my_config.py:

    # my_config.py长得如下
    A = 1
    B = 2

     读取方法:

    app.update_config("/path/to/my_config.py")

     文件路径可以用环境变量:

    $ export my_path="/path/to"

    app.update_config("${my_path}/my_config.py")

     

    • 字典类似如上
    • 从类或者对象中:
    class C:
        A = 1
        B = 2
    
    app.update_config(C)
    
    app.update_config(C())

     

    内置配置:

    Variable Default Description
    REQUEST_MAX_SIZE 100000000 How big a request may be (bytes)
    REQUEST_BUFFER_QUEUE_SIZE 100 Request streaming buffer queue size
    REQUEST_TIMEOUT 60 How long a request can take to arrive (sec)
    RESPONSE_TIMEOUT 60 How long a response can take to process (sec)
    RESPONSE_TIMEOUT True Disables keep-alive when False
    KEEP_ALIVE_TIMEOUT 5 How long to hold a TCP connection open (sec)
    WEBSOCKET_MAX_SIZE 2^20 Maximum size for incoming messages (bytes)
    WEBSOCKET_MAX_QUEUE 32 Maximum length of the queue that holds incoming messages
    WEBSOCKET_READ_LIMIT 2^16 High-water limit of the buffer for incoming bytes
    WEBSOCKET_WRITE_LIMIT 2^16 High-water limit of the buffer for outgoing bytes
    WEBSOCKET_PING_INTERVAL 20 A Ping frame is sent every ping_interval seconds.
    WEBSOCKET_PING_TIMEOUT 20 Connection is closed when Pong is not received after ping_timeout seconds
    GRACEFUL_SHUTDOWN_TIMEOUT 15.0 How long to wait to force close non-idle connection (sec)
    ACCESS_LOG True Disable or enable access log
    FORWARDED_SECRET None Used to securely identify a specific proxy server (see below)
    PROXIES_COUNT None The number of proxy servers in front of the app (e.g. nginx; see below)
    FORWARDED_FOR_HEADER “X-Forwarded-For” The name of “X-Forwarded-For” HTTP header that contains client and proxy ip
    REAL_IP_HEADER None The name of “X-Real-IP” HTTP header that contains real client ip

     

     

    日志

    demo: 

    from sanic.log import logger
    from sanic.response import text
    
    app = Sanic('logging_example')
    
    @app.route('/')
    async def test(request):
        logger.info('Here is your log')
        return text('Hello World!')
    
    if __name__ == "__main__":
      app.run(debug=True, access_log=True)
    
    

     

    配置

    使用logging.config.dictConfig或者用log_config配置日志:

    app = Sanic('logging_example', log_config=LOGGING_CONFIG)

     

    关闭日志:

    if __name__ == "__main__":
      app.run(debug=False,access_log=False)

     

    默认的日志格式:

    %(asctime)s - (%(name)s)[%(levelname)s][%(host)s]: %(request)s %(message)s %(status)d %(byte)d

     

    获取请求数据:

     

    json (any) - JSON body

    from sanic.response import json
    
    @app.route("/json")
    def post_json(request):
      return json({ "received": True, "message": request.json })

     args (dict) - Query string variables. A query string is the section of a URL that resembles ?key1=value1&key2=value2.

    from sanic.response import json
    @app.route("/query_string")
    def query_string(request):
      return json({ "parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string })
    
    

     form (dict) - Posted form variables

    @app.route("/form")
    def post_json(request):
      return json({ "received": True, "form_data": request.form, "test": request.form.get('test') })

     body (bytes) - Posted raw body

    from sanic.response import text
    
    @app.route("/users", methods=["POST",])
    def create_user(request):
        return text("You are trying to create a user with the following POST: %s" % request.body)

     

    创造响应值Response

    html

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

     json

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

     一些demo:

      Simple App with Sanic Views

    from sanic import Sanic
    from sanic.views import HTTPMethodView
    from sanic.response import text
    
    app = Sanic('some_name')
    
    class SimpleView(HTTPMethodView):
    
    def get(self, request):
            return text('I am get method')
    
    def post(self, request):
            return text('I am post method')
    
    def put(self, request):
            return text('I am put method')
    
    def patch(self, request):
            return text('I am patch method')
    
    def delete(self, request):
            return text('I am delete method')
    
    class SimpleAsyncView(HTTPMethodView):
    
    async def get(self, request):
            return text('I am async get method')
    
    async def post(self, request):
            return text('I am async post method')
    
    async def put(self, request):
            return text('I am async put method')
    
    app.add_route(SimpleView.as_view(), '/')
    app.add_route(SimpleAsyncView.as_view(), '/async')
    
    if __name__ == '__main__':
        app.run(host="0.0.0.0", port=8000, debug=True)
    
    

     

    URL Redirect

    from sanic import Sanic
    from sanic import response
    
    app = Sanic(__name__)
    
    @app.route('/')
    def handle_request(request):
        return response.redirect('/redirect')
    
    @app.route('/redirect')
    async def test(request):
        return response.json({"Redirected": True})
    
    if __name__ == '__main__':
        app.run(host="0.0.0.0", port=8000)

     

    Logging Enhancements

    from sanic import response
    import logging
    
    logging_format = "[%(asctime)s] %(process)d-%(levelname)s "
    logging_format += "%(module)s::%(funcName)s():l%(lineno)d: "
    logging_format += "%(message)s"
    
    logging.basicConfig(
        format=logging_format,
        level=logging.DEBUG
    )
    log = logging.getLogger()
    
    # Set logger to override default basicConfig
    sanic = Sanic()
    
    @sanic.route("/")
    def test(request):
        log.info("received request; responding with 'hey'")
        return response.text("hey")
    
    sanic.run(host="0.0.0.0", port=8000)from sanic import response
    import logging
    
    logging_format = "[%(asctime)s] %(process)d-%(levelname)s "
    logging_format += "%(module)s::%(funcName)s():l%(lineno)d: "
    logging_format += "%(message)s"
    
    logging.basicConfig(
        format=logging_format,
        level=logging.DEBUG
    )
    log = logging.getLogger()
    
    # Set logger to override default basicConfig
    sanic = Sanic()
    
    @sanic.route("/")
    def test(request):
        log.info("received request; responding with 'hey'")
        return response.text("hey")
    
    sanic.run(host="0.0.0.0", port=8000)

     

    去学习一下py3.5的异步编程咯... 

    展开全文
  • Sanic middleware

    2020-12-27 21:37:44
    <ul><li>some of the requirements of sanic don't seem to work on 2.7</li><li>I'm geting a failure on protobuf when attempting to run this ouside of the tests.</li></ul> <p>Sanic doesn't ...
  • Sanic学习——初识Sanic

    2020-04-04 21:01:08
    一般我们学习一个新知识的时候总需要问三个问题,我是谁?我在哪?我要干什么?...Sanic is a Python 3.6+ web server and web framework that’s written to go fast. It allows the usage of t...

    一般我们学习一个新知识的时候总需要问三个问题,我是谁?我在哪?我要干什么?哦,不不不,不是这三个灵魂拷问。哈哈,应该是是什么?为什么?怎么办?

    Sanic是什么?

    话不多说先给一个官方的解释

    Sanic is a Python 3.6+ web server and web framework that’s written to go fast. It allows the usage of the async/await syntax added in Python 3.5, which makes your code non-blocking and speedy.

    The goal of the project is to provide a simple way to get up and running a highly performant HTTP server that is easy to build, to expand, and ultimately to scale.

    Sanic是python3.6以上支持async和await语法的web符合和web框架,其目的旨在提供一个建档的方法构建和运行高性能的http服务

    从这段描述可以提取几个关键字,异步,高性能,易于构建扩展。是不是你想要的优点它全都有,就冲着这几个特性,那也得学习下。

    为什么要学Sanic?

    其实官方已经给了很好的解释,优点多多,不学是不是有点太可惜了,哈哈,我已经磨刀霍霍,准备上手了。

    来看一下官方给的demo,是不是超级简单,8行代码就可以启动一个服务,岂不妙哉。

    from sanic import Sanic
    from sanic.response import json
    
    app = Sanic()
    
    @app.route("/")
    async def test(request):
        return json({"hello": "world"})
    
    if __name__ == "__main__":
        app.run(host="0.0.0.0", port=8000)
    

    怎么学Sanic?

    1、官方文档
    2、当然是其他人写的博客咯,遇到问题查查就出来了
    3、源代码
    通过这三个途径,然后勤学苦练,有朝一日,必有所成。

    那么话不多说,赶紧操练起来

    # 注意python应该是3.6+,最好是3.6.8以上的版本为好,
    # 不知道怎么更新python3的,请移步百度
    pip install sanic
    

    唰唰唰,也不知过了多久,sanic终于装好了,将官方给到的demo代码运行,哈哈,成功了,又掌握了一种方式写hello,world了,Sanic诚不欺我。

    欲知后续如何,请听下回分解。

    展开全文
  • Sanic路由器-源码

    2021-02-15 15:03:17
    Sanic路由器 Sanic的内部处理程序路由从v21.3开始。
  • sanic-jinja2 Jinja2对Sanic的支持 安装 python3 -m pip install sanic-jinja2 特征 sanic-jinja2支持: 烧瓶状flash法 i18n和Babel支持 @jinja.template语法 支持 用于创建应用程序的工厂模式init_app方法 用法 ...
  • Sanic框架

    2019-02-19 13:12:00
    Sanic框架 1. 入门 Sanic 是一款类似Flask的Web服务器,它运行在Python 3.5+上。 除了与Flask功能类似之外,它还支持异步请求处理,这意味着你可以使用Python3.5 中新的异步/等待语法,使你的程序运行更加...

    Sanic框架

    1. 入门

    Sanic 是一款类似Flask的Web服务器,它运行在Python 3.5+上。

    除了与Flask功能类似之外,它还支持异步请求处理,这意味着你可以使用Python3.5 中新的异步/等待语法,使你的程序运行更加快速。

    1.1 简单起步

    from sanic import Sanic
    from sanic.response import json
    
    app = Sanic()
    
    @app.route("/")
    async def test(request):
        return json({"hello": "world"})
    
    if __name__ == "__main__":
        app.run(host="0.0.0.0", port=8000)
    
    1. 保存到main.py文件,运行文件python3 main.py
    2. 打开URLhttp://0.0.0.0:8000,可以看到网页显示 Hello World信息。

    2. 路由

    路由允许用户为不同的URL地址指定处理的函数。

    一个基本的路由就像下面的例子,而app就是Sanic类的一个实例。

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

    当地址http://server.url/被访问时(服务的基础地址),根地址/就会被路由匹配一个定义了返回JSON对象的test函数。

    必须使用async def语法定义函数,来保证其可以进行异步处理。

    2.1 请求参数

    Sanic的基础路由支持请求参数的操作。

    如果需要指定参数,请使用尖括号<PARAM>将指定参数括起来。请求参数将作为路由函数的关键字参数。

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

    如果需要指定添加的参数的类型,则要在参数名字后面添加:type指定参数类型。如果参数与指定的参数类型不匹配,则Sanic会抛出NotFound的异常,从而导致页面出现404: Page not found的错误。

    from sanic.response import text
    
    @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))
    

    2.2 HTTP 请求类型

    默认情况下,一个路由会定义一个仅仅适用于URL的GET请求。然而,@app.route装饰器接受一个可选的参数methods,它允许定义的函数使用列表中任何一个的HTTP方法。

    from sanic.response import text
    
    @app.route('/post', methods=['POST'])
    async def post_handler(request):
        return text('POST request - {}'.format(request.json))
    
    @app.route('/get', methods=['GET'])
    async def get_handler(request):
        return text('GET request - {}'.format(request.args))
    

    这里还有一个可选的host参数(列表或是字符串)。它限制了给主机的路由。如果存在一个没有主机的路由,它将是一个默认值。

    @app.route('/get', methods=['GET'], host='example.com')
    async def get_handler(request):
        return text('GET request - {}'.format(request.args))
    
    # if the host header doesn't match example.com, this route will be used
    @app.route('/get', methods=['GET'])
    async def get_handler(request):
        return text('GET request in default - {}'.format(request.args))
    

    这里还有一种快速使用装饰器的方法:

    from sanic.response import text
    
    @app.post('/post')
    async def post_handler(request):
        return text('POST request - {}'.format(request.json))
    
    @app.get('/get')
    async def get_handler(request):
        return text('GET request - {}'.format(request.args))
    

    2.3 add_route方法

    就像上文提到的,路由通常使用@app.route装饰器进行添加的。但是,这个装饰器只是app.add_route方法的一个封装。它看起来像下面这样:

    from sanic.response import text
    
    # Define the handler functions
    async def handler1(request):
        return text('OK')
    
    async def handler2(request, name):
        return text('Folder - {}'.format(name))
    
    async def person_handler2(request, name):
        return text('Person - {}'.format(name))
    
    # Add each handler function as a route
    app.add_route(handler1, '/test')
    app.add_route(handler2, '/folder/<name>')
    app.add_route(person_handler2, '/person/<name:[A-z]>', methods=['GET'])
    

    2.4 利用url_for生成URL

    Sanic提供了一个根据处理函数名字生成URL的方法url_for。在应用中,它使用处理程序的名字来有效避免使用实际的网络路径。

    @app.route('/')
    async def index(request):
        # generate a URL for the endpoint `post_handler`
        url = app.url_for('post_handler', post_id=5)
        # the URL is `/posts/5`, redirect to it
        return redirect(url)
    
    
    @app.route('/posts/<post_id>')
    async def post_handler(request, post_id):
        return text('Post - {}'.format(post_id))
    

    使用url_for需要注意的是:

    • 传递给url_for的关键字如果不是请求参数,将包含在URL 的查询字符串中。
    url = app.url_for('post_handler', post_id=5, arg_one='one', arg_two='two')
    # /posts/5?arg_one=one&arg_two=two
    
    • 可以传递多个参数给url_for函数。
    url = app.url_for('post_handler', post_id=5, arg_one=['one', 'two'])
    # /posts/5?arg_one=one&arg_one=two
    
    • 还可以传递一些特殊的参数给url_for方法来构造一些特殊的URL,诸如:_anchor,_external,_scheme,_server
    url = app.url_for('post_handler', post_id=5, arg_one='one', _anchor='anchor')
    # /posts/5?arg_one=one#anchor
    
    url = app.url_for('post_handler', post_id=5, arg_one='one', _external=True)
    # //server/posts/5?arg_one=one
    # _external requires passed argument _server or SERVER_NAME in app.config or url will be same as no _external
    
    url = app.url_for('post_handler', post_id=5, arg_one='one', _scheme='http', _external=True)
    # http://server/posts/5?arg_one=one
    # when specifying _scheme, _external must be True
    
    # you can pass all special arguments one time
    url = app.url_for('post_handler', post_id=5, arg_one=['one', 'two'], arg_two=2, _anchor='anchor', _scheme='http', _external=True, _server='another_server:8888')
    # http://another_server:8888/posts/5?arg_one=one&arg_one=two&arg_two=2#anchor
    
    • 所有的参数都必须正确地传递给url_for方法来构造URL。如果未提供参数或指定参数不匹配,将抛出URLBuildError的错误。

    2.5 WebSocket 路由

    使用@app.websocket装饰器定义WebSocket协议的路由。

    @app.websocket('/feed')
    async def feed(request, ws):
        while True:
            data = 'hello!'
            print('Sending: ' + data)
            await ws.send(data)
            data = await ws.recv()
            print('Received: ' + data)
    

    或者,使用app.add_websocket_route方法来代替@app.websocket装饰器。

    async def feed(request, ws):
        pass
    
    app.add_websocket_route(my_websocket_handler, '/feed')
    

    WebSocket路由的处理程序将请求作为第一个参数传递,并将WebSocket协议对象作为第二个参数传递。而协议对象具有sendrecv两个方法来进行数据的传送和接收。

    3. 请求数据

    当接收端接收到一个HTTP请求的时候,路由函数就会传递一个Request对象。

    以下的变量可以作为Request对象的属性进行访问。

    • json(任何类型)-JSON格式的数据
    from sanic.response import json
    
    @app.route("/json")
    def post_json(request):
        return json({ "received": True, "message": request.json })
    
    • arg(dict类型)-查询字符串变量。一个查询的字符串是部分的URL,类似于?key1=value1&key2=value2,如果要解析这个URL,那么arg字典看起来就像{'key1': ['value1'], 'key2': ['value2']},这个请求将使用query_string变量来保存未解析的字符串的值。
    from sanic.response import json
    
    @app.route("/query_string")
    def query_string(request):
        return json({ "parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string })
    
    • raw_args(dict类型)-在许多情况下,你只需要访问一段很小的url参数。对于前文的URL的?key1=value1&key2=value2raw_args字典看起来就像这样{'key1': 'value1', 'key2': 'value2'}

    • files(File对象的字典)-具有名称,正文和类型的文件列表。

    from sanic.response import json
    
    @app.route("/files")
    def post_json(request):
        test_file = request.files.get('test')
    
        file_parameters = {
            'body': test_file.body,
            'name': test_file.name,
            'type': test_file.type,
        }
    
        return json({ "received": True, "file_names": request.files.keys(), "test_file_parameters": file_parameters })
    
    • form(dict类型)-发布表单数据
    from sanic.response import json
    
    @app.route("/form")
    def post_json(request):
        return json({ "received": True, "form_data": request.form, "test": request.form.get('test') })
    
    • body(字节类型)-发布正文。这个属性允许检索请求的原始数据,而无需理会数据的类型。
    from sanic.response import text
    
    @app.route("/users", methods=["POST",])
    def create_user(request):
        return text("You are trying to create a user with the following POST: %s" % request.body)
    
    • ip(str类型)-请求者的IP地址

    • app-对正在处理此请求的Sanic应用程序对象的引用。在无法访问全局app对象或其它处理程序的时候很有用。

    from sanic.response import json
    from sanic import Blueprint
    
    bp = Blueprint('my_blueprint')
    
    @bp.route('/')
    async def bp_root(request):
        if request.app.config['DEBUG']:
            return json({'status': 'debug'})
        else:
            return json({'status': 'production'})
    
    • url:完整的请求URL,即:http://localhost:8000/posts/1/?foo=bar

    • scheme:与请求相关联的URL方案,http或是https

    • host:与请求相关联的主机,localhost:8080

    • path:请求的地址/post/1

    • query_string:请求的查询字符串,foo=bar或是空字符串''

    3.1 使用getgetlist来访问值

    请求的属性实际上是返回一个dict的子类RequestParameters。在使用这个对象的主要区别是getgetlist方法的不同。

    • get(key, default=None) 当给定key的值是一个列表的时候,只返回第一个项目。

    • getlist(key, default=None) 返回整个列表。

    from sanic.request import RequestParameters
    
    args = RequestParameters()
    args['titles'] = ['Post 1', 'Post 2']
    
    args.get('titles') # => 'Post 1'
    
    args.getlist('titles') # => ['Post 1', 'Post 2']
    

    4. 响应

    使用sanic.response模块中的函数来创建响应。

    4.1 纯文本

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

    4.2 HTML

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

    4.3 JSON

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

    4.4 文件

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

    4.5 Streaming

    from sanic import response
    
    @app.route("/streaming")
    async def index(request):
        async def streaming_fn(response):
            response.write('foo')
            response.write('bar')
        return response.stream(streaming_fn, content_type='text/plain')
    

    4.6 重定向

    from sanic import response
    
    @app.route('/redirect')
    def handle_request(request):
        return response.redirect('/json')
    

    4.7 元数据

    响应未编码的文本

    from sanic import response
    
    @app.route('/raw')
    def handle_request(request):
        return response.raw('raw data')
    

    4.8 修改标题或状态

    要修改标题或状态,请将标题或状态参数传递给这些函数:

    from sanic import response
    
    @app.route('/json')
    def handle_request(request):
        return response.json(
            {'message': 'Hello world!'},
            headers={'X-Served-By': 'sanic'},
            status=200
        )
    

    5. 静态文件

    静态文件和目录,例如图像文件是在Sanic创建app.static时候提供的。这个方法采用一个URL地址和一个文件名。然后通过给定的端点访问指定的文件。

    from sanic import Sanic
    app = Sanic(__name__)
    
    # Serves files from the static folder to the URL /static
    app.static('/static', './static')
    
    # Serves the file /home/ubuntu/test.png when the URL /the_best.png
    # is requested
    app.static('/the_best.png', '/home/ubuntu/test.png')
    
    app.run(host="0.0.0.0", port=8000)
    

    目前还无法利用url_for为静态文件创建一个URL地址。

    6. 异常

    异常可以从请求处理程序中抛出,并由Sanic自动处理。异常将消息作为第一个参数,也可以在HTTP响应中传回状态代码。

    6.1 抛出异常

    要抛出一个异常,只需要从sanic.exceptions中导入与raise相关的异常。

    from sanic.exceptions import ServerError
    
    @app.route('/killme')
    def i_am_ready_to_die(request):
        raise ServerError("Something bad happened", status_code=500)
    

    6.2 处理异常

    如果需要覆盖Sanic对异常的默认处理,就需要使用@app.exception装饰器。装饰器期望使用一个异常列表来处理参数。你可以传递一个SanicException来捕捉它们。装饰器异常处理函数必须使用RequestException对象来作为参数。

    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))
    

    6.3 有用的异常

    一些有用的异常如下:

    • NotFound: 找不到合适的路由请求。

    • ServerError: 服务器内部出现问题时调用。通常发生在用户代码出现错误的情况。

    7. 中间件和监听器

    中间件是在向服务器请求之前或之后执行的功能。它们可用来修改用户自定义处理函数的请求或响应。

    另外,Sanic提供程序监听器来运行应用程序生命周期中各个不同点的代码。

    7.1 中间件

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

    最简单的中间件不修改任何的请求或响应:

    @app.middleware('request')
    async def print_on_request(request):
        print("I print when a request is received by the server")
    
    @app.middleware('response')
    async def print_on_response(request, response):
        print("I print when a response is returned by the server")
    

    7.2 修改请求或响应

    只要请求或修改不返回任何值,中间件就可以修改给定的请求或响应。下面的示例即是一个简单的示范:

    app = Sanic(__name__)
    
    @app.middleware('response')
    async def custom_banner(request, response):
        response.headers["Server"] = "Fake-Server"
    
    @app.middleware('response')
    async def prevent_xss(request, response):
        response.headers["x-xss-protection"] = "1; mode=block"
    
    app.run(host="0.0.0.0", port=8000)
    

    上面的代码将按顺序应用两个中间件。首先,中间件custom_banner将HTTP响应头服务器更改为Fake-Server,第二个中间件prevent_xss将添加HTTP头以防止跨站点脚本(XSS)攻击。这两个函数在用户函数返回响应之后调用。

    如果中间件返回一个HTTPResponse对象,这个请求将停止处理并返回响应。如果这个请求在相关用户路由处理到达之前发生,则不会被调用该处理程序。返回的响应还会阻止进一步的中间件运行。

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

    7.3 监听器

    如果你想要在服务启动或关闭时执行启动/拆卸代码,可以使用以下的监听器:

    • before_server_start

    • after_server_start

    • before_server_stop

    • after_server_stop

    这些监听器在接收app对象和asyncio循环的函数上实现为装饰器。如下所示:

    @app.listener('before_server_start')
    async def setup_db(app, loop):
        app.db = await db_setup()
    
    @app.listener('after_server_start')
    async def notify_server_started(app, loop):
        print('Server successfully started!')
    
    @app.listener('before_server_stop')
    async def notify_server_stopping(app, loop):
        print('Server shutting down!')
    
    @app.listener('after_server_stop')
    async def close_db(app, loop):
        await app.db.close()
    

    如果你想要在循环开始后安排后台允许任务,则可以使用add_task方法轻松实现。

    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())
    

    8. 蓝图

    蓝图是可以用于应用程序中的子路由对象。蓝图不是向应用程序实例添加路由,而是定义了类似添加路由的方法,然后将路由以灵活且可插拔的方式注册到应用程序中。

    蓝图对于较大的应用程序十分有用,你可在逻辑上将应用程序分为几个组或责任领域。

    8.1 第一个蓝图

    下面显示了一个非常简单的蓝图,它在你的应用程序的根目录/下注册了一个处理函数。

    假设你将其保存到了my_blueprint.py文件中,则可以将其导入到你的主应用程序中。

    from sanic.response import json
    from sanic import Blueprint
    
    bp = Blueprint('my_blueprint')
    
    @bp.route('/')
    async def bp_root(request):
        return json({'my': 'blueprint'})
    

    8.2 注册蓝图

    蓝图必须在应用程序中注册。

    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)
    

    蓝图将添加到应用程序,并注册由该蓝图定义的任何路由。在此示例中,app.router中的注册路由将如下所示:

    [Route(handler=<function bp_root at 0x7f908382f9d8>, methods=None, pattern=re.compile('^/$'), parameters=[])]
    

    8.3 使用蓝图

    蓝图与应用程序有着大致相同的功能。

    8.3.1 WebSocket协议路由

    可以使用@bp.websocket装饰器或bp.add_websocket_route方法在蓝图上注册WebSocket处理程序。

    8.3.2 中间件

    使用蓝图可以在全局注册中间件。

    @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')
    

    8.3.3 异常

    只利用蓝图来应用全局的异常。

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

    8.3.4 静态文件

    静态文件可以在蓝图定义下提供给全局。

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

    8.4 启动和停止

    蓝图可以在服务启动或停止之前运行功能。如果多进程模式运行(超过1个进程),这些将在进程fork之后被触发。

    可用的事件是:

    • before_server_start: 服务开始接受连接之前执行

    • after_server_start: 服务开始接受连接后执行

    • before_server_stop: 服务停止接受连接之前执行

    • after_server_stop: 服务停止并且所有请求完成后执行

    bp = Blueprint('my_blueprint')
    
    @bp.listener('before_server_start')
    async def setup_connection(app, loop):
        global database
        database = mysql.connect(host='127.0.0.1'...)
    
    @bp.listener('after_server_stop')
    async def close_connection(app, loop):
        await database.close()
    

    8.5 用例:API版本控制

    Blueprints对于API版本控制非常有用,其中一个蓝图可能指向/v1/<routes>,另一个指向/v2/<routes>

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

    # blueprints.py
    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('Welcome to version 1 of our documentation')
    
    @blueprint_v2.route('/')
    async def api_v2_root(request):
        return text('Welcome to version 2 of our documentation')
    

    当在应用程序上注册蓝图时,路由/v1/v2现在将指向单个蓝图,这允许为每个API版本创建子站点。

    # main.py
    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)
    

    8.6 利用url_for生成URL

    如果希望为蓝图中的路由生成URL,请记住端点名称采用的格式<blueprint_name><handler_name>。例如:

    @blueprint_v1.route('/')
    async def root(request):
        url = app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5'
        return redirect(url)
    
    @blueprint_v1.route('/post/<post_id>')
    async def post_handler(request, post_id):
        return text('Post {} in Blueprint V1'.format(post_id))
    

    9. 配置

    任何复杂的应用程序都需要合理的配置。不同的环境或安装的设置可能不同。

    9.1 基本配置

    Sanic将配置保存在config应用程序对象的属性中。配置的是一个可以使用点运算进行修改或是类似字典类型的对象。

    app = Sanic('myapp')
    app.config.DB_NAME = 'appdb'
    app.config.DB_USER = 'appuser'
    

    由于配置的对象实际上是一个字典,你可以使用update方法来一次性设置几个值。

    db_settings = {
        'DB_HOST': 'localhost',
        'DB_NAME': 'appdb',
        'DB_USER': 'appuser'
    }
    app.config.update(db_settings)
    

    一般惯例是只有UPPERCASE配置参数。下面描述的用于仅查找类似于UPPERCASE参数加载配置的方法。

    9.2 加载配置

    有几种方式加载配置。

    9.2.1 从环境变量加载

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

    app = Sanic(load_vars=False)
    

    9.2.2 从对象加载

    如果有很多配置参数并且它们有合理的默认值,将它们放置于模块是有帮助的。

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

    你也可以使用类或者其它的对象类型。

    9.2.3 从文件加载

    通常情况下,你想要从文件中加载配置参数。你可以从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'
    

    9.3 内置的配置参数

    提供的几个预设值可以在创建应用程序的时候被覆盖:

    VariableDefaultDescription
    REQUEST_MAX_SIZE 100000000 How big a request may be (bytes)
    REQUEST_TIMEOUT 60 How long a request can take (sec)
    KEEP_ALIVE True Disables keep-alive when False

    10. Cookies

    Cookies是持续保存在用户浏览器中的数据片段。Sanic可以读取和写入Cookies,并以键值对的形式保存。

    10.1 读取Cookies

    可以通过Request对象的cookies字典访问访问用户的cookies。

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

    10.2 写入Cookies

    当返回一个响应时,可以在Response对象上设置Cookies。

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

    10.3 删除Cookies

    可以语义或明确地删除Cookies。

    from sanic.response import text
    
    @app.route("/cookie")
    async def test(request):
        response = text("Time to eat some cookies muahaha")
    
        # This cookie will be set to expire in 0 seconds
        del response.cookies['kill_me']
    
        # This cookie will self destruct in 5 seconds
        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
    

    响应的cookies可以设置为字典值,同时也有以下参数可用:

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

    11. 处理器装饰

    由于Sanic处理程序大都是简单的Python函数,因而你可以用类似Flask的方式对其进行装饰。典型的例子就是在你的执行程序之前运行一些你想运行的代码。

    11.1 装饰授权

    假设你需要检查用户是否有权访问特定的端点,你可以创建一个包装处理函数的装饰起,如果客户端有权访问资源,则检查请求,并发送适当的响应。

    from functools import wraps
    from sanic.response import json
    
    def authorized():
        def decorator(f):
            @wraps(f)
            async def decorated_function(request, *args, **kwargs):
                # run some method that checks the request
                # for the client's authorization status
                is_authorized = check_request_for_authorization_status(request)
    
                if is_authorized:
                    # the user is authorized.
                    # run the handler method and return the response
                    response = await f(request, *args, **kwargs)
                    return response
                else:
                    # the user is not authorized. 
                    return json({'status': 'not_authorized'}, 403)
            return decorated_function
        return decorator
    
    
    @app.route("/")
    @authorized()
    async def test(request):
        return json({status: 'authorized'})
    

    12. 流媒体

    12.1 请求流媒体

    如下所示,Sanic允许你以流的方式请求数据。当请求结束的时候,``request.stream.get()将返回None`值。只有post、put和patch装饰器有流的参数。

    from sanic import Sanic
    from sanic.views import CompositionView
    from sanic.views import HTTPMethodView
    from sanic.views import stream as stream_decorator
    from sanic.blueprints import Blueprint
    from sanic.response import stream, text
    
    bp = Blueprint('blueprint_request_stream')
    app = Sanic('request_stream')
    
    
    class SimpleView(HTTPMethodView):
    
        @stream_decorator
        async def post(self, request):
            result = ''
            while True:
                body = await request.stream.get()
                if body is None:
                    break
                result += body.decode('utf-8')
            return text(result)
    
    
    @app.post('/stream', stream=True)
    async def handler(request):
        async def streaming(response):
            while True:
                body = await request.stream.get()
                if body is None:
                    break
                body = body.decode('utf-8').replace('1', 'A')
                response.write(body)
        return stream(streaming)
    
    
    @bp.put('/bp_stream', stream=True)
    async def bp_handler(request):
        result = ''
        while True:
            body = await request.stream.get()
            if body is None:
                break
            result += body.decode('utf-8').replace('1', 'A')
        return text(result)
    
    
    async def post_handler(request):
        result = ''
        while True:
            body = await request.stream.get()
            if body is None:
                break
            result += body.decode('utf-8')
        return text(result)
    
    app.blueprint(bp)
    app.add_route(SimpleView.as_view(), '/method_view')
    view = CompositionView()
    view.add(['POST'], post_handler, stream=True)
    app.add_route(view, '/composition_view')
    
    
    if __name__ == '__main__':
        app.run(host='127.0.0.1', port=8000)
    

    12.2 响应流媒体

    Sanic允许你使用流的方法将内容以流的方式传输到客户端。当传递一个可以写入的StreamingHTTPResponse对象时,这个方法接受协程回调(coroutine callback)。一个简单的例子如下:

    from sanic import Sanic
    from sanic.response import stream
    
    app = Sanic(__name__)
    
    @app.route("/")
    async def test(request):
        async def sample_streaming_fn(response):
            response.write('foo,')
            response.write('bar')
    
        return stream(sample_streaming_fn, content_type='text/csv')
    

    在你想将以流的方式传递内容到外部服务的客户端(如数据库)的时候很有用。例如,你可以使用asyncpg提供的异步游标将数据库的记录以流的方式传递到客户端。

    @app.route("/")
    async def index(request):
        async def stream_from_db(response):
            conn = await asyncpg.connect(database='test')
            async with conn.transaction():
                async for record in conn.cursor('SELECT generate_series(0, 10)'):
                    response.write(record[0])
    
        return stream(stream_from_db)
    

    13. 基于类的视图

    基于类的视图只是为了实现对响应行为的请求的简单类。它们提供了在同一端点对不同HTTP请求类型进行区分处理的方法。

    端点可以分配一个基于类的视图,而不是定义和装饰三种不同的处理函数和一个用于每个端点的请求类型。

    13.1 定义视图

    基于类的视图是HTTPMethodView的子类。你可以为每个HTTP请求实现你想要的类方法。如果一个请求没有定义方法,一个405:Method not allowed的响应就会生成。

    要在端点上注册基于类的视图,就需要使用app.add_route方法。它的第一个参数是as_view方法定义的类,第二个参数是URL端点。

    可被使用的方法包括get,post,patch,putdelete方法。使用这些方法的方式如下所示:

    from sanic import Sanic
    from sanic.views import HTTPMethodView
    from sanic.response import text
    
    app = Sanic('some_name')
    
    class SimpleView(HTTPMethodView):
    
      def get(self, request):
          return text('I am get method')
    
      def post(self, request):
          return text('I am post method')
    
      def put(self, request):
          return text('I am put method')
    
      def patch(self, request):
          return text('I am patch method')
    
      def delete(self, request):
          return text('I am delete method')
    
    app.add_route(SimpleView.as_view(), '/')
    

    你也可以使用异步async语法。

    from sanic import Sanic
    from sanic.views import HTTPMethodView
    from sanic.response import text
    
    app = Sanic('some_name')
    
    class SimpleAsyncView(HTTPMethodView):
    
      async def get(self, request):
          return text('I am async get method')
    
    app.add_route(SimpleAsyncView.as_view(), '/')
    

    13.2 URL 参数

    如果你需要任何URL参数,就像路由章节介绍的一样,将其包含在方法定义中。

    class NameView(HTTPMethodView):
    
      def get(self, request, name):
        return text('Hello {}'.format(name))
    
    app.add_route(NameView.as_view(), '/<name>')
    

    13.3 装饰

    如果你想添加任何装饰器到类中,可以设置decorators类变量。当调用as_view方法的时候,会应用于类中。

    class ViewWithDecorator(HTTPMethodView):
      decorators = [some_decorator_here]
    
      def get(self, request, name):
        return text('Hello I have a decorator')
    
    app.add_route(ViewWithDecorator.as_view(), '/url')
    

    13.4 构造URL

    如果你希望为HTTPMethodView构造一个URL,类的名字将会当作url_for的端点。如下所示:

    @app.route('/')
    def index(request):
        url = app.url_for('SpecialClassView')
        return redirect(url)
    
    
    class SpecialClassView(HTTPMethodView):
        def get(self, request):
            return text('Hello from the Special Class View!')
    
    
    app.add_route(SpecialClassView.as_view(), '/special_class_view')
    

    13.5 使用组成视图(CompositionView)

    做为HTTPMethodView的替代方法,你可以在视图类外使用CompositionView来移动处理函数。

    处理函数在来源的每个HTTP方法中都进行了定义,然后使用CompositionView.add方法来添加视图。
    它的第一个参数应该是一个HTTP处理方法的列表(如['GET', 'POST']),第二个参数是处理函数。
    下面的例子展示了如何在CompositionView中使用外部处理函数和内联lanbda方法。

    from sanic import Sanic
    from sanic.views import CompositionView
    from sanic.response import text
    
    app = Sanic(__name__)
    
    def get_handler(request):
        return text('I am a get method')
    
    view = CompositionView()
    view.add(['GET'], get_handler)
    view.add(['POST', 'PUT'], lambda request: text('I am a post/put method'))
    
    # Use the new view to handle requests to the base URL
    app.add_route(view, '/')
    

    需要注意的是,当前你无法使用url_for方法为CompositionView构建一个URL。

    14. 定制协议

    你可以通过自定义一个协议来更改Sanic已经定义的协议行为,这个协议是asyncio.protocol的子类。这个协议能够以protocol关键字参数传递给sanic.run方法。

    自定义协议类的构造函数接受以下的关键字参数:

    • loop: asyncio兼容的事件循环;

    • connections: 存储协议的set对象。当Sanic接收SIGINTSIGTERM参数时,它会对集合中的所有协议对象执行protocol.close_if_idle

    • signal: 一个具有stop属性的sanic.server.Signal对象。当Sanic接收到SIGINTSIGTERM参数的时候,signal.stopped将被赋值为True

    • request_handler: 一个将sanic.request.Request对象和response回调为参数的协程程序。

    • error_handler: 一个当出现异常时被调用出的sanic.exceptions.Handler对象。

    • request_timeout: 请求超时前的秒数。

    • request_max_size: 指定最大的请求数,以字节为单位。

    14.1 例子

    如果处理函数没有返回HTTPResponse对象,那没在默认的协议中就会发生错误。

    通过复写write_response协议方法,如果处理程序返回一个字符串,它将被转换为一个HTTPResponse对象。

    from sanic import Sanic
    from sanic.server import HttpProtocol
    from sanic.response import text
    
    app = Sanic(__name__)
    
    
    class CustomHttpProtocol(HttpProtocol):
    
        def __init__(self, *, loop, request_handler, error_handler,
                     signal, connections, request_timeout, request_max_size):
            super().__init__(
                loop=loop, request_handler=request_handler,
                error_handler=error_handler, signal=signal,
                connections=connections, request_timeout=request_timeout,
                request_max_size=request_max_size)
    
        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(host='0.0.0.0', port=8000, protocol=CustomHttpProtocol)
    

    15. SSL范例

    SSLContext可以选择进行传递。

    import ssl
    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain("/path/to/cert", keyfile="/path/to/keyfile")
    
    app.run(host="0.0.0.0", port=8443, ssl=context)
    

    你还可以将证书和密钥的位置做为自检进行传递。

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

    16. 日志记录

    Sanic允许你使用python3 的logging API对请求不同类型日志进行记录(诸如访问记录,错误记录)。

    16.1 快速教程

    下面是一个使用默认设置的简单示例:

    from sanic import Sanic
    from sanic.config import LOGGING
    
    # The default logging handlers are ['accessStream', 'errorStream']
    # but we change it to use other handlers here for demo purpose
    LOGGING['loggers']['network']['handlers'] = [
        'accessSysLog', 'errorSysLog']
    
    app = Sanic('test')
    
    @app.route('/')
    async def test(request):
        return response.text('Hello World!')
    
    if __name__ == "__main__":
      app.run(log_config=LOGGING)
    

    如果需要关闭日志记录,只需要分配log_config=None:

    if __name__ == "__main__":
      app.run(log_config=None)
    

    这将在处理请求的时候跳过调用日志的功能,你可以加快在使用中的速度。

    if __name__ == "__main__":
      # disable internal messages
      app.run(debug=False, log_config=None)
    

    16.2 配置

    默认情况下,使用sanic.config.LOGGING字典来设置log_config参数,下面是handlers默认配置中设置的默认值:

    • internal: 使用logging.StreamHandler 内部信息在控制台输出。

    • accessStream: 使用logging.StreamHandler 登录控制台的请求信息。

    • errorStream: 使用logging.StreamHandler 控制台的错误信息和追溯信息。

    • accessSysLog: 使用logging.handlers.SysLogHandler 记录到syslog的请求信息。

    • errorSysLog: 使用logging.handlers.SysLogHandler syslog的错误消息和追溯记录。

    filters过滤:

    • accessFilter: 使用sanic.log.DefaultFilter 只允许DEBUGINFONONE(0)级别的过滤器。

    • errorFilter: 使用sanic.log.DefaultFilter 只允许在WARNINGERRORCRITICAL级别的过滤器。

    sanic中使用了两种loggers,如果要创建自己的日志记录配置,则必须对它们进行定义:

    • sanic: 记录内部信息。

    • network: 记录来自网络请求,以及请求中的任何信息。

    16.3 日志格式

    除了由python(asctime,levelname,message)提供的默认参数之外,Sanic还为accessFilter提供了网络记录器的其他参数:

    • host (str) request.ip

    • request (str) request.method + " " + request.url

    • status (int) response.status

    • byte (int) len(response.body)

    默认访问日志格式为:

    %(asctime)s - (%(name)s)[%(levelname)s][%(host)s]: %(request)s %(message)s %(status)d %(byte)d
    

    17. 测试

    Sanic端点可以使用test_client对象进行本地测试,test_client对象依赖与aiohttp库。

    test_client展示了如何在你的app使用get,post,put,delete,patch,headoptions方法。一个简单的例子(使用pytest)如下:

    # Import the Sanic app, usually created with Sanic(__name__)
    from external_server import app
    
    def test_index_returns_200():
        request, response = app.test_client.get('/')
        assert response.status == 200
    
    def test_index_put_not_allowed():
        request, response = app.test_client.put('/')
        assert response.status == 405
    

    在内部每调用一次test_client方法的时候,运行在127.0.01:42101的Sanic的app的测试请求将使用aiohttp执行测试。

    test_client方法接受以下的参数和关键字参数:

    • uri(默认'/'): 表示要测试的URL字符串。

    • gather_request(默认True): 布尔值,用于确定该函数是否返回原始请求,如果设置为True,则返回值是一个元组(request, response),如果为False则只返回响应。

    • server_kwargs(默认为{}): 在运行测试请求之前传递给app.run的附加参数。

    • debug(默认False): 布尔值,用户确定是否在调试模式下运行该服务。

    函数还将(request_args)和(*request_kwargs)直接传递给aiohttp ClientSession进行请求。

    例如,要向GET请求中提供数据,可以执行一下的操作:

    def test_get_request_includes_data():
        params = {'key1': 'value1', 'key2': 'value2'}
        request, response = app.test_client.get('/', params=params)
        assert request.args.get('key1') == 'value1'
    

    还有提供数据到 JSON POST请求中:

    def test_post_json_request_includes_data():
        data = {'key1': 'value1', 'key2': 'value2'}
        request, response = app.test_client.post('/', data=json.dumps(data))
        assert request.json.get('key1') == 'value1'
    

    17.1 pytest-sanic

    pytest-sanic是一个pytest插件,它可以帮助您异步测试代码。

    就像下面这样进行测试:

    async def test_sanic_db_find_by_id(app):
        """
        Let's assume that, in db we have,
            {
                "id": "123",
                "name": "Kobe Bryant",
                "team": "Lakers",
            }
        """
        doc = await app.db["players"].find_by_id("123")
        assert doc.name == "Kobe Bryant"
        assert doc.team == "Lakers"
    

    pytest-sanic还提供了一些有用的设置,如loopunused_porttest_servertest_client

    @pytest.yield_fixture
    def app():
        app = Sanic("test_sanic_app")
    
        @app.route("/test_get", methods=['GET'])
        async def test_get(request):
            return response.json({"GET": True})
    
        @app.route("/test_post", methods=['POST'])
        async def test_post(request):
            return response.json({"POST": True})
    
        yield app
    
    
    @pytest.fixture
    def test_cli(loop, app, test_client):
        return loop.run_until_complete(test_client(app, protocol=WebSocketProtocol))
    
    
    #########
    # Tests #
    #########
    
    async def test_fixture_test_client_get(test_cli):
        """
        GET request
        """
        resp = await test_cli.get('/test_get')
        assert resp.status == 200
        resp_json = await resp.json()
        assert resp_json == {"GET": True}
    
    async def test_fixture_test_client_post(test_cli):
        """
        POST request
        """
        resp = await test_cli.post('/test_post')
        assert resp.status == 200
        resp_json = await resp.json()
        assert resp_json == {"POST": True}
    

    18. 部署

    内建的Web服务器简化了Sanic的部署。在定义了一个sanic.Sanic的实例后,在调用run方法可以使用以下关键字参数:

    • 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的子类。

    18.1 进程

    默认情况下,Sanic在主进程中只侦听一个CPU内核。要启动其它核心,只需指定run参数中进程的数量。

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

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

    18.2 通过命令行运行

    如果你喜欢使用命令行参数,则可以通过执行模块启动Sanic服务器。例如,如果你将Sanic应用程序在名为server.py的文件中初始化,那么可以像这样运行服务:

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

    使用这种运行sanic的方法,没有必要在你的Python文件中调用app.run。如果需要这样做,请确保包装它,以便它由解释器直接执行。

    if __name__ == '__main__':
        app.run(host='0.0.0.0', port=1337, workers=4)
    

    18.3 通过Gunicorn运行

    Gunicorn'Green Unicorn'是用于UNIX的WSGI HTTP服务。

    使用Gunicorn运行Sanic应用程序,您需要使用特殊的sanic.worker.GunicornWorker对象定义Gunicornworker-class参数:

    gunicorn myapp:app --bind 0.0.0.0:1337 --worker-class sanic.worker.GunicornWorker
    

    如果您的应用程序遇到内存泄漏,您可以配置Gunicorn在处理给定数量的请求后,正常地重新启动一个工作。这可以帮助限制内存泄漏的影响。

    18.4 异步支持

    异步支持合适与其他应用程序(特别是loop)共享sanic进程。但是请注意,因为此方法不支持使用多个进程,一般不是运行应用程序的首选方式。

    下面是一个不完整的例子:

    server = app.create_server(host="0.0.0.0", port=8000)
    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(server)
    loop.run_forever()
    

    19. 扩展

    • Sessions: session的支持,允许使用redis,memcache或内存进行存储。

    • CORS: 用于处理跨域资源共享的扩展。

    • Compress: 允许您轻松地压缩Sanic响应。

    • Jinja2: Jinja2模板框架。

    • OpenAPI/Swagger:OpenAPI支持,以及Swagger UI。

    • Pagination: 简单的分页支持。

    • Motor: Simple motor wrapper。

    • Sanic CRUD:基于peewee 模型的CRUD(创建/检索/更新/删除)REST API自动生成的框架。

    • UserAgent: 添加user_agent到请求

    • Limiter: 限制sanic速率。

    • Sanic EnvConfig:将环境变量加入sanic配置。

    • Babel:借助Babel库,向Sanic应用程序添加i18n/l10n支持。

    • Dispatch: 由werkzeug的DispatcherMiddleware驱动的调度程序。可以作为Sanic-to-WSGI适配器。

    • Sanic-OAuth: 用于连接和创建自己的token授权的库。

    • Sanic-nginx-docker-example: 在nginx使用docker-compose的一个简单易用的Sanic例子。

    • sanic-graphql: Sanic的GraphQL集成。

    • sanic-prometheus: Sanic的Prometheus指标。

    • Sanic-RestPlus: Sanic的Flask-RestPlus端口。基于SwaggerUI的全功能REST API。

    • sanic-transmute: 可从python函数和类生成API,并自动生成Swagger UI文档。

    • pytest-sanic: 一个用于Sanic的pytest插件。可以测试异步代码。

    • jinja2-sanic:一个用于Sanic的jinja2模板渲染器。

    20. API参考

    http://sanic.readthedocs.io/en/latest/sanic/api_reference.html

    转载于:https://www.cnblogs.com/xiao-xue-di/p/10400433.html

    展开全文
  • sanic-prometheus:Sanic(异步python网络服务器)的Prometheus指标
  • sanic-app-源码

    2021-03-31 04:17:11
    sanic-app
  • heroku-sanic-源码

    2021-03-26 10:48:46
    heroku-sanic
  • sanic_norsk-源码

    2021-02-14 11:59:43
    sanic_norsk
  • Cookiecutter Sanic微服务 基于Sanic的RESTful微服务快速开发的模板[进行中] 要求 用法 $ cookiecutter gh:pathao-eng/cookiecutter-sanic-microservice 执照 GNU通用公共许可证
  • sanic-routing-源码

    2021-04-05 04:34:56
    Sanic路由 Sanic的内部处理程序路由从v21.3开始。
  • 本文实例讲述了Sanic框架Cookies操作。分享给大家供大家参考,具体如下: 简介 Sanic是一个类似Flask的Python 3.5+ Web服务器,它的写入速度非常快。除了Flask之外,Sanic还支持异步请求处理程序。这意味着你可以...
  • 本文实例讲述了Sanic框架配置操作。分享给大家供大家参考,具体如下: 简介 Sanic是一个类似Flask的Python 3.5+ Web服务器,它的写入速度非常快。除了Flask之外,Sanic还支持异步请求处理程序。这意味着你可以使用...
  • sanic 路由

    2018-01-31 19:01:27
    from sanic import Sanic from sanic.response import text @app.route('/tag/') async def person_handler(request, tag): return text('Tag - {}'.format(tag)) @app.route('/number/') asyn

    基本上和 flask 相似

    from sanic import Sanic
    from sanic.response import text
    
    @app.route('/tag/<tag>')
    async def person_handler(request, tag):
        return text('Tag - {}'.format(tag))
    
    @app.route('/number/<integer_arg:int>')
    async def person_handler(request, integer_arg):
        return text('Integer - {}'.format(integer_arg))
    
    @app.route('/number/<number_arg:number>')
    async def person_handler(request, number_arg):
        return text('Number - {}'.format(number))
    
    @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))
    展开全文
  • 关于sanic-cors 主页: : 软件包许可证:麻省理工学院 原料许可证: 简介:Sanic扩展添加了一个用于CORS支持的装饰器。 基于Cory Dolphin的长颈瓶。 当前构建状态 所有平台: 当前发行信息 姓名 资料下载 版本 ...
  • sanic 蓝图

    2018-01-31 19:43:28
    from sanic.response import json from sanic import Blueprint bp = Blueprint('my_blueprint') @bp.route('/') async def bp_root(): return json({'my': 'blueprint'}) # 蓝图注册到主 a
  • Sanic教程

    2019-09-27 01:58:39
    在安装Sanic之前,让我们一起来看看Python在支持异步的过程中,都经历了哪些比较重大的更新。 首先是Python3.4版本引入了asyncio,这让Python有了支持异步IO的标准库,而后3.5版本又提供了两个新的关键字async/...
  • Sanic-For-Pythoneer 这是一份记录文档,是我日常使用Sanic的一些经验之谈,希望能让你在使用Sanic的过程中少走一些弯路 可根据需求跳跃阅读 在线阅读 1.介绍 是一个可以使用 async/await 语法编写项目的异步非阻塞...
  • 本文实例讲述了Sanic框架应用部署方法。分享给大家供大家参考,具体如下: 简介 Sanic是一个类似Flask的Python 3.5+ Web服务器,它的写入速度非常快。除了Flask之外,Sanic还支持异步请求处理程序。这意味着你可以...
  • Sample Sanic Integration

    2020-11-29 05:08:04
    <div><p>Add example folder for Sanic webserver <h2>Description <p>Added a Sanic webserver example based off of the Quart example in an earlier PR Sanic lacks a CLI so the argument to <h2>Motivation ...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 1,130
精华内容 452
关键字:

sanic