1 Star 0 Fork 0

amiko / my-docsify-site

Create your Gitee Account
Explore and code with more than 12 million developers,Free private repositories !:)
Sign up
This repository doesn't specify license. Please pay attention to the specific project description and its upstream code dependency when using it.
Clone or Download
sanic.md 107.57 KB
Copy Edit Raw Blame History
胖蔡 authored 2021-11-26 22:35 . update docs.

关于

本指南是Python Sanic使用指南

安装

在开始之前,请确保同时拥有pip和Python 3.6版。Sanic使用新的async / await 语法,因此早期版本的python将无法使用。

如果您使用的是Fedora 28或更高版本的全新安装,请确保已redhat-rpm-config安装软件包,以防万一您想sanic与ujson依赖项一起使用。

pip3 install sanic

若是你不想在安装 sanic 的时候安装uvloop或者ujson,可以通过设置一个或多个环境变量,类似 SANIC_NO_X(X = UVLOOP / UJSON) 设置其为true来控制功能模块是否需要,操作命令如下:

SANIC_NO_UVLOOP=true SANIC_NO_UJSON=true pip3 install sanic

你也可以从conda-forge 安装Sanic:

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

创建一个Example

创建一个main.py文件

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)

运行服务


python main.py

查看浏览器

在浏览器中打开地址http://0.0.0.0:8000。您应该看到消息Hello world!。您现在已经可以使用Sanic服务器了!

参数配置

任何相当复杂的应用程序都需要未包含在实际代码中的配置。对于不同的环境或安装,设置可能会有所不同。

基础配置

Sanic将配置保存在应用程序对象的config属性中。配置对象仅仅是可以使用点符号或字典进行修改的对象:

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

由于config对象实际上是一个字典,因此可以使用它的update方法来一次设置多个值:

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

通常,惯例是仅具有大写配置参数。下文描述的用于加载配置的方法仅查找此类大写参数。

配置加载

Sanic 可以通过以下几种方式来加载配置项:

  • 通过系统环境变量加载 任何使用SANIC_前缀定义的变量都将应用于sanic配置。例如,设置SANIC_REQUEST_TIMEOUT将由应用程序自动加载并馈入REQUEST_TIMEOUT配置变量。您可以将不同的前缀传递给Sanic:
app = Sanic(load_env='MYAPP_')

那么上面的变量将是MYAPP_REQUEST_TIMEOUT。如果要禁用从环境变量加载,可以将其设置为 False

app = Sanic(load_env=False)
  • 通过python对象加载 如果有很多配置值,并且它们具有合理的默认值,则将它们放入模块中可能会有所帮助:
import myapp.default_settings

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

或者也可以通过配置路径:

app = Sanic('myapp')
app.config.from_object('config.path.config.Class')
  • 通过文件加载

通常,您会希望从不属于分布式应用程序的文件中加载配置。您可以使用from_pyfile(/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文件,为了加载它们而执行该文件。这使您可以使用任意逻辑来构建正确的配置。仅将大写变量添加到配置中。最常见的配置包括简单的键值对:

# config_file
DB_HOST = 'localhost'
DB_NAME = 'appdb'
DB_USER = 'appuser'
  • 内置的配置值

开箱即用,只有几个预定义的值可以在创建应用程序时覆盖。

变量 默认 描述
REQUEST_MAX_SIZE 100000000 请求的大小(字节)
REQUEST_BUFFER_QUEUE_SIZE 100 请求流缓冲区队列大小
REQUEST_TIMEOUT 60 请求到达需要多长时间(秒)
RESPONSE_TIMEOUT 60 响应需要花费多长时间(秒)
KEEP_ALIVE True False时禁用保持活动状态
KEEP_ALIVE_TIMEOUT 5 保持TCP连接打开的时间(秒)
GRACEFUL_SHUTDOWN_TIMEOUT 15.0 强制关闭非空闲连接的等待时间(秒)
ACCESS_LOG True 禁用或启用访问日志
PROXIES_COUNT -1 应用程序前面的代理服务器的数量(例如nginx;请参见下文)
FORWARDED_FOR_HEADER “X-Forwarded-For” 包含客户端和代理ip的“ X-Forwarded-For” HTTP标头的名称
REAL_IP_HEADER “X-Real-IP” 包含真实客户端ip的“ X-Real-IP” HTTP标头的名称

不同的超时变量

REQUEST_TIMEOUT

请求超时测量从新的打开的TCP连接传递到Sanic后端服务器到接收到整个HTTP请求之间的时间。如果花费的时间超过 REQUEST_TIMEOUT值(以秒为单位),则将其视为客户端错误,因此Sanic会生成HTTP 408响应并将其发送给客户端。如果您的客户端通常传递非常大的请求有效负载或非常缓慢地上传请求,则将此参数的值设置得更高。

RESPONSE_TIMEOUT

响应超时用于衡量Sanic服务器将HTTP请求传递给Sanic App的瞬间与HTTP响应被发送给客户端的瞬间之间的持续时间。如果花费的时间超过RESPONSE_TIMEOUT 值(以秒为单位),则将其视为服务器错误,因此Sanic会生成HTTP 503响应并将其发送给客户端。如果您的应用程序可能需要长时间运行以延迟响应的生成,请将此参数的值设置得更高。

KEEP_ALIVE_TIMEOUT

什么是保持生命?保持活动超时值有什么作用?

Keep-Alive是HTTP 1.1中引入的HTTP功能。发送HTTP请求时,客户端(通常是Web浏览器应用程序)可以设置Keep-Alive标头,以指示http服务器(Sanic)在发送响应后不关闭TCP连接。这使客户端可以重用现有的TCP连接以发送后续的HTTP请求,并确保客户端和服务器的网络通信效率更高。

默认情况下,KEEP_ALIVE配置变量在Sanic中设置为True。如果您的应用程序中不需要此功能,请将其设置为False,以使所有客户端连接在发送响应后立即关闭,无论请求上的Keep-Alive标头如何。

服务器保持TCP连接打开的时间由服务器本身决定。在Sanic中,使用KEEP_ALIVE_TIMEOUT值配置该值。默认情况下,它设置为5秒。这是与Apache HTTP服务器相同的默认设置,并且在允许客户端有足够的时间发送新请求和不立即打开太多连接之间取得了很好的平衡。不要超过75秒,除非您知道您的客户端正在使用支持TCP连接保持打开状态这么长时间的浏览器。

以供参考:

  • Apache httpd服务器默认keepalive超时= 5秒
  • Nginx服务器默认Keepalive超时= 75秒
  • Nginx性能调整准则使用keepalive = 15秒
  • IE(5-9)客户端硬keepalive限制= 60秒
  • Firefox客户端硬keepalive限制= 115秒
  • Opera 11客户端硬性保持活动限制= 120秒
  • Chrome 13+客户端Keepalive限制> 300+秒

代理配置

当您使用反向代理服务器(例如nginx)时,request.ip的值将包含代理的ip,通常为127.0.0.1。Sanic可以配置为使用代理标头来确定真实的客户端IP,可作为request.remote_addr使用。完整的外部URL也会从标头字段构建(如果可用)。

如果没有适当的预防措施,恶意客户端可能会使用代理标头欺骗其自身的IP。为避免此类问题,除非明确启用,Sanic不会使用任何代理标头。

反向代理服务的背后必须配置FORWARDED_SECRET,REAL_IP_HEADER和/或PROXIES_COUNT。

转发头

将FORWARDED_SECRET设置为目标代理使用的标识符。该机密用于安全地标识特定的代理服务器。给定上述标头,密码Pr0xy将使用第一行的信息,密码_1234proxy将使用第二行的信息。机密必须完全匹配secret或by的值。by by中的secret 必须以下划线开头,并且只能使用RFC 7239第6.3节中指定的字符 ,而secret没有这种限制。

Sanic会忽略没有密钥的任何元素,如果未设置密钥,甚至不会解析标头。

一旦找到受信任的转发元素,所有其他代理标头将被忽略,因为它已经包含有关客户端的完整信息。

传统代理头

  • 将REAL_IP_HEADER设置为x-real-ip,true-client-ip,cf-connecting-ip或此类标头的其他名称。
  • 将PROXIES_COUNT设置为x-forwarded-for(可通过FORWARDED_FOR_HEADER配置的名称)中预期的条目数。

如果通过这些方法之一找到了客户端IP,Sanic会对URL部分使用以下标头:

  • x-forward-proto,x-forwarded-host,x-forwarded-port,x-forwarded-path以及(如有必要)x-scheme。

如果使用代理配置...

  • 支持转发的代理:将FORWARDED_SECRET设置为代理在标头中插入的值

Apache Traffic Server:CONFIG proxy.config.http.insert_forwarded STRING for | proto | host | by = _secret
NGHTTPX:nghttpx –add-forwarded = for,proto,host,by –forwarded-for = ip –forwarded-by = _secret
NGINX:按照官方指示,在您的配置中添加任何位置:

  • 具有客户端IP的自定义标头:将REAL_IP_HEADER设置为该标头的名称
  • x-forwarded-for:将单个代理的PROXIES_COUNT设置为1,或将其设置为更大的值以允许Sanic选择正确的IP
  • 无需代理:无需配置!

Sanic 19.9中的更改

早期的Sanic版本具有不安全的默认设置。从19.9开始,必须手动设置代理设置,并且已删除了对否定PROXIES_COUNT的支持。

日志

Sanic允许您基于python3日志记录API对请求执行不同类型的日志记录(访问日志,错误日志)。如果要创建新配置,则应具有有关python3日志记录的一些基本知识。

快速入门

使用默认设置的一个简单示例如下所示:

from sanic import Sanic
from sanic.log import logger
from sanic.response import text

app = Sanic('test')

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

服务器运行后,您可以看到一些消息如下:

[2018-11-06 21:16:53 +0800] [24622] [INFO] Goin' Fast @ http://127.0.0.1:8000
[2018-11-06 21:16:53 +0800] [24667] [INFO] Starting worker [24667]

您可以将请求发送到服务器,它将打印日志消息:

[2018-11-06 21:18:53 +0800] [25685] [INFO] Here is your log
[2018-11-06 21:18:53 +0800] - (sanic.access)[INFO][127.0.0.1:57038]: GET http://localhost:8000/  200 12

要使用自己的日志记录配置,只需使用 logging.config.dictConfig,或log_config在初始化Sanic应用程序时通过:

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

要关闭日志记录,只需分配access_log = False:

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

处理请求时,这将跳过调用日志记录功能。您甚至可以进一步进行生产以提高速度:

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

配置

默认情况下,log_config参数设置为使用 sanic.log.LOGGING_CONFIG_DEFAULTS字典进行配置。

loggersSanic中使用了三种,如果要创建自己的日志记录配置,则必须定义以下三种:

记录仪名称 用例
sanic.root 用于记录内部消息。
sanic.error 用于记录错误日志。
sanic.access 用于记录访问日志。

日志格式

除了由python提供默认参数(asctime, levelname,message),Sanic也提供了用于访问记录器与另外的参数:

日志上下文参数 参数值 数据类型
host request.ip str
request request.method +”” + request.url str
status response.status 整型
byte len(response.body) 整型

默认访问日志格式为:

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

请求数据

当端点接收到HTTP请求时,将向路由功能传递一个 Request对象。

以下变量可以作为Request对象的属性访问:

  • json(any)-JSON正文
from sanic.response import json

@app.route("/json")
def post_json(request):
  return json({ "received": True, "message": request.json })
  • args(dict)-查询字符串变量。查询字符串是URL中类似于的部分?key1=value1&key2=value2

如果要解析该URL,则args词典将类似于{'key1':['value1'],'key2':['value2']}}。请求的query_string变量保存未解析的字符串值。属性提供默认的解析策略。如果要更改它,请查看下面的部分(更改queryset的默认解析规则)。


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 })
  • query_args(列表)-在许多情况下,您将需要以较少打包的形式访问url参数。query_args是(键,值)元组的列表。

