接口开发

API，全称为 应用程序编程接口（Application Programming Interface），是一组定义了不同软件组件之间如何交互的规则和协议。API 允许不同的软件系统通过预定义的接口进行通信，使得开发者能够利用现有的功能和服务，而无需从头开始构建。这种模块化的设计不仅提高了开发效率，还促进了软件生态系统的扩展和创新。

一个典型的 API 包含以下几个关键组成部分：

• 请求（Request）： 客户端向服务器发送的调用请求，通常包含请求的方法、URL、头信息和参数。
• 响应（Response）： 服务器对客户端请求的返回结果，包含状态码、头信息和数据载体。
• 端点（Endpoint）： API 提供的具体功能接口的 URL 地址。
• 方法（Method）： 定义在特定端点上可以执行的操作，如 GET、POST、PUT、DELETE 等。
• 数据格式（Data Format）： 传输数据的格式，常见的有 JSON、XML 等。

API 根据不同的分类标准可以分为 RESTful API 和 GraphQL API。

特性	RESTful API	GraphQL
端点数量	多个端点，每个资源一个 URL	单一端点
数据获取方式	固定的资源结构，可能需要多次请求	客户端指定数据结构，单次请求即可获取所需数据
灵活性	较低，资源结构固定	高，可动态查询和组合数据
缓存机制	适用于标准 HTTP 缓存机制	缓存策略复杂，需要自定义
实时性	不支持实时数据推送	支持订阅机制，实现实时数据更新
学习成本	较低，基于 HTTP 标准	较高，需要学习 GraphQL 查询语言和概念
类型系统	无明确的类型系统	强类型 Schema，支持类型安全和自动文档生成

任何一家公司都可以定义符合自己公司业务需求的 API，但是他们可能使用的都是同一套开发框架在 Python 中，有多种接口开发框架可供选择：

维度	Flask	FastAPI	Django
发布时间	2010 年	2018 年	2005 年
开发语言	Python	Python	Python
架构风格	微框架，灵活性高，插件丰富	微框架，基于 Starlette 和 Pydantic	大型框架，支持全栈开发，内置 ORM、Admin 等
性能	中等，依赖于 WSGI	高，基于 ASGI，支持异步特性	中等，依赖于 WSGI
异步支持	通过扩展支持，如 Quart	原生支持，使用 async/await	通过 Django 3.0+ 开始支持异步（仍在逐步完善）
数据验证	依赖扩展（如 Marshmallow 或 WTForms）	内置 Pydantic，支持强类型和自动验证	依赖于表单类（如 Forms 和 DRF 的 serializers）
自动文档生成	需依赖扩展（如 Swagger UI）	内置支持 OpenAPI 和 Swagger UI	不原生支持，需依赖 Django REST Framework (DRF)
学习曲线	简单，适合初学者	中等，需了解类型注解和异步编程	较高，包含多种内置组件，适合系统性学习
社区和生态	历史悠久，社区活跃，插件丰富	新兴框架，社区成长迅速，但生态略小	最悠久且成熟，生态完善，适合大型项目
可扩展性	高，通过插件或定制满足不同需求	高，强大的原生功能减少了扩展需求	中等，内置组件强大，但灵活性相对较低
依赖注入支持	无原生支持，需依赖第三方库	原生支持依赖注入	无原生支持
生产就绪	成熟稳定，广泛应用于生产环境	稳定性较高，但相比 Flask 和 Django 使用年限较短	成熟稳定，被广泛应用于大型生产系统
调试工具	内置基础调试工具	支持 Starlette 的调试功能	内置调试模式，适合开发测试
使用场景	中小型项目，快速开发的场景	高性能、异步处理或自动文档的场景	大型系统开发，支持复杂业务逻辑和多用户系统
ORM 支持	无原生支持，需借助 SQLAlchemy 等第三方	无原生支持，需借助 Tortoise-ORM 等	原生支持，提供 Django ORM
Admin 界面	无，需要自行开发	无，需要自行开发	原生支持，提供强大的 Admin 管理后台
支持的 Python 版本	Python 2 和 Python 3（旧版支持 Python 2）	Python 3.6+	Python 3.8+
对类型注解的支持	不支持，需自行实现类型注解功能	原生支持，强类型友好	不直接支持，但可结合第三方库使用

接口规范

REST API

