自学内容网 自学内容网

【Sanic 框架 / 2】进阶与实战:高级功能与项目开发

四、高级功能学习

插件开发

如何为 Sanic 开发自定义插件

Sanic 支持插件的开发,允许我们为应用程序扩展功能。可以通过定义插件类来实现自定义插件。插件类需要继承自 sanic.Plugin 类,并实现 on_app_init() 方法以在应用启动时进行初始化。

以下是一个简单的插件示例,它在应用启动时记录日志:

from sanic import Sanic
from sanic.plugins import Plugin
import logging

class MyPlugin(Plugin):
    def on_app_init(self, app, loop):
        app.logger.info("My custom plugin has been initialized!")

app = Sanic("PluginExample")
plugin = MyPlugin()

@app.route("/")
async def index(request):
    return json({"message": "Hello from the plugin!"})

app.blueprint(plugin)
app.run()

此插件会在应用启动时输出日志。

使用现有的 Sanic 插件(如身份验证插件)

Sanic 提供了一些现成的插件,可以帮助我们简化开发过程。例如,sanic-auth 插件可以方便地为应用添加身份验证功能:

pip install sanic-auth

在应用中使用 sanic-auth 插件来添加基本的身份验证:

from sanic import Sanic
from sanic.response import json
from sanic_auth import Auth

app = Sanic("AuthExample")
auth = Auth(app)

@app.route("/protected")
@auth.login_required
async def protected(request):
    return json({"message": "This is a protected route"})

app.run()

通过使用插件,可以显著提高开发效率和减少代码量。

服务端渲染与模板引擎