属性提供默认的解析策略。如果要更改它,请查看下面的部分(更改queryset的默认解析规则)。对于相同的先前URL queryset ?key1 = value1&key2 = value2,query_args列表类似于[('key1','value1'),('key2','value2')]。如果多个参数具有相同的键,例如?key1 = value1&key2 = value2&key1 = value3,则query_args列表看起来像 [('key1','value1'),('key2','value2'),('key1 ','value3')]。

queryset的Request.args和Request.query_args之间的区别?key1 = value1&key2 = value2&key1 = value3

 from sanic import Sanic
  from sanic.response import json

  app = Sanic(__name__)


  @app.route("/test_request_args")
  async def test_request_args(request):
    return json({
        "parsed": True,
        "url": request.url,
        "query_string": request.query_string,
        "args": request.args,
        "raw_args": request.raw_args,
        "query_args": request.query_args,
    })

  if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)
{
    "parsed":true,
    "url":"http:\/\/0.0.0.0:8000\/test_request_args?key1=value1&key2=value2&key1=value3",
    "query_string":"key1=value1&key2=value2&key1=value3",
    "args":{"key1":["value1","value3"],"key2":["value2"]},
    "raw_args":{"key1":"value1","key2":"value2"},
    "query_args":[["key1","value1"],["key2","value2"],["key1","value3"]]
}
  • raw_args仅包含key1的第一项。在将来的版本中将不推荐使用。
  • 文件(“ 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 })
  • 表格(dict)-发布的表格变量
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)
  • method(str)-请求的HTTP方法(即GET,POST)。
  • ip(str)-请求者的IP地址。
  • port(str)-请求者的端口地址。
  • 套接字(元组)-请求者的(IP,端口)。
  • 应用程序 -到正在处理这一请求的应用中信高科浙江对象的引用。当在无法访问全局应用程序对象的模块中的蓝图或其他处理程序内部时,此功能很有用。
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//localhost8000/posts/1/?foo=bar
  • scheme:与请求关联的URL方案:“ http | https | ws | wss”或标头提供的任意值。
  • host:与请求关联的主机(位于Host标头中):localhost:8080
  • server_name:服务器的主机名,不带端口号。值按以下顺序搜索:config.SERVER_NAME,x-forwarded-host标头,Request.host()
  • server_port:类似于server_name。按以下顺序搜索:x-forwarded-port标头,Request.host()传输层套接字使用的实际端口。
  • path:请求的路径:/posts/1/
  • query_string:请求的查询字符串:foo= ar或空白字符串”
  • uri_template:用于匹配路由处理程序的模板:/ posts / /
  • token:授权标头的值:Basic YWRtaW46YWRtaW4 =
  • url_for:就像sanic.Sanic.url_for一样,但是会根据请求自动确定方案和netloc。由于这种方法的目标是产生正确的模式及netloc,_external是隐含的。

更改默认解析查询集的规则

在args和query_args属性内部使用的默认参数来解析queryset:

  • keep_blank_values(布尔):假 -标志表示是否在百分比编码查询空白值应该为空白串进行处理。真值表示应将空格保留为空白字符串。默认的false值指示将忽略空白值并将其视为未包含空白值。
  • strict_parsing(bool):假 -指示如何处理解析错误的标志。如果为false(默认值),错误将被忽略。如果为true,则错误引发ValueError异常。
  • 编码和错误(str):'utf-8'和'replace'-指定如何将百分比编码的序列解码为Unicode字符,这是bytes.decode()方法所接受的。

如果要更改默认参数,可以使用新值调用get_args和get_query_args方法。 对于queryset /?test1=value1&test2=&test3=value3:

from sanic.response import json

@app.route("/query_string")
def query_string(request):
    args_with_blank_values = request.get_args(keep_blank_values=True)
    return json({
        "parsed": True,
        "url": request.url,
        "args_with_blank_values": args_with_blank_values,
        "query_string": request.query_string
    })

输出结果:

{
    "parsed": true,
    "url": "http:\/\/0.0.0.0:8000\/query_string?test1=value1&test2=&test3=value3",
    "args_with_blank_values": {"test1": ["value1"], "test2": "", "test3": ["value3"]},
    "query_string": "test1=value1&test2=&test3=value3"
}

使用get和getlist访问值

该request.args中收益的子类字典称为requestParameters的。使用此对象时的主要区别是get和getlist方法之间的区别。

  • get(key,default = None)正常运行,除了当给定键的值是列表时,仅返回第一项。
  • getlist(key,default = None)正常运行,返回整个list。
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']

request.endpoint

该request.endpoint属性保存处理程序的名称。例如,以下路线将返回“ hello”

from sanic.response import text
from sanic import Sanic

app = Sanic()

@app.get("/")
def hello(request):
    return text(request.endpoint)

或者,使用蓝图将包括两者,并以句点分隔。例如,以下路由将返回foo.bar:

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


app = Sanic(__name__)
blueprint = Blueprint('foo')

@blueprint.get('/')
async def bar(request):
    return text(request.endpoint)

app.blueprint(blueprint)

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

响应数据

使用sanic.response模块中的函数来创建响应。响应数据类型如下:

文本

from sanic import response


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

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!'})

File文件类型

from sanic import response


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

Stream(流)


from sanic import response

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

重定向

from sanic import response


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

原始数据

无需编码身体即可响应

from sanic import response


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

空数据

用于响应RFC 2616定义的空消息

缓存

Cookies是保留在用户浏览器中的数据。Sanic可以读取和写入cookie,它们都存储为键值对。