REST API 是符合 REST（表述性状态转移）架构样式设计原则的 API。 因此，REST API 有时被称为 RESTful API。

EST API 几乎可以使用任何编程语言进行开发，并支持多种数据格式。 唯一的要求是它们要符合以下六个 REST 设计原则 - 也称为架构约束：

• 统一接口。 无论请求来自何处，对同一资源发出的所有 API 请求都应该看起来相同。 REST API 应确保同一条数据（例如用户的姓名或电子邮件地址）仅属于一个统一资源标识符 (URI)。 资源不应过大，但应包含客户可能需要的每一条信息。
• 客户端/服务器解耦。 在 REST API 设计中，客户端和服务器应用程序必须彼此完全独立。 客户端应用程序只需知道所请求资源的 URI 即可； 它不能以任何其他方式与服务器应用程序交互。 同样，除了通过 HTTP 将客户端应用程序传递到所请求的数据外，服务器应用程序不应修改客户端应用程序。
• 无状态。 REST API 是无状态的，这意味着每个请求都需要包含处理它所需的全部信息。 换句话说，REST API 不需要任何服务器端会话。 不允许服务器应用程序存储与客户端请求相关的任何数据。
• 可缓存性。 如果可能，资源应该可以在客户端或服务器端缓存。 服务器响应还需要包含有关是否允许对交付的资源进行缓存的信息。 目标是提高客户端的性能，同时增强服务器端的可扩展性。
• 分层系统架构。 在 REST API 中，调用和响应都会经过多个不同的层。 根据经验，请不要将客户端和服务器应用程序直接相互连接。 通信环路中可能包含多个不同的中介服务器。 需要设计 REST API，让客户端和服务器都无法判断它是与最终应用程序还是中介服务器进行通信。
• 按需编码（可选）。 REST API 通常发送静态资源，但在某些情况下，响应也可以包含可执行代码（例如 Java 小程序）。 在这些情况下，代码只应按需运行。

REST API 通过 HTTP 请求进行通信，以便执行标准数据库功能，例如在资源中创建、读取、更新和删除记录（这些操作也称为 CRUD）。 例如，REST API 可使用 GET 请求检索记录、使用 POST 请求创建记录、使用 PUT 请求更新记录以及使用 DELETE 请求删除记录。 可以在 API 调用中使用所有 HTTP 方法。 精心设计的 REST API 类似于在具有内置 HTTP 功能的 Web 浏览器中运行的 Web 站点。 任何特定时刻或时间戳的资源状态称为资源表示。 几乎可以采用任何格式将该信息传递到客户端，包括 JavaScript 对象表示法 (JSON)、HTML、XLT、Python、PHP 或纯文本。 JSON 非常受欢迎，因为它可以被人类和机器读取，而且与编程语言无关。 在 REST API 调用中，请求头和参数也很重要，因为它们包含重要的标识信息，例如元数据、授权、统一资源标识符 (URI)、缓存、cookie 等。 在精心设计的 REST API 中会使用请求头、响应头和常规 HTTP 状态码。

技术实现

Flask

import flask

app = flask.Flask(__name__)


@app.route("/get_example/<int:user_id>", methods=["GET"])
def get_example(user_id):
    """
    **方法：** GET

    **示例请求：**
    http:day3.ipynb
    GET /get_example/123?name=Alice HTTP/1.1
    Host: example.com
    User-Agent: Mozilla/5.0


    **示例响应：**
    json
    {
        "user_id": 123,
        "name": "Alice",
        "user_agent": "Mozilla/5.0"
    }

    """
    # 从查询参数获取 'name'
    name = flask.request.args.get("name", "默认名字")

    # 从 Headers 获取 'User-Agent'
    user_agent = flask.request.headers.get("User-Agent", "未知")

    return {"user_id": user_id, "name": name, "user_agent": user_agent}


@app.route("/post_example", methods=["POST"])
def post_example():
    """
    **方法：** POST

    **示例请求：**
    http
    POST /post_example HTTP/1.1
    Host: example.com
    Content-Type: application/json
    Authorization: Bearer token

    {
        "title": "示例标题",
        "content": "示例内容"
    }


    **示例响应：**
    json
    {
        "title": "示例标题",
        "content": "示例内容",
        "authorization": "Bearer token"
    }

    """
    # 从 JSON Body 获取数据
    data = flask.request.get_json()
    title = data.get("title", "默认标题")
    content = data.get("content", "默认内容")

    # 从 Headers 获取 'Authorization'
    auth = flask.request.headers.get("Authorization", "无授权")

    return {"title": title, "content": content, "authorization": auth}


