快速了解FastAPI与Uvicorn是什么？-编程知识

概念

什么是Uvicorn

Python Uvicorn 是一个快速的 ASGI（Asynchronous Server Gateway Interface）服务器，用于构建异步 Web 服务。它基于 asyncio 库，支持高性能的异步请求处理，适用于各种类型的 Web 应用程序。

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

Uvicorn is an ASGI web server implementation for Python. Uvicorn
supports HTTP/1.1 and WebSockets.

源码地址: uvicorn

安装命令

pip install "uvicorn[standard]"

安装的是uvicorn-0.29.0

配置选项

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

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

查看帮助命令：

uvicorn --help

什么是FastAPI

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.8+ based on standard Python type hints.

The key features are:

Fast: Very high performance, on par with NodeJS and Go (thanks to Starlette and Pydantic). One of the fastest Python frameworks available.
Fast to code: Increase the speed to develop features by about 200% to 300%. *
Fewer bugs: Reduce about 40% of human (developer) induced errors. *
Intuitive: Great editor support. Completion everywhere. Less time debugging.
Easy: Designed to be easy to use and learn. Less time reading docs.
Short: Minimize code duplication. Multiple features from each parameter declaration. Fewer bugs.
Robust: Get production-ready code. With automatic interactive documentation.
Standards-based: Based on (and fully compatible with) the open standards for APIs: OpenAPI (previously known as Swagger) and JSON Schema.

* estimation based on tests on an internal development team, building production applications.

安装命令

pip install fastapi==0.110.1

示例

同步http服务

源代码

from typing import Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strprice: floatis_offer: Union[bool, None] = None@app.get("/")
def read_root():return {"Hello": "World"}fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):return fake_items_db[skip: skip + limit]@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):return {"item_id": item_id, "q": q}@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):return {"item_name": item.name, "item_id": item_id}

启动命令

web1中python文件的名称web1.py

uvicorn web1:app --reload

输出如下：
在这里插入图片描述

查看swagger

http://127.0.0.1:8000/docs：自动交互API文档(由Swagger UI提供)
http://127.0.0.1:8000/redoc：备选自动文档(由 ReDoc提供)

在这里插入图片描述

异步http服务

asynccontextmanager

源代码

import asyncio
from asyncio import Queue
from contextlib import asynccontextmanager
from dataclasses import dataclass
from random import randint
from uuid import UUID, uuid1import uvicorn
from fastapi import FastAPIasync def commandA(queue: Queue):while True:print("Start to get task from a_queue")user_request: UserRequest = await queue.get()print(f"Got the task from the a_queue and guess will process {str(user_request.process_time)} seconds")await asyncio.sleep(user_request.process_time)queue.task_done()print(f"Completed task, task type is {user_request.type}, task id is {user_request.tracking_id}")async def commandB(queue: Queue):while True:print("Start to get task from b_queue")user_request: UserRequest = await queue.get()await asyncio.sleep(user_request.process_time)print(f"Got the task from the b_queue and guess will process {str(user_request.process_time)} seconds")queue.task_done()print(f"Completed task, task type is {user_request.type}, task id is {user_request.tracking_id}")async def commandC(queue: Queue):while True:print("Start to get task from c_queue")user_request: UserRequest = await queue.get()await asyncio.sleep(user_request.process_time)print(f"Got the task from the c_queue and guess will process {str(user_request.process_time)} seconds")queue.task_done()print(f"Completed task, task type is {user_request.type}, task id is {user_request.tracking_id}")@asynccontextmanager
async def lifespan(app: FastAPI):asyncio.create_task(commandA(a_queue))asyncio.create_task(commandB(b_queue))asyncio.create_task(commandC(c_queue))yieldapp = FastAPI(lifespan=lifespan)
# app = FastAPI()a_queue = Queue()
b_queue = Queue()
c_queue = Queue()@dataclass
class RequestBody:type: str@dataclass
class UserRequest:tracking_id: UUID# A,B,Ctype: str = ""process_time: int = None@app.post("/")
async def post_task(reqeust_body: RequestBody):user_request = UserRequest(tracking_id=uuid1(),type=reqeust_body.type,process_time=randint(1, 5))# put requests to queueif user_request.type == "A":await a_queue.put(user_request)elif user_request.type == "B":await b_queue.put(user_request)elif user_request.type == "C":await c_queue.put(user_request)print(f"Received task, task type is {user_request.type}, task id is {user_request.tracking_id}")return "Received your request"# @app.on_event("startup")
# def start_command_listener():
#     asyncio.create_task(commandA(a_queue))
#     asyncio.create_task(commandB(b_queue))
#     asyncio.create_task(commandC(c_queue))if __name__ == "__main__":uvicorn.run(app=app, port=9000)