客户可以自由更改Cookies。因此,您不能仅将登录信息之类的数据原样存储在cookie中,因为它们可以由客户端自由更改。为确保客户端不会伪造或篡改您存储在Cookie中的数据,请使用类似危险的内容对数据进行加密签名。

读操作

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

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

写操作

返回响应时,可以在Response对象上设置cookie 。

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

删除cookie

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

响应cookie可以像字典值一样设置,并具有以下参数可用:

  • expires (日期时间):Cookie在客户端浏览器上过期的时间。
  • path(字符串):此Cookie应用于的URL的子集。默认为/。
  • comment (字符串):注释(元数据)。
  • domain(字符串):指定Cookie对其有效的域。明确指定的域必须始终以点开头。
  • max-age (数字):Cookie应存活的秒数。
  • secure (布尔值):指定是否仅通过HTTPS发送cookie。
  • httponly (布尔值):指定Javascript是否无法读取cookie。

路由

路由允许用户为不同的URL端点指定处理程序功能。

基本路线如下所示,其中app是Sanic类的实例 :

from sanic.response import json

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

当访问URL http://server.url/(服务器的基本URL)时,路由器将最后一个/与处理函数test匹配,然后返回JSON对象。

Sanic处理程序函数必须使用async def语法定义,因为它们是异步函数。

请求参数

Sanic带有支持请求参数的基本路由器。

要指定参数,请在其周围加上如下的引号:。请求参数将作为关键字参数传递给路由处理程序函数。

from sanic.response import text

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

要为参数指定类型,请在参数名称后的引号内添加:type。如果参数与指定的类型不匹配,Sanic将抛出NotFound异常,从而导致URL上出现404:找不到页面错误。

支持的类型

  • str -- “Bob”

-- “Python 3”

  • int -- 10

-- 20

                                                                                                                                                                                                                                                                                                          -- 30

-- -10

  • number

-- 1

-- 1.5

-- 10

-- -10

  • alpha

-- “Bob”

-- “Python”

  • path

-- “hello”

-- “hello.text”

-- “hello world”

  • uuid

-- 123a123a-a12a-1a1a-a1a1-1a12a1a12345 支持UUIDv4)

  • 正则表达式 如果未设置任何类型,则应为字符串。赋予函数的参数将始终是字符串,与类型无关。
from sanic.response import text

@app.route('/string/<string_arg:string>')
async def string_handler(request, string_arg):
    return text('String - {}'.format(string_arg))

@app.route('/int/<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('/alpha/<alpha_arg:alpha>')
async def number_handler(request, alpha_arg):
    return text('Alpha - {}'.format(alpha_arg))

@app.route('/path/<path_arg:path>')
async def number_handler(request, path_arg):
    return text('Path - {}'.format(path_arg))

@app.route('/uuid/<uuid_arg:uuid>')
async def number_handler(request, uuid_arg):
    return text('Uuid - {}'.format(uuid_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))

str不是有效的类型标记。如果要str识别,则必须使用字符串

HTTP请求类型

默认情况下,URL上定义的路由仅可用于对该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))

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

URL建议使用url_for

Sanic提供了url_for方法,用于根据处理程序方法名称生成URL。如果您要避免将url路径硬编码到应用程序中,这将很有用;相反,您可以仅引用处理程序名称。例如:

from sanic.response import redirect

@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
  • 也有一些特殊的参数(_anchor,_external,_scheme,_method,_SERVER传给)url_for将有特殊的URL建设(_method不是现在支持,将被忽略)。例如:
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 you to pass an argument _server or set SERVER_NAME in app.config if not 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 at once
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。

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方法代替装饰器:

async def feed(request, ws):
    pass

app.add_websocket_route(my_websocket_handler, '/feed')

调用请求的第一个参数调用WebSocket路由的处理程序,第二个参数调用WebSocket协议对象。协议对象具有send 和recv方法,分别用于发送和接收数据。

WebSocket支持需要 Aymeric Augustin 的websockets软件包。

strict_slashes

您可以将路由设置为严格禁止尾随斜杠,这是可配置的。

# provide default strict_slashes value for all routes
app = Sanic('test_route_strict_slash', strict_slashes=True)

# you can also overwrite strict_slashes value for specific route
@app.get('/get', strict_slashes=False)
def handler(request):
    return text('OK')

# It also works for blueprints
bp = Blueprint('test_bp_strict_slash', strict_slashes=True)

@bp.get('/bp/get', strict_slashes=False)
def handler(request):
    return text('OK')

app.blueprint(bp)

strict_slashes标志如何遵循已定义的层次结构的行为,该层次结构决定特定的路由是否属于strict_slashes行为。

路线/ ├──蓝图/ ├──应用/ 上面的层次结构定义了strict_slashes标志的行为。 以上述顺序找到的strict_slashes的第一个非None值将应用于有问题的路由。

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

app = Sanic("sample_strict_slashes", strict_slashes=True)

@app.get("/r1")
def r1(request):
    return text("strict_slashes is applicable from App level")

@app.get("/r2", strict_slashes=False)
def r2(request):
    return text("strict_slashes is not applicable due to  False value set in route level")

bp = Blueprint("bp", strict_slashes=False)

@bp.get("/r3", strict_slashes=True)
def r3(request):
    return text("strict_slashes applicable from blueprint route level")

bp1 = Blueprint("bp1", strict_slashes=True)

@bp.get("/r4")
def r3(request):
    return text("strict_slashes applicable from blueprint level")

用户定义的路由名称

可以通过在注册路由时传递名称参数来使用自定义路由名称,该参数将覆盖使用handler .__ name__属性生成的默认路由名称。

app = Sanic('test_named_route')

@app.get('/get', name='get_handler')
def handler(request):
    return text('OK')

# then you need use `app.url_for('get_handler')`
# instead of # `app.url_for('handler')`

# It also works for blueprints
bp = Blueprint('test_named_bp')

@bp.get('/bp/get', name='get_handler')
def handler(request):
    return text('OK')

app.blueprint(bp)

# then you need use `app.url_for('test_named_bp.get_handler')`
# instead of `app.url_for('test_named_bp.handler')`

# different names can be used for same url with different methods

@app.get('/test', name='route_test')
def handler(request):
    return text('OK')

@app.post('/test', name='route_post')
def handler2(request):
    return text('OK POST')

@app.put('/test', name='route_put')
def handler3(request):
    return text('OK PUT')

# below url are the same, you can use any of them
# '/test'
app.url_for('route_test')
# app.url_for('route_post')
# app.url_for('route_put')

# for same handler name with different methods
# you need specify the name (it's url_for issue)
@app.get('/get')
def handler(request):
    return text('OK')

@app.post('/post', name='post_handler')
def handler(request):
    return text('OK')

# then
# app.url_for('handler') == '/get'
# app.url_for('post_handler') == '/post'
生成静态文件的
Sanic支持使用url_for方法构建静态文件url如果静态url指向目录则可以忽略url_for的filename参数q

app = Sanic('test_static')
app.static('/static', './static')
app.static('/uploads', './uploads', name='uploads')
app.static('/the_best.png', '/home/ubuntu/test.png', name='best_png')

bp = Blueprint('bp', url_prefix='bp')
bp.static('/static', './static')
bp.static('/uploads', './uploads', name='uploads')
bp.static('/the_best.png', '/home/ubuntu/test.png', name='best_png')
app.blueprint(bp)

# then build the url
app.url_for('static', filename='file.txt') == '/static/file.txt'
app.url_for('static', name='static', filename='file.txt') == '/static/file.txt'
app.url_for('static', name='uploads', filename='file.txt') == '/uploads/file.txt'
app.url_for('static', name='best_png') == '/the_best.png'

# blueprint url building
app.url_for('static', name='bp.static', filename='file.txt') == '/bp/static/file.txt'
app.url_for('static', name='bp.uploads', filename='file.txt') == '/bp/uploads/file.txt'
app.url_for('static', name='bp.best_png') == '/bp/static/the_best.png'

静态文件

Sanic在使用app.static()方法注册时会提供静态文件和目录,例如图像文件。该方法采用端点URL和文件名。然后可以通过给定的端点访问指定的文件。

from sanic import Sanic
from sanic.blueprints import Blueprint

app = Sanic(__name__)

# Serves files from the static folder to the URL /static
app.static('/static', './static')
# use url_for to build the url, name defaults to 'static' and can be ignored
app.url_for('static', filename='file.txt') == '/static/file.txt'
app.url_for('static', name='static', filename='file.txt') == '/static/file.txt'

# Serves the file /home/ubuntu/test.png when the URL /the_best.png
# is requested
app.static('/the_best.png', '/home/ubuntu/test.png', name='best_png')

# you can use url_for to build the static file url
# you can ignore name and filename parameters if you don't define it
app.url_for('static', name='best_png') == '/the_best.png'
app.url_for('static', name='best_png', filename='any') == '/the_best.png'