@app.route("/put_example/<int:item_id>", methods=["PUT"])
def put_example(item_id):
    """
    **方法：** PUT

    **示例请求：**
    http
    PUT /put_example/456 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded
    X-Custom-Header: 自定义值

    name=新名称&description=新描述


    **示例响应：**
    json
    {
        "item_id": 456,
        "name": "新名称",
        "description": "新描述",
        "custom_header": "自定义值"
    }

    """
    # 从表单数据获取
    name = flask.request.form.get("name", "默认名称")
    description = flask.request.form.get("description", "默认描述")

    # 从 Headers 获取 'X-Custom-Header'
    custom_header = flask.request.headers.get("X-Custom-Header", "无自定义头")

    return {
        "item_id": item_id,
        "name": name,
        "description": description,
        "custom_header": custom_header,
    }


@app.route("/delete_example", methods=["DELETE"])
def delete_example():
    """
    **方法：** DELETE

    **示例请求（确认删除）：**
    http
    DELETE /delete_example?confirm=true HTTP/1.1
    Host: example.com
    Authorization: Bearer token


    **示例响应（确认删除）：**
    json
    {
        "deleted": true,
        "authorization": "Bearer token"
    }


    **示例请求（未确认删除）：**
    http
    DELETE /delete_example HTTP/1.1
    Host: example.com


    **示例响应（未确认删除）：**
    json
    {
        "message": "删除未确认"
    }

    HTTP 状态码：400
    """
    # 从查询参数获取 'confirm'
    confirm = flask.request.args.get("confirm", "false").lower() == "true"

    # 从 Headers 获取 'Authorization'
    auth = flask.request.headers.get("Authorization", "无授权")

    if not confirm:
        return {"message": "删除未确认"}, 400

    return {"deleted": confirm, "authorization": auth}


@app.route(
    "/api/resource/<string:resource_id>", methods=["GET", "POST", "PUT", "DELETE"]
)
def api_resource(resource_id):
    """
    **方法：** GET, POST, PUT, DELETE

    **GET 请求示例：**
    http
    GET /api/resource/abc123?detail=full HTTP/1.1
    Host: example.com


    **GET 响应示例：**
    json
    {
        "method": "GET",
        "resource_id": "abc123",
        "detail": "full"
    }


    **POST 请求示例：**
    http
    POST /api/resource/abc123 HTTP/1.1
    Host: example.com
    Content-Type: application/json

    {
        "key": "value"
    }


    **POST 响应示例：**
    json
    {
        "method": "POST",
        "resource_id": "abc123",
        "data": {
            "key": "value"
        }
    }


    **PUT 请求示例：**
    http
    PUT /api/resource/abc123 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded

    field=new%20value


    **PUT 响应示例：**
    json
    {
        "method": "PUT",
        "resource_id": "abc123",
        "data": {
            "field": "new value"
        }
    }


    **DELETE 请求示例：**
    http
    DELETE /api/resource/abc123?confirm=true HTTP/1.1
    Host: example.com


    **DELETE 响应示例：**
    json
    {
        "method": "DELETE",
        "resource_id": "abc123",
        "confirm": true
    }

    """
    if flask.request.method == "GET":
        # 从查询参数获取 'detail'
        detail = flask.request.args.get("detail", "basic")
        return {"method": "GET", "resource_id": resource_id, "detail": detail}

    elif flask.request.method == "POST":
        # 从 JSON Body 获取数据
        data = flask.request.get_json()
        return {"method": "POST", "resource_id": resource_id, "data": data}

    elif flask.request.method == "PUT":
        # 从表单数据获取
        data = flask.request.form.to_dict()
        return {"method": "PUT", "resource_id": resource_id, "data": data}

    elif flask.request.method == "DELETE":
        # 从查询参数获取 'confirm'
        confirm = flask.request.args.get("confirm", "false").lower() == "true"
        return {"method": "DELETE", "resource_id": resource_id, "confirm": confirm}

    else:
        return {"error": "不支持的HTTP方法"}, 405


if __name__ == "__main__":
    app.run(debug=True)

FastAPI

