怎样使用 Python 和 FastAPI 创建 RESTful API

介绍

FastAPI 是一个用于构建 API 和 Web 应用程序的 Python Web 框架。 它支持现代 Python 功能,如异步处理和类型提示,使其快速高效。 此外,它使用异步服务器网关接口 (ASGI) 标准进行异步、并发客户端连接,如果需要,它可以与 WSGI 一起使用。

API 是应用程序编程接口的首字母缩写。 它充当中介,允许应用程序相互交谈,成为应用程序之间交换数据或触发动作的媒介。 虽然 API 可以通过许多不同的协议进行通信,但本文指的是通过 HTTP 协议进行通信的 Web API。

REST (Representational State Transfer) 是一组架构约束,用于构建 API,而不是协议或标准。 它没有明确规定怎样创建 API。 相反,它引入了称为约束的最佳实践。 一种 RESTful API 是符合 REST 架构的 API。

对 REST API 发出的任何请求都包含四个基本部分:方法、端点、标头和正文。

  1. HTTP方法 描述动作或操作。

    • 邮政 – 创建资源
    • 得到 – 检索资源
    • – 更新资源
    • 删除 – 删除资源
  2. 端点 包含用于在 Web 上定位资源的 URI(统一资源标识符)。

  3. 标头 包含身份验证信息,例如 API 密钥。

  4. 身体 包含数据或任何附加信息。

本文演示了使用 Python 和 FastAPI 构建基本 RESTful API 的步骤。 所呈现的应用程序允许用户访问数据库中存储的对象并对其执行 CRUD 操作。 以下是显示给定应用程序端点结构的参考表。

端点方法描述
/学生得到列出所有学生对象
/学生邮政添加新的学生对象
/学生/{student_id}得到返回一个学生对象
/学生/{student_id}更新学生对象
/学生/{student_id}删除删除学生对象

先决条件

  • 部署一个全新的 Ubuntu 20.04 服务器
  • 创建非根 sudo 用户

设置 FastAPI 环境

  1. 初始化项目目录。

    为您的项目创建一个文件夹。

    $ mkdir ~/restful_demo
    

    打开项目目录。

    $ cd ~/restful_demo
    
  2. 创建虚拟环境。

    虚拟环境是用于依赖管理和项目隔离的 Python 工具。 每个项目都可以在一个独立的目录中本地安装任何包。

    安装 Python 虚拟环境包。

    $ sudo apt update
    $ sudo apt install python3-venv
    

    为您的项目创建 Python 虚拟环境。

    $ python3 -m venv venv
    

    Enter 虚拟环境。

    $ source venv/bin/activate
    

    Exit 虚拟环境。

    (venv) $ deactivate
    
  3. 安装依赖项。

    安装 wheel Python 包。

    (venv) $ pip install wheel
    

    安装 fastapi Python 包。

    (venv) $ pip install fastapi[all]
    

创建一个端点以列出所有学生

创建一个名为的新文件 app.py.

$ nano app.py

导入 FastAPI 类从 fastapi 库并将实例分配给名为的变量 app.

from fastapi import FastAPI

app = FastAPI()

通常,REST API 使用数据库服务器来存储数据,例如 PostgreSQL 或 MongoDB。 由于数据库选择取决于用户偏好,因此本文演示了使用 Python 列表作为内存数据库的步骤。 您可以使用相同的原理在任何数据库中实现它。

通过将虚拟学生对象列表分配给名为的变量来创建内存数据库 students.

students = [
    {'name': 'Student 1', 'age': 20},
    {'name': 'Student 2', 'age': 18},
    {'name': 'Student 3', 'age': 16}
]

创建要处理的路由 得到 端点上的请求 /students.

@app.get('/students')
def user_list():
    return {'students': students}

@app.get('/students') 装饰器告诉 FastAPI 执行 user_list 函数,它返回存储在数据库中的所有学生对象。 此路由允许用户通过发送一个列表来列出所有可用的学生对象 得到 上的请求 /students 端点。

Save 该文件使用 CTRL + X 然后 ENTER。

使用部署 FastAPI 应用程序 uvicorn.

(venv) $ uvicorn app:app --debug

在新终端中发送测试请求。

curl https://127.0.0.1:8000/students

预期输出:

{"students":[{"name":"Student 1","age":20},{"name":"Student 2","age":18},{"name":"Student 3","age":16}]}

停止 uvicorn 服务器使用 CTRL + C。

在列表端点中添加过滤功能

本节演示更新列表端点和添加功能以根据用户提供的年龄范围过滤输出的步骤。

打开在上一节中创建的 FastAPI 应用程序。