# you need define the name for other static files
app.static('/another.png', '/home/ubuntu/another.png', name='another')
app.url_for('static', name='another') == '/another.png'
app.url_for('static', name='another', filename='any') == '/another.png'

# also, you can use static for blueprint
bp = Blueprint('bp', url_prefix='/bp')
bp.static('/static', './static')

# specify a different content_type for your files
# such as adding 'charset'
app.static('/', '/public/index.html', content_type="text/html; charset=utf-8")

# servers the file directly
bp.static('/the_best.png', '/home/ubuntu/test.png', name='best_png')
app.blueprint(bp)

app.url_for('static', name='bp.static', filename='file.txt') == '/bp/static/file.txt'
app.url_for('static', name='bp.best_png') == '/bp/test_best.png'

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

注意:提供静态目录时,Sanic不提供目录索引。

虚拟主机

所述app.static()方法也支持虚拟主机。您可以通过host参数将静态文件与特定的虚拟主机一起使用。例如:

from sanic import Sanic

app = Sanic(__name__)

app.static('/static', './static')
app.static('/example_static', './example_static', host='www.example.com')

大流量文件

在某些情况下,您可能会使用Sanic服务器存储大文件(例如,视频,图像等)。您可以选择使用流式传输文件,而不是直接下载。

这是一个例子:

from sanic import Sanic

app = Sanic(__name__)

app.static('/large_video.mp4', '/home/ubuntu/large_video.mp4', stream_large_files=True)
当stream_large_files为True时Sanic将使用file_stream()而不是file()来提供静态文件这将使用1KB作为默认块大小而且如果需要您还可以使用自定义块大小例如

from sanic import Sanic

app = Sanic(__name__)

chunk_size = 1024 * 1024 * 8 # Set chunk size to 8KB
app.static('/large_video.mp4', '/home/ubuntu/large_video.mp4', stream_large_files=chunk_size)

蓝图配置

蓝图是可用于在应用程序内进行子路由的对象。蓝图定义了添加路由的相似方法,而不是将路由添加到应用程序实例,然后以灵活且可插入的方式向应用程序注册。 蓝图对于大型应用程序特别有用,在大型应用程序中,您的应用程序逻辑可以分为几个组或职责范围。

我的第一个蓝图

下面显示了一个非常简单的蓝图,该蓝图在应用程序的根目录/中注册了处理程序功能。 假设您将此文件另存为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'})
注册蓝图
蓝图必须在应用程序中注册

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=frozenset({'GET'}), pattern=re.compile('^/$'), parameters=[], name='my_blueprint.bp_root', uri='/')]

蓝图组和嵌套

蓝图也可以作为列表或元组的一部分进行注册,其中注册服务商将递归循环遍历任何蓝图的子序列,并进行相应的注册。所述Blueprint.group提供方法来简化这个过程,允许模拟什么从前端观察的“模拟”后端目录结构。考虑以下(非常人为的)示例:

api /
├──内容/
│├──authors.py
│├──static.py
│└──__init__。py
├──info.py
└──__init__。py
app.py

该应用程序蓝图层次结构的初始化可以如下:


# api/content/authors.py
from sanic import Blueprint

authors = Blueprint('content_authors', url_prefix='/authors')
# api/content/static.py
from sanic import Blueprint

static = Blueprint('content_static', url_prefix='/static')
# api/content/__init__.py
from sanic import Blueprint

from .static import static
from .authors import authors

content = Blueprint.group(static, authors, url_prefix='/content')
# api/info.py
from sanic import Blueprint

info = Blueprint('info', url_prefix='/info')
# api/__init__.py
from sanic import Blueprint

from .content import content
from .info import info

api = Blueprint.group(content, info, url_prefix='/api')

现在可以像这样在app.py中注册这些蓝图:

# app.py
from sanic import Sanic

from .api import api

app = Sanic(__name__)

app.blueprint(api)

使用蓝图

蓝图具有与应用程序实例几乎相同的功能。

WebSocket路由

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

蓝图中间件

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

@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')
蓝图组中间件
使用此中间件将确保您可以将公共中间件应用于构成当前正在考虑的蓝图组的所有蓝图

bp1 = Blueprint('bp1', url_prefix='/bp1')
bp2 = Blueprint('bp2', url_prefix='/bp2')

@bp1.middleware('request')
async def bp1_only_middleware(request):
    print('applied on Blueprint : bp1 Only')

@bp1.route('/')
async def bp1_route(request):
    return text('bp1')

@bp2.route('/<param>')
async def bp2_route(request, param):
    return text(param)

group = Blueprint.group(bp1, bp2)

@group.middleware('request')
async def group_middleware(request):
    print('common middleware applied for both bp1 and bp2')

# Register Blueprint group under the app
app.blueprint(group)

异常

例外可以专门应用于全局蓝图。

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

静态文件 可以在蓝图前缀下全局提供静态文件。

# suppose bp.name == 'bp'

bp.static('/web/path', '/folder/to/serve')
# also you can pass name parameter to it for url_for
bp.static('/web/path', '/folder/to/server', name='uploads')
app.url_for('static', name='bp.uploads', filename='file.txt') == '/bp/web/path/file.txt'

启动和停止

蓝图可以在服务器的启动和停止过程中运行功能。如果以多处理器模式运行(超过1个工作程序),则在工作程序派生后触发这些操作。

可用的事件有:

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

用例:API版本控制

蓝图对于API版本控制非常有用,其中一个蓝图可能指向/ v1 / ,而另一个蓝图则指向/ v2 / 。

初始化蓝图时,它可以使用可选的version参数,该参数将放在蓝图中定义的所有路由之前。此功能可用于实现我们的API版本控制方案。

# blueprints.py
from sanic.response import text
from sanic import Blueprint

blueprint_v1 = Blueprint('v1', url_prefix='/api', version="v1")
blueprint_v2 = Blueprint('v2', url_prefix='/api', version="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 / api和/ v2 / api现在将指向各个蓝图从而可以 为每个API版本创建子站点

# main.py
from sanic import Sanic
from blueprints import blueprint_v1, blueprint_v2

app = Sanic(__name__)
app.blueprint(blueprint_v1)
app.blueprint(blueprint_v2)

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

URL建设与url_for

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

@blueprint_v1.route('/')
async def root(request):
    url = request.app.url_for('v1.post_handler', post_id=5) # --> '/v1/api/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))

版本控制

您可以将version关键字传递给路由装饰器或蓝图初始化器。这将产生v {version}网址前缀,其中{version}是版本号。

单个路由

您可以将版本号直接传递给路由。

from sanic import response


@app.route('/text', version=1)
def handle_request(request):
    return response.text('Hello world! Version 1')

@app.route('/text', version=2)
def handle_request(request):
    return response.text('Hello world! Version 2')

app.run(port=80)

可以通过curl获取:

curl localhost/v1/text
curl localhost/v2/text

全局蓝图版本

您也可以将版本号传递给蓝图,该版本号将应用于所有路线。

from sanic import response
from sanic.blueprints import Blueprint

bp = Blueprint('test', version=1)

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

curl 获取如下:

curl localhost/v1/html

异常处理

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

抛出异常

要抛出异常,只需从sanic.exceptions模块引发相关异常 。

from sanic.exceptions import ServerError

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

您还可以将中止功能与相应的状态代码一起使用:

from sanic.exceptions import abort
from sanic.response import text

@app.route('/youshallnotpass')
async def no_no(request):
        abort(401)
        # this won't happen
        text("OK")

处理异常

要覆盖Sanic对异常的默认处理,请使用@ app.exception 装饰器。装饰器期望将异常列表作为参数处理。您可以通过SanicException来捕获它们!装饰的异常处理程序函数必须将Request和Exception对象作为参数。

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

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

您还可以像这样添加异常处理程序:

from sanic import Sanic

async def server_error_handler(request, exception):
    return text("Oops, server error", status=500)

app = Sanic()
app.error_handler.add(Exception, server_error_handler)

在某些情况下,您可能希望向默认提供的功能中添加更多错误处理功能。在这种情况下,您可以这样子化Sanic的默认错误处理程序:

from sanic import Sanic
from sanic.handlers import ErrorHandler

class CustomErrorHandler(ErrorHandler):
    def default(self, request, exception):
        ''' handles errors that have no error handlers assigned '''
        # You custom error handling logic...
        return super().default(request, exception)

app = Sanic()
app.error_handler = CustomErrorHandler()

常见异常

下面列出了一些最有用的例外:

  • NotFound:在找不到适合该请求的路由时调用。
  • ServerError:在服务器内部出现问题时调用。如果用户代码中引发异常,通常会发生这种情况。 有关要抛出的异常的完整列表,请参见sanic.exceptions模块。

中间件

中间件是在对服务器的请求之前或之后执行的功能。它们可用于修改对 用户定义的处理函数的请求或响应。 此外,Sanic还提供了侦听器,使您可以在应用程序生命周期的各个阶段运行代码。

中间件

中间件有两种类型:请求和响应。两者都使用@ 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")
    

修改request和response

中间件可以修改给定的请求或响应参数,只要它不返回即可。以下示例显示了一个实际的用例。

app = Sanic(__name__)

@app.middleware('request')
async def add_key(request):
    # Arbitrary data may be stored in request context:
    request.ctx.foo = 'bar'

@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.get("/")
async def index(request):
    return sanic.response.text(request.ctx.foo)

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

三个中间件按顺序执行:

  • 第一个请求中间件add_key将新的密钥foo添加到请求上下文中。
  • 请求被路由到处理程序索引,该索引从上下文获取键并返回文本响应。
  • 第一个响应中间件custom_banner改变HTTP响应头服务器说假服务器
  • 第二个响应中间件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') 

自定义context

任意数据可以存储在request.ctx中。典型的用例是将从数据库中获取的用户对象存储在身份验证中间件中。在请求期间,所有以后的中间件以及处理程序都可以访问添加的密钥。

自定义上下文保留给应用程序和扩展。Sanic本身不使用它。

监听器

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

  • before_server_start
  • after_server_start
  • before_server_stop -after_server_stop

这些侦听器在接受应用程序对象以及异步循环的函数上作为装饰器实现。

示例:

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

也可以使用register_listener方法注册一个侦听器。如果您在实例化应用程序的模块之外的其他模块中定义了侦听器,这可能会很有用。

app = Sanic()

async def setup_db(app, loop):
    app.db = await db_setup()

app.register_listener(setup_db, 'before_server_start')

如果您希望安排后台任务在循环开始后运行,那么Sanic提供了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())

