【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 默认会对插入的内容进行转义,因此只需确保使用安全的模板渲染方式即可。
队列与任务调度
集成任务队列(如 Celery
或 Huey
)
Sanic 可以与异步任务队列集成,Celery
和 Huey
是常用的任务队列系统。可以使用这些队列来执行后台任务,而不会阻塞主线程。
以 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)进行身份验证,确保用户的请求安全。
-
用户注册: 用户提交用户名和密码,系统将密码加密存储并返回注册成功的响应。
-
用户登录: 用户通过用户名和密码进行登录,验证成功后生成 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>"
注意事项:
- 确保已安装
sanic
和sanic_jwt
库,可以使用pip install sanic sanic_jwt
进行安装。 - 替换
<your_jwt_token>
为登录成功后获得的有效 JWT Token。 - 在测试 API 功能时,确保 API 服务已启动,可以通过运行上述 Python 脚本启动服务。
验证步骤:
- 首先启动服务,运行 Python 脚本。
- 使用
curl
或 Postman 等工具发送注册请求,检查用户是否能成功注册。 - 发送登录请求,检查是否能成功登录并获取 JWT Token。
- 使用获取到的 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)
分布式任务管理系统
后台定时任务调度与分发
在分布式任务管理系统中,后台定时任务调度与分发是非常重要的。为了有效地处理异步任务,通常会使用 Celery
或 Huey
等任务队列工具。以下是 Huey
作为任务调度的示例,任务会根据指定的时间周期自动执行。
- 设置定时任务调度: 定时任务会定期执行一个特定的函数,例如发送邮件或进行数据备份。
from huey import RedisHuey
huey = RedisHuey('tasks')
# 每隔一分钟执行一次的定时任务
@huey.periodic_task(crontab(minute='*'))
def send_email_reminder():
print("Sending email reminder...")
- 后台任务分发: 在任务队列中,我们可以创建后台任务并将其分发到不同的工作者进程中处理。每个任务可以被异步执行,确保主应用不会被阻塞。
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)
后台任务的调度和执行:
任务一旦被分发到队列中,后台工作进程会从队列中取出任务并异步执行。Celery
和 Huey
都允许创建多个工作进程来并行处理任务,从而提高系统的吞吐量。
使用异步任务队列处理大量数据
处理大量数据时,常常会导致请求的延迟或服务器的负载过高。通过异步任务队列,可以将耗时的任务分离到后台执行,避免阻塞主应用。
在以下示例中,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)!