Python sanic框架教程

Sanic 是一个基于 Python 的异步 Web 框架，它使用 async/await 语法来编写异步代码，具有高性能和易用性的特点。本教程将详细介绍 Sanic 的核心概念和使用方法。

目录

  1. 安装与快速开始
  2. 基本路由
  3. 请求与响应
  4. 中间件
  5. 蓝图
  6. WebSocket
  7. 配置
  8. 数据库集成
  9. 测试
  10. 部署

安装与快速开始

安装 Sanic

pip install sanic

最小应用

from sanic import Sanic
from sanic.response import text

app = Sanic("MyApp")

@app.route("/")
async def hello(request):
    return text("Hello, Sanic!")

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

运行后访问 http://localhost:8000 即可看到 "Hello, Sanic!"。

基本路由

路由装饰器

@app.route("/")
async def home(request):
    return text("Home Page")

@app.route("/about")
async def about(request):
    return text("About Page")

HTTP 方法

@app.get("/get")
async def get_handler(request):
    return text("GET request")

@app.post("/post")
async def post_handler(request):
    return text("POST request")

@app.put("/put")
async def put_handler(request):
    return text("PUT request")

@app.delete("/delete")
async def delete_handler(request):
    return text("DELETE request")

路径参数

@app.route("/user/<user_id>")
async def user_info(request, user_id):
    return text(f"User ID: {user_id}")

# 类型限定
@app.route("/post/<post_id:int>")
async def post_info(request, post_id):
    return text(f"Post ID: {post_id} (integer)")

支持的类型标记：

  • string (默认)
  • int
  • number
  • alpha
  • path
  • uuid

请求与响应

请求对象

@app.post("/login")
async def login(request):
    # 获取表单数据
    username = request.form.get("username")
    password = request.form.get("password")

    # 获取JSON数据
    data = request.json

    # 获取查询参数
    query_param = request.args.get("param")

    # 获取headers
    user_agent = request.headers.get("user-agent")

    # 获取cookies
    session_id = request.cookies.get("session_id")

    return text(f"Welcome {username}!")

响应类型

from sanic.response import (
    text, json, html, 
    file, file_stream, 
    redirect, empty
)

@app.route("/text")
async def text_response(request):
    return text("Plain text")

@app.route("/json")
async def json_response(request):
    return json({"message": "JSON response"})

@app.route("/html")
async def html_response(request):
    return html("<h1>HTML response</h1>")

@app.route("/file")
async def file_response(request):
    return await file("/path/to/file.txt")

@app.route("/redirect")
async def redirect_response(request):
    return redirect("/new-location")

@app.route("/empty")
async def empty_response(request):
    return empty()

自定义响应

from sanic.response import HTTPResponse

@app.route("/custom")
async def custom_response(request):
    return HTTPResponse(
        body="Custom response",
        status=201,
        headers={"X-Custom-Header": "Value"},
        content_type="text/plain"
    )

中间件

请求中间件

@app.middleware("request")
async def print_request(request):
    print(f"Incoming request: {request.url}")

@app.middleware("response")
async def add_header(request, response):
    response.headers["X-Server"] = "Sanic"

异常处理

from sanic.exceptions import NotFound

@app.exception(NotFound)
async def ignore_404s(request, exception):
    return text(f"Oops! {request.url} not found", status=404)

@app.exception(Exception)
async def catch_all_exceptions(request, exception):
    return text(f"Server error: {str(exception)}", status=500)

蓝图

蓝图用于组织大型应用的路由。

创建蓝图

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

auth_bp = Blueprint("auth", url_prefix="/auth")

@auth_bp.route("/login")
async def login(request):
    return text("Login page")

@auth_bp.route("/logout")
async def logout(request):
    return text("Logout page")

注册蓝图

from blueprints.auth import auth_bp

app.blueprint(auth_bp)

WebSocket

Sanic 支持 WebSocket 通信。

@app.websocket("/ws")
async def websocket_handler(request, ws):
    while True:
        data = await ws.recv()
        print(f"Received: {data}")
        await ws.send(f"Echo: {data}")

配置

基本配置

app.config.API_VERSION = "1.0"
app.config.DB_NAME = "appdb"
app.config.DB_USER = "appuser"

从文件加载配置

# config.py
DB_HOST = "localhost"
DB_PORT = 5432

# app.py
app.update_config("config.py")

环境变量

import os

app.config.DB_PASSWORD = os.getenv("DB_PASSWORD")

数据库集成

使用 asyncpg (PostgreSQL)

import asyncpg

@app.listener('before_server_start')
async def setup_db(app, loop):
    app.ctx.pool = await asyncpg.create_pool(
        user='user',
        password='password',
        database='database',
        host='localhost'
    )

@app.listener('after_server_stop')
async def close_db(app, loop):
    await app.ctx.pool.close()

@app.route("/users")
async def get_users(request):
    async with request.app.ctx.pool.acquire() as connection:
        users = await connection.fetch("SELECT * FROM users")
        return json([dict(user) for user in users])

使用 Tortoise-ORM

from tortoise.contrib.sanic import register_tortoise

register_tortoise(
    app,
    db_url="postgres://user:password@localhost/database",
    modules={"models": ["models"]},
    generate_schemas=True
)

# models.py
from tortoise.models import Model
from tortoise import fields

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(50)
    email = fields.CharField(100)

测试

使用 pytest

# test_app.py
from sanic_testing import TestManager

def test_hello():
    app = Sanic("TestApp")
    TestManager(app)

    @app.route("/")
    async def hello(request):
        return text("Hello, Test!")

    request, response = app.test_client.get("/")
    assert response.text == "Hello, Test!"
    assert response.status == 200

运行测试：

pytest test_app.py

部署

使用 Gunicorn

pip install gunicorn
gunicorn myapp:app --worker-class sanic.worker.GunicornWorker -b 0.0.0.0:8000 -w 4

使用 uvicorn (ASGI)

pip install uvicorn
uvicorn myapp:app --host 0.0.0.0 --port 8000

生产配置

app.run(
    host="0.0.0.0",
    port=8000,
    workers=4,
    access_log=False,  # 生产环境可以关闭访问日志
    debug=False
)

高级特性

信号

from sanic import Sanic
from sanic.signals import Event

app = Sanic("SignalApp")

@app.signal(Event.HTTP_LIFECYCLE_BEGIN)
async def before_request(*args, **kwargs):
    print("Request is starting")

@app.signal(Event.HTTP_LIFECYCLE_COMPLETE)
async def after_request(*args, **kwargs):
    print("Request is complete")

后台任务

async def notify_server_started(app, loop):
    print("Server successfully started!")

app.add_task(notify_server_started)

版本控制

@app.route("/resource", version=1)
async def resource_v1(request):
    return text("Version 1")

@app.route("/resource", version=2)
async def resource_v2(request):
    return text("Version 2")

访问时使用 /v1/resource 或 /v2/resource。

总结

Sanic 是一个高性能的 Python 异步 Web 框架，具有以下特点：

  1. 支持 async/await 语法
  2. 高性能，适合高并发场景
  3. 简单易用的 API 设计
  4. 支持 WebSocket
  5. 丰富的扩展生态系统
  6. 良好的测试支持
  7. 多种部署选项

通过本教程，您应该已经掌握了 Sanic 的核心功能。要了解更多，请参考 Sanic 官方文档。









results matching ""

    No results matching ""