相较于 Flask，FastAPI 会自动生成 2 种风格的 API 文档，地址为：http://127.0.0.1:8000/docs和http://127.0.0.1:8000/redoc

from fastapi import FastAPI, Header, HTTPException, Query, Path
from pydantic import BaseModel
from typing import Optional, Dict

app = FastAPI()

# Pydantic Models
class PostExampleModel(BaseModel):
    title: Optional[str] = "默认标题"
    content: Optional[str] = "默认内容"

class PutExampleModel(BaseModel):
    name: Optional[str] = "默认名称"
    description: Optional[str] = "默认描述"

class ApiResourcePostModel(BaseModel):
    key: str

class ApiResourcePutModel(BaseModel):
    field: str

# GET /get_example/{user_id}
@app.get("/get_example/{user_id}")
def get_example(
    user_id: int,
    name: Optional[str] = Query("默认名字"),
    user_agent: Optional[str] = Header("未知")
):
    """
    **方法：** GET

    **示例请求：**
    http:day3.ipynb
    GET /get_example/123?name=Alice HTTP/1.1
    Host: example.com
    User-Agent: Mozilla/5.0

    **示例响应：**
    json
    {
        "user_id": 123,
        "name": "Alice",
        "user_agent": "Mozilla/5.0"
    }

    """
    return {"user_id": user_id, "name": name, "user_agent": user_agent}

# POST /post_example
@app.post("/post_example")
def post_example(
    data: PostExampleModel,
    authorization: Optional[str] = Header("无授权")
):
    """
    **方法：** POST

    **示例请求：**
    http
    POST /post_example HTTP/1.1
    Host: example.com
    Content-Type: application/json
    Authorization: Bearer token

    {
        "title": "示例标题",
        "content": "示例内容"
    }

    **示例响应：**
    json
    {
        "title": "示例标题",
        "content": "示例内容",
        "authorization": "Bearer token"
    }

    """
    return {"title": data.title, "content": data.content, "authorization": authorization}

# PUT /put_example/{item_id}
@app.put("/put_example/{item_id}")
def put_example(
    item_id: int,
    data: PutExampleModel,
    x_custom_header: Optional[str] = Header("无自定义头")
):
    """
    **方法：** PUT

    **示例请求：**
    http
    PUT /put_example/456 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded
    X-Custom-Header: 自定义值

    name=新名称&description=新描述

    **示例响应：**
    json
    {
        "item_id": 456,
        "name": "新名称",
        "description": "新描述",
        "custom_header": "自定义值"
    }

    """
    return {
        "item_id": item_id,
        "name": data.name,
        "description": data.description,
        "custom_header": x_custom_header,
    }

# DELETE /delete_example
@app.delete("/delete_example")
def delete_example(
    confirm: bool = Query(False),
    authorization: Optional[str] = Header("无授权")
):
    """
    **方法：** DELETE

    **示例请求（确认删除）：**
    http
    DELETE /delete_example?confirm=true HTTP/1.1
    Host: example.com
    Authorization: Bearer token

    **示例响应（确认删除）：**
    json
    {
        "deleted": true,
        "authorization": "Bearer token"
    }


    **示例请求（未确认删除）：**
    http
    DELETE /delete_example HTTP/1.1
    Host: example.com

    **示例响应（未确认删除）：**
    json
    {
        "message": "删除未确认"
    }

    HTTP 状态码：400
    """
    if not confirm:
        raise HTTPException(status_code=400, detail="删除未确认")
    return {"deleted": confirm, "authorization": authorization}

# GET /api/resource/{resource_id}
@app.get("/api/resource/{resource_id}")
def api_resource_get(
    resource_id: str,
    detail: Optional[str] = Query("basic")
):
    """
    **方法：** GET

    **请求示例：**
    GET /api/resource/abc123?detail=full HTTP/1.1
    Host: example.com

    **响应示例：**
    json
    {
        "method": "GET",
        "resource_id": "abc123",
        "detail": "full"
    }

    """
    return {"method": "GET", "resource_id": resource_id, "detail": detail}

# POST /api/resource/{resource_id}
@app.post("/api/resource/{resource_id}")
def api_resource_post(
    resource_id: str,
    data: ApiResourcePostModel
):
    """
    **方法：** POST

    **请求示例：**
    POST /api/resource/abc123 HTTP/1.1
    Host: example.com
    Content-Type: application/json

    {
        "key": "value"
    }

    **响应示例：**
    json
    {
        "method": "POST",
        "resource_id": "abc123",
        "data": {
            "key": "value"
        }
    }

    """
    return {"method": "POST", "resource_id": resource_id, "data": data.dict()}