$ nano app.py

进口 Optional 类从 typing 图书馆。

from typing import Optional

将给定的行放在您导入的第一行下方 FastAPI 班级。

更新 user_list 函数接受最小/最大参数并重新排列函数内的代码以返回过滤列表。

@app.get('/students')
def user_list(min: Optional[int] = None, max: Optional[int] = None):

    if min and max:

        filtered_students = list(
            filter(lambda student: max >= student['age'] >= min, students)
        )

        return {'students': filtered_students}

    return {'students': students}

对列表端点所做的修改使用户能够使用 min 和 max 参数传递一系列年龄来过滤学生对象列表。 这 user_list 函数有两个可选参数:min 和 max。 内置 Python 函数 filter 遍历数据库中的每个学生对象,检查它们是否在请求的范围之间。 如果缺少最小/最大参数,端点将返回存储在数据库中的所有学生对象。

Save 该文件使用 CTRL + X 然后 ENTER。

使用部署 FastAPI 应用程序 uvicorn.

(venv) $ uvicorn app:app --debug

在新终端中发送测试请求。

curl "https://127.0.0.1:8000/students?min=16&max=18"

预期输出:

{"students":[{"name":"Student 2","age":18},{"name":"Student 3","age":16}]}

停止 uvicorn 服务器使用 CTRL + C。

创建一个函数来检查学生对象是否存在

本节演示创建名为的函数的步骤 user_check,它检查请求的用户是否存在于数据库中。

打开 FastAPI 应用程序。

$ nano app.py

导入 HTTPException 类从 fastapi 图书馆。

from fastapi import FastAPI, HTTPException

编辑您导入的行 FastAPI 类导入另一个类。

创建一个名为的新函数 student_check 这需要 student_id 范围。

def student_check(student_id):
    if not students[student_id]:
        raise HTTPException(status_code=404, detail="Student Not Found")

在以下步骤中,对单个学生对象执行操作的路由使用 student_check 函数来验证是否 student_id 提供的存在于数据库中。

创建端点以获取单个学生对象

本节演示添加新端点的步骤,使用户能够从数据库中获取单个学生对象。

打开 FastAPI 应用程序。

$ nano app.py

创建要处理的路由 得到 端点上的请求 /students/{student_id}.

@app.get('/students/{student_id}')
def user_detail(student_id: int):
    student_check(student_id)
    return {'student': students[student_id]}

@app.get('/students/{student_id}') 装饰器告诉 FastAPI 执行 user_detail 函数,它返回用户请求的所有学生对象。 此路由使用户能够通过发送一个 得到 上的请求 /students/{student_id} 端点。 如果 student_id 用户提供的索引中不存在 students 列表(内存数据库),然后函数返回带有 404 错误代码的 HTTP 异常。

Save 该文件使用 CTRL + X 然后 ENTER。

使用部署 FastAPI 应用程序 uvicorn.

(venv) $ uvicorn app:app --debug

在新终端中发送测试请求。

curl https://127.0.0.1:8000/students/0

预期输出:

{"student":{"name":"Student 1","age":20}}

停止 uvicorn 服务器使用 CTRL + C。

创建一个模型来处理 POST/UPDATE 请求正文

本节演示添加新类以定义用于处理 POST/UPDATE 请求的请求正文的模式的步骤。

打开在上一节中创建的 FastAPI 应用程序。

$ nano app.py

进口 BaseModel 类从 pydantic 图书馆。

from pydantic import BaseModel

将给定的行放在您导入的第二行下方 Optional 班级。

继承一个名为的新类 StudentBaseModel 并添加所需的属性。

class Student(BaseModel):
    name: str
    age: int

在以下步骤中,添加/更新学生对象的路由使用此模式来验证用户提供的值。

创建端点以添加新的学生对象

本节演示添加新端点的步骤,使用户能够将新学生对象添加到数据库。

打开 FastAPI 应用程序。

$ nano app.py

创建要处理的路由 邮政 端点上的请求 /students/{student_id}.

@app.post('/students')
def user_add(student: Student):
    students.append(student)

    return {'student': students[-1]}

@app.post('/students') 装饰器告诉 FastAPI 执行 user_add 函数,它在使用 Student 模式并返回新添加的学生对象。 该路由允许用户通过发送一个新的学生对象到数据库中 邮政 上的请求 /students 包含要添加的学生对象的数据的端点。

Save 该文件使用 CTRL + X 然后 ENTER。

使用部署 FastAPI 应用程序 uvicorn.

(venv) $ uvicorn app:app --debug

在新终端中发送测试请求。

