uvicorn

uvicorn #

Uvicorn 是由 Starlette 框架的作者编写的 ASGI 服务器，旨在提供高性能的异步请求处理能力。它使用 asyncio 库实现异步 I/O 操作，支持 HTTP 和 WebSocket 协议，可与各种 ASGI 应用程序框架（如 FastAPI、Django、Starlette 等）配合使用。

示例 #

# main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

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

配置选项 #

Uvicorn 提供了丰富的配置选项，以满足不同需求。可以通过命令行参数或配置文件来配置 Uvicorn 的行为。

以下是一些常用的配置选项：

--host：指定主机地址，默认为 127.0.0.1。
--port：指定端口号，默认为 8000。
--workers：指定工作进程数量，默认为 CPU 核心数的 1 倍。
--log-level：指定日志级别，默认为 info。
--reload：在代码修改时自动重新加载应用程序。

高级功能 #

在本节中，更深入地探讨 Python Uvicorn 的高级功能，并提供丰富的示例代码。

1 SSL 支持 #

Uvicorn 支持通过 SSL 加密来提供安全的通信。可以使用 --ssl-keyfile 和 --ssl-certfile 参数来指定 SSL 密钥文件和证书文件。

uvicorn main:app --ssl-keyfile key.pem --ssl-certfile cert.pem

2 WebSocket 支持 #

除了处理 HTTP 请求外，Uvicorn 还支持处理 WebSocket 连接，用于实时通信应用程序。可以在 FastAPI 中使用 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 中间件 #

Uvicorn 支持使用中间件来修改请求和响应，以及执行其他自定义操作。可以通过 --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 异步任务 #

Uvicorn 支持在异步 Web 服务中执行异步任务。可以在 FastAPI 应用程序中定义异步函数，并在其中执行耗时操作，而不会阻塞主事件循环。

# 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 服务 #

使用 Uvicorn 可以轻松构建异步 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 服务 #

Uvicorn 支持 WebSocket 协议，可以使用它来构建实时通信的 Web 应用程序。

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