Sanic将尝试自动注入应用程序,并将其作为参数传递给任务:

async def notify_server_started_after_five_seconds(app):
    await asyncio.sleep(5)
    print(app.name)

app.add_task(notify_server_started_after_five_seconds)

或者,您可以显式地传递应用程序以达到相同的效果:

async def notify_server_started_after_five_seconds(app):
    await asyncio.sleep(5)
    print(app.name)

app.add_task(notify_server_started_after_five_seconds(app))

Websocket

Sanic在websockets之上提供了易于使用的抽象。Sanic支持websocket版本7和8。

设置WebSocket:

from sanic import Sanic
from sanic.response import json
from sanic.websocket import WebSocketProtocol

app = Sanic()

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

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, protocol=WebSocketProtocol)
    

或者,app.add_websocket_route可以使用该方法代替装饰器:

async def feed(request, ws):
    pass

app.add_websocket_route(feed, '/feed')

调用WebSocket路由的处理程序时,将请求作为第一个参数,将WebSocket协议对象作为第二个参数。协议对象具有send 和recv分别发送和接收数据的方法。

您可以通过设置自己的WebSocket配置app.config,例如

app.config.WEBSOCKET_MAX_SIZE = 2 ** 20
app.config.WEBSOCKET_MAX_QUEUE = 32
app.config.WEBSOCKET_READ_LIMIT = 2 ** 16
app.config.WEBSOCKET_WRITE_LIMIT = 2 ** 16

在本Configuration节中找到更多内容。

装饰器

处理程序装饰器

由于Sanic处理程序是简单的Python函数,因此可以以类似于Flask的方式将装饰器应用于它们。一个典型的用例是当您希望在执行处理程序的代码之前运行一些代码时。

授权装饰器

假设您要检查用户是否有权访问特定端点。您可以创建装饰器,该装饰器包装处理程序函数,检查请求是否授权客户端访问资源并发送适当的响应。

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

请求流

Sanic允许您按流获取请求数据,如下所示。请求结束时,等待request.stream.read()返回None。只有post,put和patch装饰器具有stream参数。

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.read()
            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.read()
            if body is None:
                break
            body = body.decode('utf-8').replace('1', 'A')
            await response.write(body)
    return stream(streaming)


@bp.put('/bp_stream', stream=True)
async def bp_put_handler(request):
    result = ''
    while True:
        body = await request.stream.read()
        if body is None:
            break
        result += body.decode('utf-8').replace('1', 'A')
    return text(result)


# You can also use `bp.add_route()` with stream argument
async def bp_post_handler(request):
    result = ''
    while True:
        body = await request.stream.read()
        if body is None:
            break
        result += body.decode('utf-8').replace('1', 'A')
    return text(result)

bp.add_route(bp_post_handler, '/bp_stream', methods=['POST'], stream=True)


async def post_handler(request):
    result = ''
    while True:
        body = await request.stream.read()
        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)

响应流

Sanic允许您使用stream方法将内容流式传输到客户端。此方法接受协程回调,该回调传递给写入的StreamingHTTPResponse对象。一个简单的例子如下:

from sanic import Sanic
from sanic.response import stream

app = Sanic(__name__)

@app.route("/")
async def test(request):
    async def sample_streaming_fn(response):
        await response.write('foo,')
        await 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)'):
                await response.write(record[0])

    return stream(stream_from_db)

如果客户端支持HTTP / 1.1,Sanic将使用分块传输编码;您可以使用流功能的分块选项显式启用或禁用它。

文件流

Sanic提供sanic.response.file_stream函数,该函数在您要发送大文件时很有用。它返回一个StreamingHTTPResponse对象,默认情况下将使用分块传输编码。因此,Sanic不在响应中添加Content-Length HTTP标头。如果要使用此标头,则可以禁用分块传输编码并手动添加它:

from aiofiles import os as async_os
from sanic.response import file_stream

@app.route("/")
async def index(request):
    file_path = "/srv/www/whatever.png"

    file_stat = await async_os.stat(file_path)
    headers = {"Content-Length": str(file_stat.st_size)}

    return await file_stream(
        file_path,
        headers=headers,
        chunked=False,
    )

视图

基于类的视图只是实现对请求的响应行为的类。它们提供了一种在同一个端点上分隔处理不同HTTP请求类型的方法。可以为终结点分配一个基于类的视图,而不是定义和修饰三个不同的处理函数,每个终结点用于一个端点支持的请求类型。

定义视图

基于类的视图应该子类HTTPMethodView。然后,您可以为要支持的每种HTTP请求类型实现类方法。如果收到没有定义方法的请求,则会生成405:方法不允许响应。 要在端点上注册基于类的视图,请使用app.add_route方法。第一个参数应该是使用方法as_view 调用的已定义类,第二个参数应该是URL端点。 可用的方法有get,post,put,patch和delete。使用所有这些方法的类如下所示。

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(), '/')

您还可以使用异步语法。

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(), '/')

URL参数

如路由指南中所述,如果您需要任何URL参数,请将其包含在方法定义中。

class NameView(HTTPMethodView):

  def get(self, request, name):
    return text('Hello {}'.format(name))

app.add_route(NameView.as_view(), '/<name>')

装饰器

如果要将任何装饰器添加到该类,则可以设置装饰器 类变量。这些将在调用as_view时应用于类。

class ViewWithDecorator(HTTPMethodView):
  decorators = [some_decorator_here]

  def get(self, request, name):
    return text('Hello I have a decorator')

  def post(self, request, name):
    return text("Hello I also have a decorator")

app.add_route(ViewWithDecorator.as_view(), '/url')

但是,如果您只想装饰一些功能而不是装饰所有功能,则可以执行以下操作:

class ViewWithSomeDecorator(HTTPMethodView):

    @staticmethod
    @some_decorator_here
    def get(request, name):
        return text("Hello I have a decorator")

    def post(self, request, name):
        return text("Hello I don't have any decorators")

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

使用CompositionView

作为HTTPMethodView的替代方法,可以使用CompositionView将处理程序函数移至视图类之外。

每个支持的HTTP方法的处理函数都在源代码的其他地方定义,然后使用CompositionView.add方法添加到视图中。第一个参数是要处理的HTTP方法的列表(例如['GET','POST']),第二个参数是处理函数。以下示例显示 了既有外部处理函数又有内联lambda的CompositionView用法:

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 。

自定义协议

这是高级用法,大多数读者将不需要这种功能。

您可以通过指定自定义协议来更改Sanic协议的行为,该协议应该是asyncio.protocol的子类。然后可以将该协议作为关键字参数传递protocol给该sanic.run方法。 定制协议类的构造函数从Sanic接收以下关键字参数。

  • loop:一个asyncio兼容的事件循环。
  • connections:set用于存储协议对象。当Sanic收到 SIGINT或时SIGTERM,它将protocol.close_if_idle对存储在此集中的所有协议对象执行。
  • signal:sanic.server.Signal具有stopped属性的对象。当Sanic收到SIGINT或时SIGTERM,signal.stopped被分配True。
  • request_handler:一个协程,将一个sanic.request.Request对象和一个response回调作为参数。
  • error_handler:sanic.exceptions.Handler引发异常时调用的a 。
  • request_timeout:请求超时之前的秒数。
  • request_max_size:一个整数,指定请求的最大大小,以字节为单位。

