大家好,今天为大家分享一个无敌的 Python 库 – uvicorn。 Github地址:https://github.com/encode/uvicorn Python Uvicorn 是一个快速的 ASGI(Asynchronous Server Gateway Interface)服务器,用于构建异步 Web 服务。它基于 asyncio 库,支持高性能的异步请求处理,适用于各种类型的 Web 应用程序。本文将介绍 Uvicorn 的基本概念、使用方法以及一些实际示例,帮助快速上手构建异步 Web 服务。 Uvicorn 是由 Starlette 框架的作者编写的 ASGI 服务器,旨在提供高性能的异步请求处理能力。它使用 asyncio 库实现异步 I/O 操作,支持 HTTP 和 WebSocket 协议,可与各种 ASGI 应用程序框架(如 FastAPI、Django、Starlette 等)配合使用。 可以使用 pip 包管理工具来安装 Uvicorn: 安装完成后,就可以在命令行中使用 下面是一个简单的示例,演示了如何使用 Uvicorn 启动一个异步 Web 服务: 保存以上代码到 这将启动一个名为 Uvicorn 提供了丰富的配置选项,以满足不同需求。可以通过命令行参数或配置文件来配置 Uvicorn 的行为。 以下是一些常用的配置选项: 在本节中,更深入地探讨 Python Uvicorn 的高级功能,并提供丰富的示例代码。 Uvicorn 支持通过 SSL 加密来提供安全的通信。可以使用 除了处理 HTTP 请求外,Uvicorn 还支持处理 WebSocket 连接,用于实时通信应用程序。可以在 FastAPI 中使用 Uvicorn 支持使用中间件来修改请求和响应,以及执行其他自定义操作。可以通过 Uvicorn 支持在异步 Web 服务中执行异步任务。可以在 FastAPI 应用程序中定义异步函数,并在其中执行耗时操作,而不会阻塞主事件循环。 可以通过自定义异常处理器来处理异常情况,如未找到页面、服务器错误等。 使用 Uvicorn 可以轻松构建异步 API 服务,处理大量并发请求,提高系统的性能和吞吐量。 Uvicorn 支持 WebSocket 协议,可以使用它来构建实时通信的 Web 应用程序。 通过本文的介绍,了解了如何使用 Python Uvicorn 构建异步 Web 服务。学会了安装 Uvicorn、启动简单的 Web 应用程序、配置 Uvicorn 的行为、使用高级功能和处理实际应用场景。现在,可以开始使用 Uvicorn 构建自己的异步 Web 服务了!
什么是 Uvicorn?
安装 Uvicorn
pip install uvicorn
uvicorn
命令了。使用示例
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}
main.py
文件中。然后,在命令行中执行以下命令:uvicorn main:app --reload
main
的 ASGI 应用程序,使用 Uvicorn 服务器运行在本地主机的默认端口 8000 上,并监听根路径 /
的 GET 请求。在浏览器中访问 http://localhost:8000
,将看到 “Hello, World!” 的消息。配置选项
--host
:指定主机地址,默认为 127.0.0.1
。--port
:指定端口号,默认为 8000
。--workers
:指定工作进程数量,默认为 CPU 核心数的 1 倍。--log-level
:指定日志级别,默认为 info
。--reload
:在代码修改时自动重新加载应用程序。高级功能
1 SSL 支持
--ssl-keyfile
和 --ssl-certfile
参数来指定 SSL 密钥文件和证书文件。uvicorn main:app --ssl-keyfile key.pem --ssl-certfile cert.pem
2 WebSocket 支持
WebSocket
类来处理 WebSocket 连接。# websocket_app.py
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
3 中间件
--middleware
参数来指定中间件。# middleware.py
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from uvicorn.middleware.debug import DebugMiddleware
from fastapi import FastAPI
app = FastAPI()
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["10.0.0.1"])
app.add_middleware(DebugMiddleware)
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}
4 异步任务
# async_task.py
from fastapi import FastAPI
import asyncio
app = FastAPI()
async def background_task():
while True:
print("Background task is running...")
await asyncio.sleep(5)
@app.on_event("startup")
async def startup_event():
asyncio.create_task(background_task())
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}
5 自定义错误处理
# error_handling.py
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 42:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
实际应用场景
1 异步 API 服务
# async_api.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
2 Websocket 服务
# websocket_server.py
import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
await websocket.send(message)
start_server = websockets.serve(echo, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
总结
发表评论 取消回复