FastAPI入门系列1-搭建项目基本架构

Cheng 2024-12-05 20:17:54
Categories： Tags：

FastAPI 是一个现代、快速（高性能）的 Web 框架，旨在帮助开发者快速构建可靠的 API。基于 Python 的类型注解，FastAPI 提供了强大的数据验证和文档生成功能，同时支持异步编程，使其在处理高并发请求时表现得尤为出色。得益于其优秀的性能和开发效率，FastAPI 在构建 RESTful API 和微服务架构时变得越来越流行。

在本系列博客中，我们将带领大家逐步了解如何使用 FastAPI 从零开始构建一个功能完备的 API 项目。通过实例演示，我们将深入讲解 FastAPI 的基本概念、常用功能以及最佳实践，帮助你快速上手并掌握这一框架的核心技术。

1.环境安装

首先，我们通过以下命令安装 FastAPI 及相关依赖包：

1
pip install fastapi uvicorn SQLAlchemy aiomysql redis gunicorn -i https://pypi.tuna.tsinghua.edu.cn/simple

2.项目文件结构

如下为一个基本的 FastAPI 项目结构，通过组织不同的功能模块和代码，使得项目更易于维护和扩展。

1
2
3
4
5
6
7
8
9
├── api
│   └── users.py      # 用户相关的 API 逻辑
├── forms
│   └── user_form.py  # Pydantic 表单模型，用于数据验证
├── utils
│   ├── depends.py    # 依赖注入工具
│   └── response.py   # HTTP 响应工具类
├── main.py           # FastAPI 应用启动文件
└── requirements.txt  # 项目依赖文件

3.main.py

在 main.py 文件中，我们通过 FastAPI 创建应用实例，并将不同的 API 路由（如 users）引入到应用中。

1
2
3
4
5
6
from fastapi import FastAPI, Request

from api import users

app = FastAPI()
app.include_router(users.users_router)  # 引用路由

4.api/users.py

在 users.py 中，我们定义了与用户相关的 API 路由。FastAPI 提供了非常灵活的方式来定义路径参数、查询参数、请求体参数，并支持详细的验证。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
from typing import Optional
from fastapi import APIRouter, Depends, Query, Body, Path

from forms.user_form import UserReq
from utils.depends import default_user
from utils.response import HttpResponse

users_router = APIRouter(
    prefix="/api/v1",  # 路径前缀
    tags=["users"],    # 文档分组
    responses={404: {"description": "Not found"}},  # 自定义某个状态码的响应内容
)

# 路径参数及验证
@users_router.get("/users_v1/{user_id}")
async def get_users_v1(
    user_id: int = Path(
        ..., 
        ge=1,  # 大于等于1（gt lt le ge）
        title="The ID of the user to get"
    )  
):
    return HttpResponse.success(data={"user_id": user_id})

# 查询参数及验证
@users_router.get("/users_v2/")
async def get_users_v2(
    p: str = Query(
        ...,                   # ... 代表必需填
        min_length=1,          # 最短为1
        max_length=20,         # 最长为20
  regex="^@",            # 以@开头
  title="Query string",  # 标题
  description="Query string to search",
  alias="params",        # url查询参数的别名
  deprecated=True,       # 文档中展示为已弃用
    ),  
 q: Optional[str] = Query(None)   # 不是必填，默认为None
):
    result = p + (q if q else '')
    return HttpResponse.success(data={"result": result})

# 请求体参数示例
@users_router.post(
    "/users_v3/",
 tags=["users"],          # 用于文档中的标签分组
 summary="Get an user",   # 简短描述
 description="Get an user detail",  # 详细的描述
    response_model=UserReq,            # 响应数据的模型类型
    response_model_exclude_unset=True,          # 忽略默认参数
    response_model_include={"name", "mobile"},  # 包含（忽略其他的）
 response_description="The user item",       # 对响应内容的描述
 deprecated=True          # 文档中展示为已弃用
)
async def get_users_v3(
    user: UserReq = Body(...)  
):
    return HttpResponse.success(data={"user": user})