# PUT /api/resource/{resource_id}
@app.put("/api/resource/{resource_id}")
def api_resource_put(
    resource_id: str,
    data: ApiResourcePutModel
):
    """
    **方法：** PUT

    **请求示例：**
    PUT /api/resource/abc123 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded

    field=new%20value

    **响应示例：**
    json
    {
        "method": "PUT",
        "resource_id": "abc123",
        "data": {
            "field": "new value"
        }
    }

    """
    return {"method": "PUT", "resource_id": resource_id, "data": data.dict()}

# DELETE /api/resource/{resource_id}
@app.delete("/api/resource/{resource_id}")
def api_resource_delete(
    resource_id: str,
    confirm: bool = Query(False)
):
    """
    **方法：** DELETE

    **请求示例：**
    DELETE /api/resource/abc123?confirm=true HTTP/1.1
    Host: example.com

    **响应示例：**
    json
    {
        "method": "DELETE",
        "resource_id": "abc123",
        "confirm": true
    }

    """
    return {"method": "DELETE", "resource_id": resource_id, "confirm": confirm}

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

下面是异步函数作为FastAPI的返回值的示例

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import random

app = FastAPI()

async def async_range(n):
    for i in range(n):
        yield f"<p>{i}</p>"
        await asyncio.sleep(random.randint(1, 3) * 1)

@app.get("/stream")
async def stream_numbers():
    return StreamingResponse(async_range(5), media_type="html", background=None)

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

Django

Django 的接口开发，需要先创建项目和应用，然后创建视图函数，最后在 urls.py 中配置路由。

虽然 Django 的接口开发比较繁琐，但是 Django 的接口开发比较规范，易于维护。

安装 Django 和 Django REST framework

pip install django djangorestframework

创建项目和应用

django-admin startproject myproject
cd myproject
django-admin startapp apiapp

项目结构如下：

myproject/
├── apiapp/
│   ├── __init__.py
│   ├── views.py
│   ├── urls.py
│   └── serializers.py
├── myproject/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── manage.py

在 myproject/settings.py 中添加 rest_framework 和 apiapp 到 INSTALLED_APPS 列表中：

INSTALLED_APPS = [
    ... # 保持原有配置
    'rest_framework',
    'apiapp',
]

定义序列化器，有助于验证和解析输入数据。

apiapp/serializers.py

from rest_framework import serializers

class PostExampleSerializer(serializers.Serializer):
    title = serializers.CharField(default="默认标题", required=False)
    content = serializers.CharField(default="默认内容", required=False)

class PutExampleSerializer(serializers.Serializer):
    name = serializers.CharField(default="默认名称", required=False)
    description = serializers.CharField(default="默认描述", required=False)

class ApiResourcePostSerializer(serializers.Serializer):
    key = serializers.CharField()

class ApiResourcePutSerializer(serializers.Serializer):
    field = serializers.CharField()

定义视图函数

apiapp/views.py

from rest_framework import status
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework.views import APIView

from .serializers import (
    PostExampleSerializer,
    PutExampleSerializer,
    ApiResourcePostSerializer,
    ApiResourcePutSerializer,
)

@api_view(['GET'])
def get_example(request, user_id):
    """
    **方法：** GET

    **示例请求：**
    GET /get_example/123?name=Alice HTTP/1.1
    Host: example.com
    User-Agent: Mozilla/5.0

    **示例响应：**    ```json
    {
        "user_id": 123,
        "name": "Alice",
        "user_agent": "Mozilla/5.0"
    }
    """
    name = request.query_params.get('name', '默认名字')
    user_agent = request.headers.get('User-Agent', '未知')
    return Response({"user_id": user_id, "name": name, "user_agent": user_agent})