示例

如果处理程序函数不返回HTTPResponse对象,则默认协议中会发生错误。

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

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)

Socket

Sanic可以使用python 套接字模块来容纳非IPv4套接字。

IPv6示例:

from sanic import Sanic
from sanic.response import json
import socket

sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
sock.bind(('::', 7777))

app = Sanic()


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

if __name__ == "__main__":
    app.run(sock=sock)

测试IPv6:

 curl -g -6 "http://[::1]:7777/"

UNIX套接字示例:

import signal
import sys
import socket
import os
from sanic import Sanic
from sanic.response import json


server_socket = '/tmp/sanic.sock'

sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(server_socket)

app = Sanic()


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


def signal_handler(sig, frame):
        print('Exiting')
        os.unlink(server_socket)
        sys.exit(0)


if __name__ == "__main__":
    app.run(sock=sock)

测试UNIX:

 curl -v --unix-socket /tmp/sanic.sock http://localhost/hello

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)

调试模式

启用S​​anic的调试模式时,Sanic将提供更详细的日志记录输出,默认情况下将启用自动重新加载功能。

Sanic的更多调试功能会减慢服务器的性能,因此建议仅在开发环境中启用它。

设置调试模式

通过设置debug模式,将输出来自Sanic的更详细的输出,并且将激活自动重新加载器。

from sanic import Sanic
from sanic.response import json

app = Sanic()

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

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

手动设置自动重载

Sanic提供了一种手动启用或禁用自动重新加载器的方法,该auto_reload参数将激活或停用自动重新加载器。

from sanic import Sanic
from sanic.response import json

app = Sanic()

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

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000, auto_reload=True)

部署

使用以下三个选项之一,部署Sanic非常简单:内置Web服务器,ASGI Web服务器或gunicorn。将Sanic放在反向代理(如nginx)后面也很常见。

通过Sanic Web服务器运行 定义sanic.Sanic的实例之后,我们可以使用以下关键字参数调用run方法:

  • host (默认为“ 127.0.0.1”):托管服务器的地址。
  • 端口 (默认为8000):用于托管服务器的端口。
  • debug (默认为False):启用调试输出(降低服务器速度)。
  • ssl (默认为“无”):SSLContext用于对工作人员进行SSL加密。
  • sock (默认为None):服务器接受其连接的套接字。
  • worker (默认值为1):要产生的工作进程数。
  • 循环 (默认为“无”):异步兼容的事件循环。如果未指定,Sanic将创建其自己的事件循环。
  • 协议 (默认为HttpProtocol):asyncio.protocol的子类。
  • access_log (默认为True):启用登录以处理请求(显着降低服务器速度)。
app.run(host='0.0.0.0', port=1337, access_log=False)

在上面的示例中,我们决定关闭访问日志以提高性能。

worker

默认情况下,Sanic仅使用一个CPU内核侦听主进程。要提高效率,只需在运行参数中指定工作者数。

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

Sanic将自动启动多个进程并在它们之间路由流量。我们建议您使用尽可能多的核心。

通过命令运行

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

通过这种方式运行sanic,无需在Python文件中调用app.run。如果这样做,请确保将其包装起来,以便仅在由解释器直接运行时才执行。

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

通过ASGI运行

Sanic也符合ASGI。这意味着您可以使用首选的ASGI Web服务器来运行Sanic。ASGI的三个主要的实现是 达芙妮,Uvicorn和Hypercorn。

按照他们的文档来运行它们的正确方法,但是它看起来应该像这样:

daphne myapp:app
uvicorn myapp:app
hypercorn myapp:app

使用ASGI时要注意以下几点:

  1. 使用Sanic Web服务器时,Websockets将使用websockets软件包运行。在ASGI模式下,由于websocket是在ASGI服务器中管理的,因此不需要此软件包。
  2. ASGI 寿命协议https://asgi.readthedocs.io/en/latest/specs/lifespan.html仅支持两个服务器事件:启动和关闭。Sanic有四个:启动之前,启动之后,关闭之前和关闭之后。因此,在ASGI模式下,启动和关闭事件将连续运行,而实际上不会围绕服务器进程的开始和结束运行(因为现在由ASGI服务器控制)。因此,最好使用after_server_start和 before_server_stop。
  3. 从Sanic v19.6开始,ASGI模式仍处于“ beta”状态。

通过Gunicorn运行

Gunicorn'Green Unicorn'是用于UNIX的WSGI HTTP服务器。这是从Ruby的Unicorn项目移植来的前叉工作模型。

为了运行与Gunicorn中信高科应用程序,您需要使用特殊sanic.worker.GunicornWorker 为Gunicorn 工人类的说法:

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

如果您的应用程序遭受内存泄漏的困扰,您可以将Gunicorn配置为在处理了给定数量的请求之后正常重启工作器。这是帮助限制内存泄漏影响的便捷方法。

有关更多信息,请参见Gunicorn文档。

其他部署注意事项

在反向代理后面运行

Sanic可以与反向代理(例如nginx)一起使用。有一个简单的nginx配置示例:

server {
  listen 80;
  server_name example.org;
  location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  }
}

如果要获取真实的客户端ip,则应配置X-Real-IP和X-Forwarded-For HTTP标头,并将app.config.PROXIES_COUNT设置为1;有关更多信息,请参见配置页面。

禁用调试日志记录以提高性能

为了提高性能,请在运行参数中添加debug = False和access_log = False。

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

通过Gunicorn运行,您可以设置环境变量SANIC_ACCESS_LOG =“ False”

env SANIC_ACCESS_LOG="False" gunicorn myapp:app --bind 0.0.0.0:1337 --worker-class sanic.worker.GunicornWorker --log-level warning\

或者您可以直接重写应用程序配置

app.config.ACCESS_LOG = False

异步支持和共享环路

如果您需要与其他应用程序(特别是loop)共享Sanic进程,则此方法非常适合。但是,请注意,此方法不支持使用多个进程,并且通常不是运行该应用程序的首选方法。

这是一个不完整的示例(请参阅示例中的run_async.py了解更多实用信息):

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

注意:使用此方法,调用app.create_server()将触发“ before_server_start”服务器事件,但不会触发“ after_server_start”,“ before_server_stop”或“ after_server_stop”服务器事件。

对于更高级的用例,您可以使用AsyncioServer对象触发这些事件,该对象是通过等待服务器任务返回的。

这是一个不完整的示例(请参阅示例中的run_async_advanced.py了解更完整的内容):

serv_coro = app.create_server(host="0.0.0.0", port=8000, return_asyncio_server=True)
loop = asyncio.get_event_loop()
serv_task = asyncio.ensure_future(serv_coro, loop=loop)
server = loop.run_until_complete(serv_task)
server.after_start()
try:
    loop.run_forever()
except KeyboardInterrupt as e:
    loop.stop()
finally:
    server.before_stop()

    # Wait for server to close
    close_task = server.close()
    loop.run_until_complete(close_task)

    # Complete all tasks on the loop
    for connection in server.connections:
        connection.close_if_idle()
    server.after_stop()

基础范例

本文档的这一部分是示例代码的简单集合,可以帮助您快速开始应用程序开发。这些示例大多数都是经过分类的,并为您提供了Sanic存储库中的工作代码示例的链接。

基本的例子

示例的这一部分是代码的集合,这些代码提供了sanic应用程序的简单用例示例。

简单的应用程序

一个简单的sanic应用程序,async具有text和jsontype响应的单一方法。

from sanic import Sanic
from sanic import response as res

app = Sanic(__name__)


@app.route("/")
async def test(req):
    return res.text("I\'m a teapot", status=418)


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)
from sanic import Sanic
from sanic import response

app = Sanic(__name__)


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


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)

简单的应用程序与 Sanic Views

展示了使用的简单机制sanic.viewes.HTTPMethodView以及扩展使用async行为的方法view。

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重定向

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)

命名URL重定向

Sanic提供了一种易于使用的方法,该方法通过名为的帮助程序方法重定向请求,该方法url_for采用唯一的url名称作为参数,并为您返回为其分配的实际路由。这将有助于简化在应用程序的不同部分之间重定向用户所需的工作。

from sanic import Sanic
from sanic import response

app = Sanic(__name__)


@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 response.redirect(url)


@app.route('/posts/<post_id>')
async def post_handler(request, post_id):
    return response.text('Post - {}'.format(post_id))
    
if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000, debug=True)

蓝图