集成模板引擎(如 Jinja2

Sanic 支持与模板引擎集成,常用的模板引擎是 Jinja2,它提供了动态生成 HTML 页面和渲染变量的能力。要使用 Jinja2,首先需要安装该库:

pip install jinja2

然后,将 Jinja2 集成到 Sanic 中:

from sanic import Sanic
from sanic.response import html
from jinja2 import Environment, FileSystemLoader

app = Sanic("TemplateExample")
env = Environment(loader=FileSystemLoader("templates"))

@app.route("/")
async def index(request):
    template = env.get_template("index.html")
    return html(template.render(name="Sanic"))

app.run()

渲染动态页面的示例

在上述代码中,index.html 文件会被渲染并返回给客户端。render() 方法接收一个字典,动态插入数据到模板中。以下是 index.html 的示例:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>{{ name }}</title>
</head>
<body>
    <h1>Hello, {{ name }}!</h1>
</body>
</html>

通过这种方式,Sanic 能够在服务端生成动态内容并返回给客户端。

安全性

保护敏感数据(如 API 密钥)

在 Web 开发中,保护敏感数据(如 API 密钥、数据库密码)非常重要。为了避免在代码中硬编码敏感信息,可以使用环境变量或配置文件来管理这些数据。

首先,安装 python-dotenv

pip install python-dotenv

然后,创建 .env 文件来存储敏感数据:

API_KEY=your-api-key-here
DB_PASSWORD=your-database-password-here

在代码中加载并使用这些敏感数据:

from dotenv import load_dotenv
import os

load_dotenv()

API_KEY = os.getenv("API_KEY")
DB_PASSWORD = os.getenv("DB_PASSWORD")

@app.route("/api")
async def api(request):
    return json({"api_key": API_KEY})

防范常见的 Web 攻击(如 CSRF 和 XSS)

Sanic 本身并不直接提供 CSRF 和 XSS 防护,但可以通过手动实现防护机制来增强安全性。

防范 CSRF 攻击

跨站请求伪造(CSRF)攻击通过伪造请求来冒充合法用户。在处理敏感请求时,通常使用 CSRF Token 来防止攻击。可以在表单中包含随机生成的 CSRF Token,并在每次提交表单时验证 Token。

import secrets

@app.route("/form", methods=["GET", "POST"])
async def form(request):
    if request.method == "POST":
        token = request.form.get("csrf_token")
        if token != request.ctx.csrf_token:
            return json({"error": "CSRF Token mismatch"}, status=403)
        return json({"message": "Form submitted successfully"})

    # 生成 CSRF Token
    csrf_token = secrets.token_hex(16)
    request.ctx.csrf_token = csrf_token
    return html(f"<form method='POST'><input type='hidden' name='csrf_token' value='{csrf_token}'><button type='submit'>Submit</button></form>")
防范 XSS 攻击

跨站脚本攻击(XSS)允许攻击者在网页中注入恶意脚本。为了防止 XSS,应该避免直接将用户输入渲染到页面中,而是对用户输入进行转义处理,确保不执行任何恶意代码。

在模板渲染时,Jinja2 默认会对插入的内容进行转义,因此只需确保使用安全的模板渲染方式即可。

队列与任务调度

集成任务队列(如 CeleryHuey

Sanic 可以与异步任务队列集成,CeleryHuey 是常用的任务队列系统。可以使用这些队列来执行后台任务,而不会阻塞主线程。

Huey 为例,首先安装:

pip install huey

然后集成到 Sanic 中:

from sanic import Sanic
from sanic.response import json
from huey import Huey
from sanic import Blueprint

huey = Huey("tasks")

app = Sanic("TaskQueueExample")

@app.route("/send_email")
async def send_email(request):
    # 调用异步任务
    send_email_task()
    return json({"status": "Email sending in background"})

@huey.task()
def send_email_task():
    # 模拟发送邮件的任务
    print("Sending email...")

app.run()

此代码将异步执行邮件发送任务,而不会阻塞主线程。

定义后台任务处理逻辑

可以为后台任务定义定时器或周期性任务,Huey 支持定时执行任务:

@huey.periodic_task()
def print_hello():
    print("Hello, world!")

该任务将每分钟执行一次。

多服务架构

使用蓝图拆分项目模块

蓝图(Blueprints)是 Sanic 的模块化开发方案,允许将应用分拆成多个独立模块,方便管理和维护。

from sanic import Blueprint

bp = Blueprint("example")

@bp.route("/")
async def index(request):
    return json({"message": "Hello from blueprint"})

app.blueprint(bp)
app.run()

蓝图支持独立定义路由和中间件,非常适合大型项目的组织。

将 Sanic 应用于微服务架构

Sanic 非常适合用于微服务架构。在微服务架构中,应用程序拆分为多个独立的服务,每个服务都可以使用 Sanic 来处理特定的功能和任务。每个服务通过 RESTful API 或消息队列进行通信。

例如,一个电商平台可以将订单、支付、库存等功能拆分成不同的微服务,每个服务使用 Sanic 构建,并且通过 HTTP 或消息队列进行交互。

部署

使用 gunicorn 部署 Sanic 应用

为了将 Sanic 应用部署到生产环境,通常使用 gunicorn 作为 WSGI 服务器来启动应用。首先安装 gunicorn

pip install gunicorn

然后,通过 gunicorn 启动 Sanic 应用:

gunicorn -w 4 -b 0.0.0.0:8000 app:app

此命令会启动 4 个工作进程,并将应用绑定到 8000 端口。

容器化部署(使用 Docker)

为了便于部署和扩展,可以将 Sanic 应用容器化。首先创建一个 Dockerfile 文件:

FROM python:3.9-slim

WORKDIR /app
COPY . /app

RUN pip install -r requirements.txt

CMD ["python", "app.py"]

然后,使用 Docker 构建并运行容器:

docker build -t sanic-app .
docker run -p 8000:8000 sanic-app

这样,就能将 Sanic 应用容器化,方便部署到任何支持 Docker 的环境中。


五、实战项目与综合练习

开发一个 RESTful API 服务

用户注册与登录功能

开发一个 RESTful API 服务时,首先需要实现用户注册与登录功能。在此示例中,我们使用 JWT(JSON Web Token)进行身份验证,确保用户的请求安全。

  1. 用户注册: 用户提交用户名和密码,系统将密码加密存储并返回注册成功的响应。

  2. 用户登录: 用户通过用户名和密码进行登录,验证成功后生成 JWT Token,用户通过该 Token 进行后续请求的身份验证。

以下是实现用户注册和登录功能的完整代码:

from sanic import Sanic
from sanic.response import json
from sanic_jwt import Initialize, protected, exceptions
import hashlib
import jwt
import datetime


app = Sanic("AuthExmp")
# 密钥设置
app.config.JWT_SECRET = 'secret_key_114514'  

users_db = {}


def authenticate(request):
    username = request.json.get("username")
    password = request.json.get("password")
    hashed_password = hashlib.sha256(password.encode()).hexdigest()
    if users_db.get(username)!= hashed_password:
        raise exceptions.AuthenticationFailed("Invalid credentials")
    return username


@app.post("/register")
async def register(request):
    username = request.json.get("username")
    password = request.json.get("password")
    if username in users_db:
        return json({"error": "User already exists"}, status=400)
    # 存储哈希后的密码
    users_db[username] = hashlib.sha256(password.encode()).hexdigest()
    return json({"message": "User registered successfully"})


@app.post("/login")
async def login(request):
    username = authenticate(request)
    # 生成 JWT Token
    token = jwt.encode(
        {'user': username, 'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=1)},
        app.config.JWT_SECRET, algorithm='HS256'
    )
    return json({"message": "Login successful", "token": token})


Initialize(app, authenticate=authenticate)


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

代码解释:

  • authenticate 函数用于验证用户登录信息,将用户输入的密码进行哈希处理,与存储在 users_db 中的密码比较,如果不匹配则抛出异常,否则返回用户名。
  • register 函数首先检查用户名是否已存在,若不存在则将用户的密码进行哈希存储,同时返回注册成功消息。
  • login 函数调用 authenticate 函数进行身份验证,验证通过后使用 jwt.encode 函数生成一个有效期为 1 小时的 JWT Token,包含用户信息和过期时间,使用 app.config.JWT_SECRET 作为密钥,并设置为 HS256 算法。

为了验证用户注册和登录功能,可以使用 curl 或 Postman 等工具发送 HTTP 请求:

  • 注册用户:
curl -X POST http://localhost:8000/register -H "Content-Type: application/json" -d '{"username":"testuser","password":"testpass"}'
  • 登录用户:
curl -X POST http://localhost:8000/login -H "Content-Type: application/json" -d '{"username":"testuser","password":"testpass"}'

数据的增删改查接口实现

RESTful API 通常包括四个基本操作:增(Create)、查(Read)、改(Update)、删(Delete)。例如,在我们的应用中,用户可以管理自己的任务:

以下是实现任务管理部分的代码:

# 接着上面用户注册与登录的代码
 
@app.post("/tasks")
@protected()
async def create_task(request):
    task_id = len(tasks_db) + 1
    task = {"id": task_id, "title": request.json.get("title"), "completed": False}
    tasks_db[task_id] = task
    return json(task)


@app.get("/tasks")
@protected()
async def get_tasks(request):
    return json(list(tasks_db.values()))


@app.put("/tasks/<task_id:int>")
@protected()
async def update_task(request, task_id):
    task = tasks_db.get(task_id)
    if not task:
        return json({"error": "Task not found"}, status=404)
    task["title"] = request.json.get("title", task["title"])
    task["completed"] = request.json.get("completed", task["completed"])
    return json(task)


@app.delete("/tasks/<task_id:int>")
@protected()
async def delete_task(request, task_id):
    task = tasks_db.pop(task_id, None)
    if not task:
        return json({"error": "Task not found"}, status=404)
    return json({"message": "Task deleted"})


Initialize(app, authenticate=authenticate)


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

代码解释:

  • create_task 函数使用 @protected() 装饰器确保只有经过身份验证的用户才能调用。它根据请求中的任务标题创建一个新任务,将其存储在 tasks_db 中,并返回任务信息。
  • get_tasks 函数使用 @protected() 装饰器,它会返回存储在 tasks_db 中的所有任务列表。
  • update_task 函数使用 @protected() 装饰器,根据 task_id 找到任务,更新任务的标题和完成状态。如果任务不存在返回错误信息。
  • delete_task 函数使用 @protected() 装饰器,根据 task_id 删除任务,如果任务不存在返回错误信息。

为了验证任务的 CRUD 操作,可以使用 curl 或 Postman 等工具发送 HTTP 请求:

  • 创建任务:
curl -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -H "Authorization: Bearer <your_jwt_token>" -d '{"title":"Buy groceries"}'
  • 获取任务列表:
curl -X GET http://localhost:8000/tasks -H "Authorization: Bearer <your_jwt_token>"
  • 更新任务:
curl -X PUT http://localhost:8000/tasks/1 -H "Content-Type: application/json" -H "Authorization: Bearer <your_jwt_token>" -d '{"title":"Buy fruits", "completed":true}'
  • 删除任务:
curl -X DELETE http://localhost:8000/tasks/1 -H "Authorization: Bearer <your_jwt_token>"

注意事项:

  • 确保已安装 sanicsanic_jwt 库,可以使用 pip install sanic sanic_jwt 进行安装。
  • 替换 <your_jwt_token> 为登录成功后获得的有效 JWT Token。
  • 在测试 API 功能时,确保 API 服务已启动,可以通过运行上述 Python 脚本启动服务。

验证步骤:

  1. 首先启动服务,运行 Python 脚本。
  2. 使用 curl 或 Postman 等工具发送注册请求,检查用户是否能成功注册。
  3. 发送登录请求,检查是否能成功登录并获取 JWT Token。
  4. 使用获取到的 JWT Token 发送任务的 CRUD 请求,验证任务操作是否能正常执行。
    • 创建任务时,检查任务是否正确存储在 tasks_db 中。
    • 获取任务列表时,检查返回的任务列表是否正确。
    • 更新任务时,检查任务的信息是否正确更新。
    • 删除任务时,检查任务是否正确从 tasks_db 中删除。

总结:
通过上述修改后的文章和代码,提供了完整的 RESTful API 服务的开发和验证流程,包括用户注册、登录以及任务的增删改查功能。使用 JWT 进行身份验证确保了用户操作的安全性,通过代码解释和验证步骤,可以帮助开发者更好地理解和测试 API 服务的实现。

WebSocket 聊天应用

实现实时消息的发送与接收

WebSocket 是一种双向通信协议,非常适合实现实时聊天功能。以下是一个简单的 WebSocket 聊天应用,允许用户发送和接收消息:

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

app = Sanic("ChatApp")

@app.websocket("/chat")
async def chat(request, ws: WebSocket):
    while True:
        message = await ws.recv()
        await ws.send(f"Server received: {message}")

app.run()

添加在线用户列表功能

在聊天应用中,通常需要维护一个在线用户列表。可以使用一个全局变量存储当前连接的 WebSocket 客户端,实现在线用户功能。

online_users = set()

@app.websocket("/chat")
async def chat(request, ws: WebSocket):
    online_users.add(ws)
    try:
        while True:
            message = await ws.recv()
            # Broadcast to all users
            for user in online_users:
                if user != ws:
                    await user.send(f"New message: {message}")
    except:
        online_users.remove(ws)
        await ws.close()

TODO 应用

实现任务管理功能(添加、删除、标记完成)

TODO 应用是一个经典的任务管理工具,允许用户添加、删除和标记任务为已完成。以下是实现示例:

tasks = []

@app.post("/tasks")
async def add_task(request):
    task_title = request.json.get("title")
    task = {"id": len(tasks) + 1, "title": task_title, "completed": False}
    tasks.append(task)
    return json(task)

@app.put("/tasks/<task_id>")
async def mark_completed(request, task_id):
    task_id = int(task_id)
    task = next((task for task in tasks if task["id"] == task_id), None)
    
    if task:
        task["completed"] = True
        return json(task)
    else:
        return json({"error": "Task not found"}, status=404)

@app.delete("/tasks/<task_id>")
async def delete_task(request, task_id):
    task = next((task for task in tasks if task["id"] == int(task_id)), None)
    
    if task:
        tasks.remove(task)
        return json({"message": "Task deleted"})
    else:
        return json({"error": "Task not found"}, status=404)

集成前端界面与 API 调试

我们可以通过 HTML 和 JavaScript 创建前端界面,使用 fetch API 与后端接口进行交互,实现任务管理功能。

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>TODO App</title>
</head>
<body>
    <h1>TODO List</h1>
    <ul id="task-list"></ul>
    <input type="text" id="task-input">
    <button onclick="addTask()">Add Task</button>
    
    <script>
        async function addTask() {
            const taskTitle = document.getElementById("task-input").value;
            const response = await fetch('/tasks', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({ title: taskTitle })
            });
            const task = await response.json();
            displayTasks();
        }

        async function displayTasks() {
            const response = await fetch('/tasks');
            const tasks = await response.json();
            const taskList = document.getElementById("task-list");
            taskList.innerHTML = '';
            tasks.forEach(task => {
                const li = document.createElement('li');
                li.textContent = task.title;
                taskList.appendChild(li);
            });
        }

        displayTasks();
    </script>
</body>
</html>

博客系统

文章发布与评论功能

博客系统通常包括文章发布和评论功能。在这个示例中,用户可以发布文章,并对文章进行评论。以下是一个简单的实现:

articles_db = {}
comments_db = {}

@app.post("/articles")
async def create_article(request):
    article_id = len(articles_db) + 1
    article = {"id": article_id, "title": request.json.get("title"), "content": request.json.get("content")}
    articles_db[article_id] = article
    return json(article)

@app.post("/articles/<article_id>/comments")
async def add_comment(request, article_id):
    comment = {"content": request.json.get("content")}
    if article_id not in comments_db:
        comments_db[article_id] = []
    comments_db[article_id].append(comment)
    return json(comment)

@app.get("/articles/<article_id>/comments")
async def get_comments(request, article_id):
    return json(comments_db.get(int(article_id), []))

集成简单的用户权限管理

简单的权限管理可以通过 JWT 实现,确保只有认证的用户才能发布文章和评论。通过 JWT Token 验证用户身份:

from sanic_jwt import protected

@app.post("/articles")
@protected()
async def create_article(request):
    article_id = len(articles_db) + 1
    article = {"id": article_id, "title": request.json.get("title"), "content": request.json.get("content")}
    articles_db[article_id] = article
    return json(article)

分布式任务管理系统

后台定时任务调度与分发

在分布式任务管理系统中,后台定时任务调度与分发是非常重要的。为了有效地处理异步任务,通常会使用 CeleryHuey 等任务队列工具。以下是 Huey 作为任务调度的示例,任务会根据指定的时间周期自动执行。

  1. 设置定时任务调度: 定时任务会定期执行一个特定的函数,例如发送邮件或进行数据备份。
from huey import RedisHuey

huey = RedisHuey('tasks')

# 每隔一分钟执行一次的定时任务
@huey.periodic_task(crontab(minute='*'))
def send_email_reminder():
    print("Sending email reminder...")
  1. 后台任务分发: 在任务队列中,我们可以创建后台任务并将其分发到不同的工作者进程中处理。每个任务可以被异步执行,确保主应用不会被阻塞。
from huey import RedisHuey

huey = RedisHuey('tasks')

@huey.task()
def process_data(data):
    # 模拟耗时的数据处理任务
    result = heavy_computation(data)
    return result

def heavy_computation(data):
    # 假设这是一个资源消耗较大的计算
    return sum(data)

后台任务的调度和执行:
任务一旦被分发到队列中,后台工作进程会从队列中取出任务并异步执行。CeleryHuey 都允许创建多个工作进程来并行处理任务,从而提高系统的吞吐量。

使用异步任务队列处理大量数据

处理大量数据时,常常会导致请求的延迟或服务器的负载过高。通过异步任务队列,可以将耗时的任务分离到后台执行,避免阻塞主应用。

在以下示例中,process_large_data 任务会异步执行,因此不会影响主应用的响应速度。

@app.route("/process")
async def process_large_data(request):
    # 假设有大量数据需要处理
    data = [i for i in range(1000000)]
    
    # 将任务分发到异步队列中
    process_data(data)
    
    return json({"message": "Data is being processed"})

这个设计让主应用快速响应,而数据处理的任务则在后台进行。


六、官方文档与社区资源

官方文档

  • Sanic 官方文档
    • 官方文档是学习 Sanic 框架的首要资源,详细介绍了框架的安装、配置、使用方法以及 API 文档等。文档中还包括了各种常见问题的解答和最佳实践,适合从入门到进阶的开发者。
    • 在官网上,可以找到丰富的示例代码,并通过文档快速掌握如何在不同的场景中使用 Sanic。
    • 官方文档还包括了关于安全性、性能优化、部署等高级话题的深入讲解,是开发者必不可少的参考资料。

GitHub 源码学习

  • 阅读 Sanic 的开源代码,了解其设计与实现
    • Sanic 是一个开源项目,托管在 GitHub 上:Sanic GitHub
    • 阅读源码可以帮助我们深入理解 Sanic 的内部架构、请求处理流程以及中间件和路由的实现等。对于有一定开发经验的用户,源代码学习是提高框架使用能力的一个非常有效途径。
    • GitHub 上的 issues 和 pull requests 也能帮助我们了解框架的最新更新和常见问题,甚至可以参与到项目的贡献中。

社区与教程

  • 加入 Sanic 的官方论坛或社区

    • Sanic 拥有一个活跃的社区,可以通过访问官方论坛、GitHub 讨论区、Slack 频道等参与讨论,与其他开发者共享经验,寻求帮助。
    • 官方社区提供了很多开发者发布的教程和示例项目,适合不同水平的开发者学习和讨论。
  • 查阅社区提供的优秀教程和文章

    • 社区成员和第三方网站经常会发布关于 Sanic 的深度教程和实战项目。这些教程不仅覆盖了基础使用,还会深入讨论性能优化、异步编程、数据库集成等高级主题。
    • 一些著名的教程网站(如 Medium、Dev.to、Real Python 等)定期发布关于 Sanic 的文章,可以通过这些资源进一步提升自己的技能。

通过以上官方文档和社区资源的学习,能够使我们更深入地了解 Sanic 框架,及时了解框架的更新与变化,并在遇到困难时得到及时的支持和解决方案。


原文地址:https://blog.csdn.net/Dreaming_TI/article/details/145125993

免责声明:本站文章内容转载自网络资源,如侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!