@api_view(['POST'])
def post_example(request):
    """
    **方法：** POST

    **示例请求：**
    POST /post_example HTTP/1.1
    Host: example.com
    Content-Type: application/json
    Authorization: Bearer token

    {
        "title": "示例标题",
        "content": "示例内容"
    }

    **示例响应：**    ```json
    {
        "title": "示例标题",
        "content": "示例内容",
        "authorization": "Bearer token"
    }
    """
    serializer = PostExampleSerializer(data=request.data)
    if serializer.is_valid():
        title = serializer.validated_data.get('title', '默认标题')
        content = serializer.validated_data.get('content', '默认内容')
        auth = request.headers.get('Authorization', '无授权')
        return Response({"title": title, "content": content, "authorization": auth})
    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['PUT'])
def put_example(request, item_id):
    """
    **方法：** PUT

    **示例请求：**
    PUT /put_example/456 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded
    X-Custom-Header: 自定义值

    name=新名称&description=新描述

    **示例响应：**    ```json
    {
        "item_id": 456,
        "name": "新名称",
        "description": "新描述",
        "custom_header": "自定义值"
    }
    """
    serializer = PutExampleSerializer(data=request.data)
    if serializer.is_valid():
        name = serializer.validated_data.get('name', '默认名称')
        description = serializer.validated_data.get('description', '默认描述')
        custom_header = request.headers.get('X-Custom-Header', '无自定义头')
        return Response({
            "item_id": item_id,
            "name": name,
            "description": description,
            "custom_header": custom_header,
        })
    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['DELETE'])
def delete_example(request):
    """
    **方法：** DELETE

    **示例请求（确认删除）：**
    DELETE /delete_example?confirm=true HTTP/1.1
    Host: example.com
    Authorization: Bearer token

    **示例响应（确认删除）：**    ```json
    {
        "deleted": true,
        "authorization": "Bearer token"
    }    ```

    **示例请求（未确认删除）：**
    DELETE /delete_example HTTP/1.1
    Host: example.com

    **示例响应（未确认删除）：**    ```json
    {
        "message": "删除未确认"
    }    ```
    HTTP 状态码：400
    """
    confirm = request.query_params.get('confirm', 'false').lower() == 'true'
    auth = request.headers.get('Authorization', '无授权')
    if not confirm:
        return Response({"message": "删除未确认"}, status=status.HTTP_400_BAD_REQUEST)
    return Response({"deleted": confirm, "authorization": auth})

class ApiResourceView(APIView):
    """
    **方法：** GET, POST, PUT, DELETE

    **GET 请求示例：**
    GET /api/resource/abc123?detail=full HTTP/1.1
    Host: example.com

    **GET 响应示例：**    ```json
    {
        "method": "GET",
        "resource_id": "abc123",
        "detail": "full"
    }
    """

    def get(self, request, resource_id):
        detail = request.query_params.get('detail', 'basic')
        return Response({"method": "GET", "resource_id": resource_id, "detail": detail})

    def post(self, request, resource_id):
        """
        **POST 请求示例：**
        POST /api/resource/abc123 HTTP/1.1
        Host: example.com
        Content-Type: application/json

        {
            "key": "value"
        }

        **POST 响应示例：**        ```json
        {
            "method": "POST",
            "resource_id": "abc123",
            "data": {
                "key": "value"
            }
        }        ```
        """
        serializer = ApiResourcePostSerializer(data=request.data)
        if serializer.is_valid():
            data = serializer.validated_data
            return Response({"method": "POST", "resource_id": resource_id, "data": data})
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    def put(self, request, resource_id):
        """
        **PUT 请求示例：**
        PUT /api/resource/abc123 HTTP/1.1
        Host: example.com
        Content-Type: application/x-www-form-urlencoded

        field=new%20value

        **PUT 响应示例：**        ```json
        {
            "method": "PUT",
            "resource_id": "abc123",
            "data": {
                "field": "new value"
            }
        }        ```
        """
        serializer = ApiResourcePutSerializer(data=request.data)
        if serializer.is_valid():
            data = serializer.validated_data
            return Response({"method": "PUT", "resource_id": resource_id, "data": data})
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    def delete(self, request, resource_id):
        """
        **DELETE 请求示例：**
        DELETE /api/resource/abc123?confirm=true HTTP/1.1
        Host: example.com

        **DELETE 响应示例：**        ```json
        {
            "method": "DELETE",
            "resource_id": "abc123",
            "confirm": true
        }        ```
        """
        confirm = request.query_params.get('confirm', 'false').lower() == 'true'
        return Response({"method": "DELETE", "resource_id": resource_id, "confirm": confirm})

在 apiapp/urls.py 中配置路由