Sanic 提供了一项了不起的功能,可将您的API和路由分组到一个逻辑集合下,该逻辑集合可轻松导入并插入到您的任何sanic应用程序中,这称为 blueprints

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

app = Sanic(__name__)
blueprint = Blueprint('name', url_prefix='/my_blueprint')
blueprint2 = Blueprint('name2', url_prefix='/my_blueprint2')
blueprint3 = Blueprint('name3', url_prefix='/my_blueprint3')


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


@blueprint3.route('/foo')
async def index(request):
    return await file('websocket.html')


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

app.blueprint(blueprint)
app.blueprint(blueprint2)
app.blueprint(blueprint3)

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

日志记录增强功能

即使Sanic附带了一系列日志支持,它也允许最终用户自定义应用程序运行时处理日志的方式。

from sanic import Sanic
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)

以下示例提供了一个示例代码,演示了的用法,sanic.app.Sanic.middleware()以提供一种机制,为每个传入请求分配唯一的请求ID并通过aiotask-context记录它们 。

'''
Based on example from https://github.com/Skyscanner/aiotask-context
and `examples/{override_logging,run_async}.py`.

Needs https://github.com/Skyscanner/aiotask-context/tree/52efbc21e2e1def2d52abb9a8e951f3ce5e6f690 or newer

$ pip install git+https://github.com/Skyscanner/aiotask-context.git
'''

import asyncio
import uuid
import logging
from signal import signal, SIGINT

from sanic import Sanic
from sanic import response

import uvloop
import aiotask_context as context

log = logging.getLogger(__name__)


class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = context.get('X-Request-ID')
        return True


LOG_SETTINGS = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'DEBUG',
            'formatter': 'default',
            'filters': ['requestid'],
        },
    },
    'filters': {
        'requestid': {
            '()': RequestIdFilter,
        },
    },
    'formatters': {
        'default': {
            'format': '%(asctime)s %(levelname)s %(name)s:%(lineno)d %(request_id)s | %(message)s',
        },
    },
    'loggers': {
        '': {
            'level': 'DEBUG',
            'handlers': ['console'],
            'propagate': True
        },
    }
}


app = Sanic(__name__, log_config=LOG_SETTINGS)


@app.middleware('request')
async def set_request_id(request):
    request_id = request.headers.get('X-Request-ID') or str(uuid.uuid4())
    context.set("X-Request-ID", request_id)


@app.route("/")
async def test(request):
    log.debug('X-Request-ID: %s', context.get('X-Request-ID'))
    log.info('Hello from test!')
    return response.json({"test": True})


if __name__ == '__main__':
    asyncio.set_event_loop(uvloop.new_event_loop())
    server = app.create_server(host="0.0.0.0", port=8000, return_asyncio_server=True)
    loop = asyncio.get_event_loop()
    loop.set_task_factory(context.task_factory)
    task = asyncio.ensure_future(server)
    try:
        loop.run_forever()
    except:
        loop.stop()
        

Sanic Stream支持

Sanic框架附带了对大文件流的内置支持,以下代码说明了设置Sanic具有流支持的应用程序的过程。

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')
            await 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='0.0.0.0', port=8000)

示例客户端应用程序,以通过客户端代码显示流式应用程序的用法。

import requests

# Warning: This is a heavy process.

data = ""
for i in range(1, 250000):
    data += str(i)

r = requests.post('http://0.0.0.0:8000/stream', data=data)
print(r.text)

Sanic并发支持

Sanic支持在多个工作人员支持下启动应用程序的功能。但是,重要的是能够限制每个进程/循环的并发性,以确保有效执行。该代码的以下部分提供了有关如何借助以下方法限制并发的简短示例。asyncio.Semaphore

from sanic import Sanic
from sanic.response import json

import asyncio
import aiohttp

app = Sanic(__name__)

sem = None


@app.listener('before_server_start')
def init(sanic, loop):
    global sem
    concurrency_per_worker = 4
    sem = asyncio.Semaphore(concurrency_per_worker, loop=loop)

async def bounded_fetch(session, url):
    """
    Use session object to perform 'get' request on url
    """
    async with sem, session.get(url) as response:
        return await response.json()


@app.route("/")
async def test(request):
    """
    Download and serve example JSON
    """
    url = "https://api.github.com/repos/channelcat/sanic"

    async with aiohttp.ClientSession() as session:
        response = await bounded_fetch(session, url)
        return json(response)


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

通过Docker进行Sanic部署

sanic通过docker和部署应用程序docker-compose很容易完成,以下示例提供了示例的部署simple_server.py

FROM python:3.5
MAINTAINER Channel Cat <channelcat@gmail.com>

ADD . /code
RUN pip3 install git+https://github.com/channelcat/sanic

EXPOSE 8000

WORKDIR /code

CMD ["python", "simple_server.py"]
version: '2'
services:
  sanic:
    build: .
    ports:
      - "8000:8000"

监控和错误处理

Sanic通过提供了全局异常处理程序的可扩展的最低要求最低实现 sanic.handlers.ErrorHandler。本示例说明如何扩展它以启用某些自定义行为。


"""
Example intercepting uncaught exceptions using Sanic's error handler framework.
This may be useful for developers wishing to use Sentry, Airbrake, etc.
or a custom system to log and monitor unexpected errors in production.
First we create our own class inheriting from Handler in sanic.exceptions,
and pass in an instance of it when we create our Sanic instance. Inside this
class' default handler, we can do anything including sending exceptions to
an external service.
"""
from sanic.handlers import ErrorHandler
from sanic.exceptions import SanicException
"""
Imports and code relevant for our CustomHandler class
(Ordinarily this would be in a separate file)
"""


class CustomHandler(ErrorHandler):

    def default(self, request, exception):
        # Here, we have access to the exception object
        # and can do anything with it (log, send to external service, etc)

        # Some exceptions are trivial and built into Sanic (404s, etc)
        if not isinstance(exception, SanicException):
            print(exception)

        # Then, we must finish handling the exception by returning
        # our response to the client
        # For this we can just call the super class' default handler
        return super().default(request, exception)


"""
This is an ordinary Sanic server, with the exception that we set the
server's error_handler to an instance of our CustomHandler
"""

from sanic import Sanic

app = Sanic(__name__)

handler = CustomHandler()
app.error_handler = handler


@app.route("/")
async def test(request):
    # Here, something occurs which causes an unexpected exception
    # This exception will flow to our custom handler.
    raise SanicException('You Broke It!')

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

使用外部服务提供商进行监控

import logging
import socket
from os import getenv
from platform import node
from uuid import getnode as get_mac

from logdna import LogDNAHandler

from sanic import Sanic
from sanic.response import json
from sanic.request import Request

log = logging.getLogger('logdna')
log.setLevel(logging.INFO)


def get_my_ip_address(remote_server="google.com"):
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.connect((remote_server, 80))
        return s.getsockname()[0]


def get_mac_address():
    h = iter(hex(get_mac())[2:].zfill(12))
    return ":".join(i + next(h) for i in h)


logdna_options = {
    "app": __name__,
    "index_meta": True,
    "hostname": node(),
    "ip": get_my_ip_address(),
    "mac": get_mac_address()
}

logdna_handler = LogDNAHandler(getenv("LOGDNA_API_KEY"), options=logdna_options)

logdna = logging.getLogger(__name__)
logdna.setLevel(logging.INFO)
logdna.addHandler(logdna_handler)

app = Sanic(__name__)


@app.middleware
def log_request(request: Request):
    logdna.info("I was Here with a new Request to URL: {}".format(request.url))


@app.route("/")
def default(request):
    return json({
        "response": "I was here"
    })


if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=getenv("PORT", 8080)
    )
from os import getenv

from raygun4py.raygunprovider import RaygunSender

from sanic import Sanic
from sanic.exceptions import SanicException
from sanic.handlers import ErrorHandler


class RaygunExceptionReporter(ErrorHandler):

    def __init__(self, raygun_api_key=None):
        super().__init__()
        if raygun_api_key is None:
            raygun_api_key = getenv("RAYGUN_API_KEY")

        self.sender = RaygunSender(raygun_api_key)

    def default(self, request, exception):
        self.sender.send_exception(exception=exception)
        return super().default(request, exception)


raygun_error_reporter = RaygunExceptionReporter()
app = Sanic(__name__, error_handler=raygun_error_reporter)


@app.route("/raise")
async def test(request):
    raise SanicException('You Broke It!')


if __name__ == '__main__':
    app.run(
        host="0.0.0.0",
        port=getenv("PORT", 8080)
    )
import rollbar

from sanic.handlers import ErrorHandler
from sanic import Sanic
from sanic.exceptions import SanicException
from os import getenv

rollbar.init(getenv("ROLLBAR_API_KEY"))