# 依赖注入示例
@users_router.get("/users_v4/")
async def get_users_v4(user: UserReq = Depends(default_user)):
    return HttpResponse.success(data={"user": user})

5.forms/user_form.py

在 user_form.py 文件中，我们通过 Pydantic 创建了一个 UserReq 类，用于对请求体数据进行验证和序列化。

1
2
3
4
5
6
7
8
9
10
11
12
13
from typing import Optional
from pydantic import BaseModel, Field

class UserReq(BaseModel):
    name: str
    mobile: str
    age: int = Field(..., gt=0, description="The age must be greater than zero")
    introduction: Optional[str] = Field(
        None, 
        title="The introduction of the user", 
        max_length=300, 
        example="A very nice man"
    )

6.utils/depends.py

在 depends.py 文件中，我们定义了一个默认的用户对象，使用 FastAPI 的依赖注入系统返回一个 UserReq 对象。

default_user 函数创建了一个默认的用户对象，返回给调用者。它被用作依赖注入，可以在路由中通过 Depends() 获取。

1
2
3
4
5
6
7
8
9
10
from forms.user_form import UserReq

async def default_user() -> UserReq:
    user = UserReq(
        name='default',
        mobile='123',
        age=18,
        introduction='default'
    )
    return user

7.utils/response.py

在 response.py 文件中，我们定义了一个 HttpResponse 工具类，简化了响应的生成过程，并提供了常见的 HTTP 状态码。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse, Response
from typing import Any, Dict, Optional

class HttpResponse:
    """
    响应工具类
    """
    ok = 200
    unautherror = 401
    paramserror = 400
    servererror = 500

    @classmethod
    def response(cls, code: int, msg: str, data: Optional[Any], state: Optional[Any]) -> Response:
        result = {
            "code": code,
            "msg": msg,
            "data": data,
            "state": state
        }
        return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder(result))

    @classmethod
    def success(
        cls,
        msg: str = '操作成功',
        data: Optional[Dict] = None,
        state: Optional[Any] = None,
    ) -> Response:
        return cls.response(cls.ok, msg, data, state)

    @classmethod
    def unauth_error(
        cls,
        msg: str = '登录信息已过期',
        data: Optional[Dict] = None,
        state: Optional[Any] = None,
    ) -> Response:
        return cls.response(cls.unautherror, msg, data, state)

    @classmethod
    def params_error(
        cls,
        msg: str = '客户端参数错误',
        data: Optional[Dict] = None,
        state: Optional[Any] = None,
    ) -> Response:
        return cls.response(cls.paramserror, msg, data, state)
    

    @classmethod
    def server_error(
        cls,
        msg: str = '服务器内部错误',
        data: Optional[Dict] = None,
        state: Optional[Any] = None,
    ) -> Response:
        return cls.response(cls.servererror, msg, data, state)

8.测试

使用 curl 或浏览器文档来测试我们实现的 API 接口。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 启动服务
uvicorn main:app --reload --port 5000

# 查看文档，可以直接在文档中调试
http://127.0.0.1:5000/docs

# 路径参数验证
curl -X GET 127.0.0.1:5000/api/v1/users_v1/0

{"detail":[{"type":"greater_than_equal","loc":["path","user_id"],"msg":"Input should be greater than or equal to 1","input":"0","ctx":{"ge":1}}]}

# 查询参数及验证
curl -X GET 127.0.0.1:5000/api/v1/users_v2/?params=@hello

{"code":200,"msg":"操作成功","data":{"result":"@hello"},"state":null}

# 请求体参数示例
curl -H "Content-Type: application/json" -X POST -d '{"name":"myname", "mobile":"123", "age": 18}' 127.0.0.1:5000/api/v1/users_v3/

{"code":200,"msg":"操作成功","data":{"user":{"name":"myname","mobile":"123","age":18,"introduction":null}},"state":null}

# 依赖注入示例
curl -X GET 127.0.0.1:5000/api/v1/users_v4/

{"code":200,"msg":"操作成功","data":{"user":{"name":"default","mobile":"123","age":18,"introduction":"default"}},"state":null}
FastAPI入门系列2-连接SQLAlchemy异步数据库
Django入门系列8-项目部署