apiapp/urls.py

from django.urls import path
from . import views

urlpatterns = [
    path('get_example/<int:user_id>/', views.get_example, name='get_example'),
    path('post_example/', views.post_example, name='post_example'),
    path('put_example/<int:item_id>/', views.put_example, name='put_example'),
    path('delete_example/', views.delete_example, name='delete_example'),
    path('api/resource/<str:resource_id>/', views.ApiResourceView.as_view(), name='api_resource'),
]

在 myproject/urls.py 中包含 apiapp.urls

myproject/urls.py

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('apiapp.urls')),
]

运行项目

python manage.py runserver

接口的进阶使用

接口除了可以传输文本数据，还可传输图片、视频等数据，cv2 可以捕获屏幕，将两者结合起来，实现内网直播功能，可以在局域网内通过浏览器观看屏幕共享。

import os
from importlib import import_module
from flask import Flask, render_template, Response,render_template_string


from io import BytesIO
import cv2
from PIL import ImageGrab, Image
import time
import threading
try:
    from greenlet import getcurrent as get_ident
except ImportError:
    try:
        from thread import get_ident
    except ImportError:
        from _thread import get_ident


class CameraEvent(object):
    def __init__(self):
        self.events = {}

    def wait(self):
        ident = get_ident()
        if ident not in self.events:
            self.events[ident] = [threading.Event(), time.time()]
        return self.events[ident][0].wait()

    def set(self):
        now = time.time()
        remove = None
        for ident, event in self.events.items():
            if not event[0].isSet():
                event[0].set()
                event[1] = now
            else:
                if now - event[1] > 5:
                    remove = ident
        if remove:
            del self.events[remove]

    def clear(self):
        self.events[get_ident()][0].clear()


class BaseCamera(object):
    thread = None
    frame = None
    last_access = 0
    event = CameraEvent()

    def __init__(self):
        if BaseCamera.thread is None:
            BaseCamera.last_access = time.time()

            BaseCamera.thread = threading.Thread(target=self._thread)
            BaseCamera.thread.start()

            while self.get_frame() is None:
                time.sleep(0)

    def get_frame(self):
        BaseCamera.last_access = time.time()

        BaseCamera.event.wait()
        BaseCamera.event.clear()

        return BaseCamera.frame

    @staticmethod
    def frames():
        raise RuntimeError('Must be implemented by subclasses.')

    @classmethod
    def _thread(cls):
        print('Starting camera thread.')
        frames_iterator = cls.frames()
        for frame in frames_iterator:
            BaseCamera.frame = frame
            BaseCamera.event.set()
            time.sleep(0)
            if time.time() - BaseCamera.last_access > 10:
                frames_iterator.close()
                print('Stopping camera thread due to inactivity.')
                break
        BaseCamera.thread = None



class Camera(BaseCamera):
    video_source = 0

    @staticmethod
    def set_video_source(source):
        Camera.video_source = source

    @staticmethod
    def frames():
        camera = cv2.VideoCapture(Camera.video_source)
        if not camera.isOpened():
            raise RuntimeError('Error')

        while True:
            image = ImageGrab.grab()  # 获取屏幕数据
            # w, h = image.size
            image = image.resize((1366, 750), Image.ANTIALIAS)  # 图片缩放
            output_buffer = BytesIO()  # 创建二进制对象
            image.save(output_buffer, format='JPEG', quality=100)  # quality提升图片分辨率
            frame = output_buffer.getvalue()  # 获取二进制数据
            yield frame  # 生成器返回一张图片的二进制数据

app = Flask(__name__)


@app.route('/')
def index():
    """
    视图函数
    :return:
    """
    return render_template_string('''<html>

<head>
    <title>屏幕共享</title>
</head>

<body>
    <img src="{{ url_for('video_feed') }}">
</body>

</html>''')


def gen(camera):
    """
    流媒体发生器
    """
    while True:
        frame = camera.get_frame()

        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')


@app.route('/video_feed')
def video_feed():
    """流媒体数据"""
    return Response(gen(Camera()),
                    mimetype='multipart/x-mixed-replace; boundary=frame')


if __name__ == '__main__':
    ip_host = '127.0.0.1'# 本机ip地址
    ip_host2 = '0.0.0.0'# 内网ip地址
    app.run(threaded=True, host=ip_host2, port=80)