curl -X 'POST' https://127.0.0.1:8000/students -H 'Content-Type: application/json' -d '{"name":"New Student", "age": 24}'

预期输出:

{"student":{"name":"New Student","age":24}}

停止 uvicorn 服务器使用 CTRL + C。

创建端点以更新学生对象

本节演示添加新端点的步骤,使用户能够更新数据库中的现有学生对象。

打开 FastAPI 应用程序。

$ nano app.py

创建要处理的路由 端点上的请求 /students/{student_id}.

@app.put('/students/{student_id}')
def user_update(student: Student, student_id: int):
    student_check(student_id)
    students[student_id].update(student)

    return {'student': students[student_id]}

@app.put('/students/{student_id}') 装饰器告诉 FastAPI 执行 user_update 函数,它使用提供的更新数据库中的学生对象 student_id 使用验证后作为索引 Student 模式并返回新更新的学生对象。 此路由使用户能够通过发送一个更新现有的学生对象 上的请求 /students/{student_id} 包含要更新的学生对象的数据的端点。 如果 student_id 用户提供的索引中不存在 students 列表(内存数据库),然后函数返回带有 404 错误代码的 HTTP 异常。

Save 该文件使用 CTRL + X 然后 ENTER。

使用部署 FastAPI 应用程序 uvicorn.

(venv) $ uvicorn app:app --debug

在新终端中发送测试请求。

curl -X 'PUT' https://127.0.0.1:8000/students/0 -H 'Content-Type: application/json' -d '{"name":"Student X", "age": 18}'

预期输出:

{"student":{"name":"Student X","age":18}}    

停止 uvicorn 服务器使用 CTRL + C。

创建端点以删除学生对象

本节演示添加新端点的步骤,使用户能够从数据库中删除学生对象。

打开 FastAPI 应用程序。

$ nano app.py

创建要处理的路由 删除 端点上的请求 /students/{student_id}.

@app.delete('/students/{student_id}')
def user_delete(student_id: int):
    student_check(student_id)
    del students[student_id]

    return {'students': students}

@app.delete('/students/{student_id}') 装饰器告诉 FastAPI 执行 user_delete 函数,它通过使用提供的删除数据库中的请求学生对象 student_id 作为索引。 该路由允许用户通过发送一个从数据库中删除一个学生对象 删除 上的请求 /students/{student_id} 端点。 如果 student_id 用户提供的索引中不存在 students 列表(内存数据库),然后该函数返回带有 404 错误代码的 HTTP 异常。

Save 该文件使用 CTRL + X 然后 ENTER。

使用部署 FastAPI 应用程序 uvicorn.

(venv) $ uvicorn app:app --debug

在新终端中发送测试请求。

curl -X 'DELETE' https://127.0.0.1:8000/students/0

预期输出:

{"students":[{"name":"Student 2","age":18},{"name":"Student 3","age":16}]}

停止 uvicorn 服务器使用 CTRL + C。

最终代码

这是最终的演示应用程序代码。 您可以将其用作故障排除的参考。

from fastapi import FastAPI, HTTPException
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

students = [
    {'name': 'Student 1', 'age': 20},
    {'name': 'Student 2', 'age': 18},
    {'name': 'Student 3', 'age': 16}
]

class Student(BaseModel):
    name: str
    age: int

@app.get('/students')
def user_list(min: Optional[int] = None, max: Optional[int] = None):

    if min and max:

        filtered_students = list(
            filter(lambda student: max >= student['age'] >= min, students)
        )

        return {'students': filtered_students}

    return {'students': students}

@app.get('/students/{student_id}')
def user_detail(student_id: int):
    student_check(student_id)
    return {'student': students[student_id]}

@app.post('/students')
def user_add(student: Student):
    students.append(student)

    return {'student': students[-1]}

@app.put('/students/{student_id}')
def user_update(student: Student, student_id: int):
    student_check(student_id)
    students[student_id].update(student)

    return {'student': students[student_id]}

@app.delete('/students/{student_id}')
def user_delete(student_id: int):
    student_check(student_id)
    del students[student_id]

    return {'students': students}

def student_check(student_id):
    if not students[student_id]:
        raise HTTPException(status_code=404, detail="Student Not Found")

结论

在本文中,您了解了 REST API 的基础知识以及怎样使用 Python 和 FastAPI 自己创建一个。 要将您的 FastAPI 应用程序部署到生产环境,请按照步骤在 Ubuntu 20.04 上使用 Gunicorn 和 Nginx 部署 FastAPI 应用程序。 更多FastAPI相关信息,请访问官方 FastAPI 网站.

注:本教程在Vultr VPS上测试通过,如需部署请前往Vultr.com