class RollbarExceptionHandler(ErrorHandler):

    def default(self, request, exception):
        rollbar.report_message(str(exception))
        return super().default(request, exception)


app = Sanic(__name__, error_handler=RollbarExceptionHandler())


@app.route("/raise")
def create_error(request):
    raise SanicException("I was here and I don't like where I am")


if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=getenv("PORT", 8080)
    )
    
from os import getenv

from sentry_sdk import init as sentry_init
from sentry_sdk.integrations.sanic import SanicIntegration

from sanic import Sanic
from sanic.response import json

sentry_init(
    dsn=getenv("SENTRY_DSN"),
    integrations=[SanicIntegration()],
)

app = Sanic(__name__)


# noinspection PyUnusedLocal
@app.route("/working")
async def working_path(request):
    return json({
        "response": "Working API Response"
    })


# noinspection PyUnusedLocal
@app.route("/raise-error")
async def raise_error(request):
    raise Exception("Testing Sentry Integration")


if __name__ == '__main__':
    app.run(
        host="0.0.0.0",
        port=getenv("PORT", 8080)
    )
    

安全

以下示例代码显示了一个基于装饰器的简单身份验证和授权机制,可以将其设置为保护sanicapi端点。

# -*- coding: utf-8 -*-

from sanic import Sanic
from functools import wraps
from sanic.response import json

app = Sanic()


def check_request_for_authorization_status(request):
    # Note: Define your check, for instance cookie, session.
    flag = True
    return flag


def authorized(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


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

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)
    

Sanic WebSocket

Sanic提供了轻松添加路由并将其映射到websocket处理程序的功能。

<!DOCTYPE html>
<html>
    <head>
        <title>WebSocket demo</title>
    </head>
    <body>
        <script>
            var ws = new WebSocket('ws://' + document.domain + ':' + location.port + '/feed'),
                messages = document.createElement('ul');
            ws.onmessage = function (event) {
                var messages = document.getElementsByTagName('ul')[0],
                    message = document.createElement('li'),
                    content = document.createTextNode('Received: ' + event.data);
                message.appendChild(content);
                messages.appendChild(message);
            };
            document.body.appendChild(messages);
            window.setInterval(function() {
                data = 'bye!'
                ws.send(data);
                var messages = document.getElementsByTagName('ul')[0],
                    message = document.createElement('li'),
                    content = document.createTextNode('Sent: ' + data);
                message.appendChild(content);
                messages.appendChild(message);
            }, 1000);
        </script>
    </body>
</html>
from sanic import Sanic
from sanic.response import file

app = Sanic(__name__)


@app.route('/')
async def index(request):
    return await file('websocket.html')


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


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

虚拟主机Suppport 
from sanic import response
from sanic import Sanic
from sanic.blueprints import Blueprint

# Usage
# curl -H "Host: example.com" localhost:8000
# curl -H "Host: sub.example.com" localhost:8000
# curl -H "Host: bp.example.com" localhost:8000/question
# curl -H "Host: bp.example.com" localhost:8000/answer

app = Sanic()
bp = Blueprint("bp", host="bp.example.com")


@app.route('/', host=["example.com",
                      "somethingelse.com",
                      "therestofyourdomains.com"])
async def hello(request):
    return response.text("Some defaults")


@app.route('/', host="sub.example.com")
async def hello(request):
    return response.text("42")


@bp.route("/question")
async def hello(request):
    return response.text("What is the meaning of life?")


@bp.route("/answer")
async def hello(request):
    return response.text("42")

app.blueprint(bp)

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)
带有并行测试运行支持的单元测试
以下示例显示了如何sanic使用pytest-xdist插件提供的并行测试执行支持来启动并运行单元测试应用程序

"""pytest-xdist example for sanic server

Install testing tools:

    $ pip install pytest pytest-xdist

Run with xdist params:

    $ pytest examples/pytest_xdist.py -n 8  # 8 workers
"""
import re
from sanic import Sanic
from sanic.response import text
from sanic.testing import PORT as PORT_BASE, SanicTestClient
import pytest


@pytest.fixture(scope="session")
def test_port(worker_id):
    m = re.search(r'[0-9]+', worker_id)
    if m:
        num_id = m.group(0)
    else:
        num_id = 0
    port = PORT_BASE + int(num_id)
    return port


@pytest.fixture(scope="session")
def app():
    app = Sanic()

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

    return app


@pytest.fixture(scope="session")
def client(app, test_port):
    return SanicTestClient(app, test_port)


@pytest.mark.parametrize('run_id', range(100))
def test_index(client, run_id):
    request, response = client._sanic_endpoint_test('get', '/')
    assert response.status == 200
    assert response.text == 'OK'
修订请求对象
中的request对象Sanic是一种dict对象这意味着request可以将对象作为常规dict对象进行操作

from sanic import Sanic
from sanic.response import text
from random import randint

app = Sanic()


@app.middleware('request')
def append_request(request):
    # Add new key with random value
    request['num'] = randint(0, 100)


@app.get('/pop')
def pop_handler(request):
    # Pop key from request object
    num = request.pop('num')
    return text(num)


@app.get('/key_exist')
def key_exist_handler(request):
    # Check the key is exist or not
    if 'num' in request:
        return text('num exist in request')

    return text('num does not exist in reqeust')


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

有关更多示例和有用示例,请访问Huge-Sanic的GitHub页面

调试模式

Sanic端点可以使用test_client对象在本地进行测试,该对象取决于一个附加程序包:httpx 库,该库实现了一个镜像请求库的API 。

该test_client自曝获得,后,放,删除,补丁,头和选择方法,为您对您的应用程序运行。一个简单的示例(使用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方法之一时,Sanic应用程序将在127.0.0.1:42101上运行,并使用httpx针对您的应用程序执行测试请求。

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

  • uri (默认值'/'`)表示要测试的URI的字符串。
  • collect_request (默认为True)一个布尔值,确定函数是否将返回原始请求。如果设置为True,则返回值为(request,response)的元组,如果为False,则仅返回响应。
  • server_kwargs (默认为{}),用于在运行测试请求之前将其他参数传递给app.run。
  • debug (默认为False)一个布尔值,用于确定是否以调试模式运行服务器。

该函数还接受* request_args和** request_kwargs,它们直接传递给请求。

例如,要将数据提供给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'

有关httpx可用参数的更多信息,可以在[ httpx文档中找到。

使用随机端口

如果您需要使用内核选择的免费非特权端口而不是SanicTestClient的默认端口进行测试,则可以通过指定 port = None进行测试。在大多数系统上,端口范围为1024至65535。

# Import the Sanic app, usually created with Sanic(__name__)
from external_server import app
from sanic.testing import SanicTestClient

def test_index_returns_200():
    request, response = SanicTestClient(app, port=None).get('/')
    assert response.status == 200
pytest中信高科
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还提供了一些有用的固定装置,例如循环,未使用的端口,test_server,test_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}

扩展

API

  • Sanic CRUD:使用Peewee模型生成CRUD REST API。
  • Sanic-GraphQL:GraphQL与Sanic的集成
  • Sanic-RestPlus:Sanic的Flask-RestPlus的端口。具有SwaggerUI生成功能的全功能REST API。
  • Sanic-Transmute:Sanic扩展,可从python函数和类生成API,并自动生成Swagger UI /文档。

认证方式

  • Sanic-JWT:JSON Web令牌(JWT)的身份验证扩展。
  • Sanic-JWT-Extended:flask-jwt-extended的端口,提供访问/刷新令牌,具有新鲜,轻松的自定义声明插入和基于角色的访问控制
  • Sanic-OAuth:具有许多提供程序和OAuth1 / OAuth2支持的OAuth库。
  • Sanic-Token-Auth:基于令牌的简单身份验证。
  • Sanic-HTTPAuth:Flask-HTTPAuth的叉子,为Sanic路由提供基本,摘要和令牌HTTP身份验证

Development

前端

监控

ORM

  • GINO:基于SQLAlchemy核心的轻量级异步ORM,带有asyncpg方言和Sanic扩展。
  • Sanic-Motor:简单的motor包装器。
  • Sanic-mongodb-extension:具有对Sanic框架的μMongoODM支持的MongoDB扩展

Requests and Responses

Caching

Queues

脚手架

  • Cookiecutter-Sanic:在定义良好的项目结构中,只需几秒钟即可启动并运行sanic应用程序。包括用于部署,单元测试,自动发布管理和更改日志生成的电池。

Session

  • Sanic Sessions:对人类的会话支持。与不同的后端Redis,Mongodb,memcache或内存存储一起使用。

工具

资源


例子

视频

1
https://gitee.com/amiko/my-docsify-site.git
git@gitee.com:amiko/my-docsify-site.git
amiko
my-docsify-site
my-docsify-site
master

Search

53164aa7 5694891 3bd8fe86 5694891