FastAPI使用详解

零、前提python补充

1、列表：

2、元组和集合

3、字典

一、FastAPI的场景优势

二、FastAPI并发 async / await

三、环境变量

1、创建和使用环境变量

你在 shell（终端）中就可以创建和使用环境变量，并不需要用到 Python：

💬 你可以使用以下命令创建一个名为 MY_NAME 的环境变量
export MY_NAME="Wade Wilson"💬 然后，你可以在其他程序中使用它，例如
echo "Hello $MY_NAME"Hello Wade Wilson

2、在 Python 中读取环境变量

然后你可以调用这个 Python 程序：

💬 这里我们还没有设置环境变量
python main.py💬 因为我们没有设置环境变量，所以我们得到的是默认值Hello World from Python💬 但是如果我们事先创建过一个环境变量
export MY_NAME="Wade Wilson"💬 然后再次调用程序
python main.py💬 现在就可以读取到环境变量了Hello Wade Wilson from Python

3、特定程序设置环境变量

4、PATH 环境变量

四、虚拟环境

当你在 Python 工程中工作时，你可能会有必要用到一个虚拟环境（或类似的机制）来隔离你为每个工程安装的包。

五、对接指南

1、代码示例

也可以直接把代码复制到入口文件中main.py启动项目。

2、路径参数

3、在线文档

4、枚举的使用

from enum import Enumfrom fastapi import FastAPIclass ModelName(str, Enum):alexnet = "alexnet"resnet = "resnet"lenet = "lenet"app = FastAPI()@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):if model_name is ModelName.alexnet:return {"model_name": model_name, "message": "Deep Learning FTW!"}if model_name.value == "lenet":return {"model_name": model_name, "message": "LeCNN all the images"}return {"model_name": model_name, "message": "Have some residuals"}

5、查询参数

6、请求体

7、处理错误异常

但该异常将会被 unicorn_exception_handler 处理。

接收到的错误信息清晰明了，HTTP 状态码为 418，JSON 内容如下：

{“message”: “Oops! yolo did something. There goes a rainbow…”}

8、中间件

9、CORS（跨域资源共享）

9、数据库

from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, Boolean
from sqlalchemy.sql import func
from database import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True, autoincrement=True)username = Column(String(50), unique=True, index=True, nullable=False)email = Column(String(100), unique=True, index=True, nullable=False)hashed_password = Column(String(255), nullable=False)is_active = Column(Boolean, default=True)created_at = Column(DateTime(timezone=True), server_default=func.now())updated_at = Column(DateTime(timezone=True), onupdate=func.now())class Category(Base):__tablename__ = "categories"id = Column(Integer, primary_key=True, index=True, autoincrement=True)name = Column(String(50), unique=True, nullable=False)description = Column(Text)created_at = Column(DateTime(timezone=True), server_default=func.now())class Product(Base):__tablename__ = "products"id = Column(Integer, primary_key=True, index=True, autoincrement=True)name = Column(String(200), nullable=False)description = Column(Text)price = Column(Integer, nullable=False)  # 单位：分stock = Column(Integer, default=0)category_id = Column(Integer, ForeignKey("categories.id", ondelete="CASCADE"))created_at = Column(DateTime(timezone=True), server_default=func.now())is_active = Column(Boolean, default=True)

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
import models
from database import engine, get_db# 创建所有表
models.Base.metadata.create_all(bind=engine)app = FastAPI(title="MySQL FastAPI Demo")@app.get("/")
def read_root():return {"message": "MySQL FastAPI 演示项目"}@app.get("/health")
def health_check(db: Session = Depends(get_db)):try:# 测试数据库连接db.execute("SELECT 1")return {"status": "healthy","database": "connected","timestamp": "2024-01-01T00:00:00Z"}except Exception as e:raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE,detail=f"Database connection failed: {str(e)}")if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

六、部署

1、命令启动部署

这在大多数情况下都能正常运行。😎

例如，您可以使用该命令在容器、服务器等环境中启动您的 FastAPI 应用。

2、docker启动部署

依赖项:
你通常会在某个文件中包含应用程序的依赖项。

具体做法取决于你安装这些依赖时所使用的工具。

最常见的方法是创建一个requirements.txt文件，其中每行包含一个包名称和它的版本。

你当然也可以使用在关于 FastAPI 版本 中讲到的方法来设置版本范围。

例如，你的requirements.txt可能如下所示：

fastapi>=0.68.0,<0.69.0
pydantic>=1.8.0,<2.0.0
uvicorn>=0.15.0,<0.16.0

创建 FastAPI 代码:

Dockerfile:
现在在相同的project目录创建一个名为Dockerfile的文件:

# 
FROM python:3.9# 
WORKDIR /code# 
COPY ./requirements.txt /code/requirements.txt# 
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt# 
COPY ./app /code/app# 
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

七、问题总结

1、python是入口文件启动程序，如何启动其他接口文件供调用

要让其他文件中的接口生效，你必须正确导入并挂载 -> 会启动：

方法1：使用 APIRouter（推荐）

方法2：直接在 main.py 中导入整个应用

现在可以访问：

• http://localhost:8000/auth/login
• http://localhost:8000/auth/register

方法3：使用包结构（便于统一管理）

api/init.py：

# 集中导出所有路由
from .users import router as user_router
from .products import router as product_router
from .orders import router as order_router__all__ = ["user_router", "product_router", "order_router"]

main.py：

from fastapi import FastAPI
from api import user_router, product_router, order_routerapp = FastAPI()
app.include_router(user_router)
app.include_router(product_router)
app.include_router(order_router)

最佳实践示例：

api/items.py：

from fastapi import APIRouterrouter = APIRouter(prefix="/items", tags=["物品管理"])@router.get("/")
def list_items():return [{"id": 1, "name": "苹果"}]@router.get("/{item_id}")
def get_item(item_id: int):return {"id": item_id, "name": f"物品{item_id}"}

api/users.py：

from fastapi import APIRouterrouter = APIRouter(prefix="/users", tags=["用户管理"])@router.get("/")
def list_users():return [{"id": 1, "name": "张三"}]

main.py：

from fastapi import FastAPI
from api import items, users  # 导入所有路由模块app = FastAPI()# 一次性挂载所有路由
app.include_router(items.router)
app.include_router(users.router)if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

2、接口顺序问题