Python-FastAPI框架使用介绍_python fastapi

Python-FastAPI框架介绍

1. FastAPI简介

​ FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建基于 Python 的 API。它具有简单易用的特性，同时也提供了高度自动化的文档生成功能，使得开发者可以更加高效地构建和部署 API 服务。功能强大、易于使用且高性能的 Web 框架，适用于构建各种规模的 API 服务。它的简洁语法、自动生成文档和异步支持等特性使得开发 API 更加轻松和愉快

中文文档：https://fastapi.tiangolo.com/zh/
GitHub项目地址：https://github.com/tiangolo/fastapi

本篇文章会简单的介绍一下概念，然后通过一个完整的实战项目来展开介绍各个功能模块。

FastAPI 关键特性

1. 快速高效：基于 Starlette 和 Pydantic 构建，具有高性能和低延迟的特点，支持异步处理请求，利用 Python 的协程提高并发性能。
2. 简单易用：使用标准的 Python 类型注解来定义 API 的输入和输出参数，无需编写大量的文档和验证代码。
3. 自动生成文档：通过访问 /docs 路径可以查看自动生成的交互式 API 文档，包含了每个端点的详细说明、请求和响应的模型结构以及示例请求和响应。
4. 数据验证：利用 Pydantic 提供的数据验证功能，可以自动验证请求数据的格式和类型，并进行数据转换。
5. 依赖注入：支持依赖注入，可以方便地将依赖项注入到处理函数中，例如数据库连接、配置等。
6. 类型检查：利用 Python 类型提示和 Pydantic 的数据模型，可以在开发过程中进行类型检查，减少错误和调试时间。
7. 中间件支持：支持使用中间件扩展框架的功能，例如认证、日志记录等。
8. 标准化：支持 OpenAPI 规范，可以生成符合规范的 API 文档和客户端代码。

2. FastAPI安装

安装FastAPI模块

pip install fastapi
pip install uvicorn
1
2

uvicorn用于运行 Python 的异步 Web 应用程序，与许多流行的 Python 框架（如 FastAPI、Starlette 等）兼容，可以帮助开发者构建高效的异步 Web 服务

3. FastAPI入门体验

小试牛刀demo，跑一遍官方小示例

• 创建一个main.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :wangting_FastAPI
# @File     :main.py.py
# @Author   :wangting_666

from typing import Union
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

   async 关键字用于定义异步函数。异步函数可以在执行过程中暂停并允许其他代码执行，直到某些条件满足后再恢复执行。在 FastAPI 中，使用 async 可以使函数能够处理异步操作，例如异步的数据库查询、IO 操作等，以提高性能和并发能力。
在这个例子中，read_root 和 read_item 函数都是异步函数，它们使用了 async 关键字来定义。这样的函数可以通过 await 关键字调用其他异步函数，或者执行需要等待的异步操作，而不会阻塞整个应用程序的执行。
1
2

• 在 PyCharm 的 Terminal 中运行 uvicorn 命令启动 FastAPI 应用程序

uvicorn main:app --reload
1

这里的 main 是你的 Python 文件名（不含扩展名main方法的意思），app 是你创建的 FastAPI 应用程序实例。

uvicorn main:app --reload 命令含义如下:

• main：main.py 文件（一个 Python “模块”）。
• app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
• --reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

打开浏览器并验证访问：

http://localhost:8000/

http://localhost:8000/items/1

图例1

图例2

通过以上代码创建了一个具有以下功能的 API：

• 通过 路径 / 和 /items/{item_id} 接受 HTTP 请求。
• 以上 路径 都接受 GET 操作（也被称为 HTTP 方法）。
• /items/{item_id} 路径 有一个 路径参数 item_id 并且应该为 int 类型。
• /items/{item_id} 路径 有一个可选的 str 类型的 查询参数 q。

4. FastAPI交互式文档

访问 http://127.0.0.1:8000/docs 可打开API文档界面

5. FastAPI入门体验代码升级

小试牛刀demo，官方小示例迭代

main.py

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float
    is_offer: Union[bool, None] = None


@app.get("/")
def read_root():
    return {"hello": "world"}


@app.get("/item/{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}
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

服务器将会自动重载（因为在上面的步骤中你向 uvicorn 命令添加了 --reload 选项）

由于之前在 PyCharm 的 Terminal 中运行 uvicorn 命令启动 FastAPI 应用程序，有--reload参数，所以无需重新运行，如没有增加这个参数，需要在Terminal中重新执行uvicorn

打开URLhttp://localhost:8000/docs

找到put，点击 Try it out ； 可以填写参数并直接调用 API

图例1：

点击「Execute」按钮，用户界面将会和 API 进行通信，发送参数，获取结果并在屏幕上展示

图例2：

6. FastAPI路径介绍

fastapi支持各种请求方式：

7. 项目实战示例1 (CRUD)

​ 让我们一起探索一个完整的 FastAPI 项目实战！在这个项目中，帮助大家深入了解 FastAPI 框架的强大功能，并通过实际示例演示其灵活性和易用性。项目示例可以提供一个直观、易理解的教程，帮助您快速掌握 FastAPI 的核心概念和最佳实践。跟随我们一起探索，开启您的 FastAPI 之旅。通过一个魔兽世界游戏的职业信息库的增删改查需求来慢慢了解各个功能。

7-1. 项目目录结构

app0415

• app ( package包，功能模块 )
  • __init__.py
  • logger.py
  • main.py
  • todo.py
• bin （ 项目执行入口 ）
  • run_server.py
• config （ 配置文件 ）
  • config.py
• log （ 日志输出 ）
  • app.log
• tmp （存放临时文件用）
• requirements.txt （ 模块依赖包清单 ）

7-2. 准备工作

• 需要一个MySQL数据库

-- 建测试库
CREATE DATABASE `wow`;

use wow;

# 创建测试表
CREATE TABLE `wow_info` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '角色id',
  `role` varchar(255) DEFAULT NULL COMMENT '角色简称',
  `role_cn` varchar(255) DEFAULT NULL COMMENT '角色类型',
  `role_pinyin` varchar(255) DEFAULT NULL COMMENT '角色拼音',
  `zhuangbei` varchar(255) DEFAULT NULL COMMENT '装备类型',
  `tianfu` varchar(255) DEFAULT NULL COMMENT '天赋类型',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=667 DEFAULT CHARSET=utf8;

# 插入样例数据
INSERT INTO `wow_info` VALUES (1, 'fs', '法师', 'fashi', '布甲', '冰法|火法|奥法');
INSERT INTO `wow_info` VALUES (2, 'ms', '牧师', 'mushi', '布甲', '神牧|戒律|暗牧');
INSERT INTO `wow_info` VALUES (3, 'ss', '术士', 'shushi', '布甲', '毁灭|痛苦|恶魔');
INSERT INTO `wow_info` VALUES (4, 'dz', '盗贼', 'daozei', '皮甲', '狂徒|刺杀|敏锐');
INSERT INTO `wow_info` VALUES (5, 'ws', '武僧', 'wuseng', '皮甲', '酒仙|踏风|织雾');
INSERT INTO `wow_info` VALUES (6, 'xd', '德鲁伊', 'xiaode', '皮甲', '恢复|平衡|野性|守护');
INSERT INTO `wow_info` VALUES (7, 'dh', '恶魔猎手', 'emolieshou', '皮甲', '复仇|浩劫');
INSERT INTO `wow_info` VALUES (8, 'lr', '猎人', 'lieren', '锁甲', '兽王|生存|射击');
INSERT INTO `wow_info` VALUES (9, 'sm', '萨满', 'saman', '锁甲', '恢复|增强|元素');
INSERT INTO `wow_info` VALUES (10, 'long', '龙人', 'longren', '锁甲', '湮灭|恩护|增辉');
INSERT INTO `wow_info` VALUES (11, 'dk', '死亡骑士', 'siwangqishi', '板甲', '鲜血|冰霜|邪恶');
INSERT INTO `wow_info` VALUES (12, 'zs', '战士', 'zhanshi', '板甲', '武器|狂暴|防护');
INSERT INTO `wow_info` VALUES (13, 'sq', '圣骑士', 'shengqi', '板甲', '神圣|防护|惩戒');
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

• Postman

有条件建议安装一个Postman接口调试工具，非常实用方便

• pycharm开发工具和python环境

7-3. 代码介绍

run_server.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :app0415
# @File     :run_server.py
# @Time     :2024/4/15 19:46
# @Author   :wangting_666

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)
1
2
3
4
5
6
7
8
9
10
11

todo.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :app0415
# @File     :todo.py
# @Time     :2024/4/15 19:46
# @Author   :wangting_666


from typing import List, Optional
from pydantic import BaseModel
import pymysql

# MySQL连接配置
config = {
    'host': 'wangting_host_ip',
    'port': 3306,
    'user': 'root',
    'password': '123456',
    'database': 'wow',
}


# 连接到MySQL数据库
def connect_to_mysql():
    return pymysql.connect(**config)


# 定义WowInfo模型
class WowInfo(BaseModel):
    id: int
    role: str
    role_cn: str
    role_pinyin: str
    zhuangbei: str
    tianfu: str


# 获取所有魔兽职业信息
def get_wowinfo_all() -> List[WowInfo]:
    try:
        conn = connect_to_mysql()
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute("SELECT * FROM wow_info")
            info = [WowInfo(**row) for row in cursor.fetchall()]
        return info
    except Exception as e:
        print(f"查不到职业信息: {e}")
        return []


# 获取单个魔兽职业信息
def get_wowinfo(role: str) -> Optional[WowInfo]:
    try:
        conn = connect_to_mysql()
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute("SELECT * FROM wow_info WHERE role = %s", (role,))
            info = cursor.fetchone()
        return WowInfo(**info) if info else None
    except Exception as e:
        print(f"查不到职业信息: {e}")
        return None


# 创建魔兽职业信息
def create_wowinfo(wowinfo: WowInfo) -> WowInfo:
    try:
        conn = connect_to_mysql()
        with conn.cursor() as cursor:
            cursor.execute(
                "INSERT INTO wow_info (id,role, role_cn,role_pinyin,zhuangbei,tianfu) VALUES (%s, %s, %s,%s,%s,%s)",
                (wowinfo.id, wowinfo.role, wowinfo.role_cn, wowinfo.role_pinyin, wowinfo.zhuangbei, wowinfo.tianfu))
            conn.commit()
        return wowinfo
    except Exception as e:
        print(f"创建职业信息失败: {e}")
        return None


# 更新魔兽职业信息
def update_wowinfo(id: int, wowinfo: WowInfo) -> Optional[WowInfo]:
    try:
        conn = connect_to_mysql()
        with conn.cursor() as cursor:
            cursor.execute("UPDATE wow_info SET role=%s, role_cn=%s, role_pinyin=%s,zhuangbei=%s,tianfu=%s WHERE id=%s",
                           (wowinfo.role, wowinfo.role_cn, wowinfo.role_pinyin, wowinfo.zhuangbei, wowinfo.tianfu, id))
            conn.commit()
        return wowinfo
    except Exception as e:
        print(f"更新职业信息失败: {e}")
        return None


# 删除魔兽职业信息
def delete_wowinfo(id: int) -> bool:
    try:
        conn = connect_to_mysql()
        with conn.cursor() as cursor:
            cursor.execute("DELETE FROM wow_info WHERE id = %s", (id,))
            conn.commit()
        return True
    except Exception as e:
        print(f"删除职业信息失败: {e}")
        return False
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

main.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :app0415
# @File     :main.py
# @Time     :2024/4/15 19:46
# @Author   :wangting_666

from fastapi import FastAPI, HTTPException
from config.config import settings
from app.logger import logger
from app.todo import WowInfo, get_wowinfo_all, get_wowinfo, create_wowinfo, update_wowinfo, delete_wowinfo
from typing import List

app = FastAPI(title=settings.app_name)


@app.get("/get_wowinfo_all/", response_model=List[WowInfo])
async def read_wowinfos():
    logger.info("查询所有魔兽世界角色...")
    return get_wowinfo_all()


@app.get("/get_wowinfo/{role}", response_model=WowInfo)
async def read_wowinfo(role: str):
    logger.info(f"查询魔兽世界角色名称为: {role}...")
    info = get_wowinfo(role)
    if info is None:
        logger.error(f" 角色 {role} 不存在.")
        raise HTTPException(status_code=404, detail="查询失败")
    return info


@app.post("/create_wowinfo/", response_model=WowInfo)
async def create_new_wowinfo(info: WowInfo):
    logger.info("创建魔兽世界角色")
    return create_wowinfo(info)


@app.put("/update_wowinfo/{id}", response_model=WowInfo)
async def update_existing_wowinfo(id: int, wowinfo: WowInfo):
    logger.info(f"更新魔兽世界角色 id {id}...")
    existing_info = update_wowinfo(id, wowinfo)
    if existing_info is None:
        logger.error(f"职业信息ID:{id}不存在")
        raise HTTPException(status_code=404, detail="职业信息ID不存在")
    return existing_info


@app.delete("/delete_wowinfo/{id}")
async def delete_existing_wowinfo(id: int):
    logger.info(f"删除魔兽世界角色 id {id}...")
    success = delete_wowinfo(id)
    if not success:
        logger.error(f"兽世界角色 id {id} 不存在")
        raise HTTPException(status_code=404, detail="id 不存在")
    return {"status": "success", "message": "删除魔兽世界角色 id {id}成功"}
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

logger.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :app0415
# @File     :logger.py
# @Time     :2024/4/15 19:46
# @Author   :wangting_666

import logging
from config.config import settings
import os

log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../log")
os.makedirs(log_dir, exist_ok=True)

logging.basicConfig(filename=os.path.join(log_dir, "app.log"), level=settings.log_level)
logger = logging.getLogger(__name__)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

config.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :app0415
# @File     :config.py
# @Time     :2024/4/15 19:46
# @Author   :wangting_666

from pydantic import BaseSettings


class Settings(BaseSettings):
    app_name: str = "TodoApp"
    log_level: str = "INFO"


settings = Settings()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

7-4. 运行服务进行验证

7-4-1. 启动服务

pycharm方式

我这里是通过pycharm方式，如果是服务器运行，只需要执行python3 /home/wangting/run_server.py运行起来即可

开始功能测试之前，可以通过上方介绍的/docs接口查看自动生成的接口文档

http://127.0.0.1:8000/docs

7-4-2. 验证服务

后续验证服务，get请求可以通过浏览器、postman、接口的/docs功能均可；我这里使用postman进行演示，也可以通过上方介绍的/docs中try it out 中进行execute进行调试验证

• get_wowinfo_all

get_wowinfo_all 对应了get_wowinfo_all方法，@app.get可以看出是get请求，路由是/get_wowinfo_all

@app.get("/get_wowinfo_all/", response_model=List[WowInfo])
async def read_wowinfos():
    logger.info("查询所有魔兽世界角色...")
    return get_wowinfo_all()
1
2
3
4

• /get_wowinfo/{role}

/get_wowinfo/{role} 对应了get_wowinfo方法，@app.get可以看出也是get请求，路由是/get_wowinfo_all/{role}

@app.get("/get_wowinfo/{role}", response_model=WowInfo)
async def read_wowinfo(role: str):
    logger.info(f"查询魔兽世界角色名称为: {role}...")
    info = get_wowinfo(role)
    if info is None:
        logger.error(f" 角色 {role} 不存在.")
        raise HTTPException(status_code=404, detail="查询失败")
    return info
1
2
3
4
5
6
7
8

• create_wowinfo

create_wowinfo 对应了create_wowinfo方法，@app.post可以看出是post请求，路由是/create_wowinfo/

但与get请求不同的是，post请求需要提供body请求体

{
    "id": 666,
    "role": "create_role",
    "role_cn": "create_role_cn",
    "role_pinyin": "create_role_pinyin",
    "zhuangbei": "create_zhuangbei",
    "tianfu": "create_tianfu"
}
1
2
3
4
5
6
7
8

@app.post("/create_wowinfo/", response_model=WowInfo)
async def create_new_wowinfo(info: WowInfo):
    logger.info("创建魔兽世界角色")
    return create_wowinfo(info)
1
2
3
4

此时查看一下MySQL数据库，验证是否真的写入到数据库

• update_wowinfo/{id}

update_wowinfo/{id} 对应了update_wowinfo方法，@app.put可以看出是put请求，路由是/update_wowinfo/{id}

与post请求一样，也需要提供body请求体

{
    "id": 666,
    "role": "update_role",
    "role_cn": "update_role_cn",
    "role_pinyin": "update_role_pinyin",
    "zhuangbei": "update_zhuangbei",
    "tianfu": "update_tianfu"
}
1
2
3
4
5
6
7
8

@app.put("/update_wowinfo/{id}", response_model=WowInfo)
async def update_existing_wowinfo(id: int, wowinfo: WowInfo):
    logger.info(f"更新魔兽世界角色 id {id}...")
    existing_info = update_wowinfo(id, wowinfo)
    if existing_info is None:
        logger.error(f"职业信息ID:{id}不存在")
        raise HTTPException(status_code=404, detail="职业信息ID不存在")
    return existing_info
1
2
3
4
5
6
7
8

查询数据库数据，id=666的数据内容已经被修改

MariaDB [wow]> select * from wow_info where id=666 \G;
*************************** 1. row ***************************
         id: 666
       role: update_role
    role_cn: update_role_cn
role_pinyin: update_role_pinyin
  zhuangbei: update_zhuangbei
     tianfu: update_tianfu
1
2
3
4
5
6
7
8

• delete_wowinfo/{id}

delete_wowinfo/{id} 对应了delete_wowinfo方法，@app.delete可以看出是delete请求，路由是/delete_wowinfo/{id}

@app.delete("/delete_wowinfo/{id}")
async def delete_existing_wowinfo(id: int):
    logger.info(f"删除魔兽世界角色 id {id}...")
    success = delete_wowinfo(id)
    if not success:
        logger.error(f"兽世界角色 id {id} 不存在")
        raise HTTPException(status_code=404, detail="id 不存在")
    return {"status": "success", "message": "删除魔兽世界角色 id {id}成功"}
1
2
3
4
5
6
7
8

此时查询数据库数据id=666的数据内容已经不存在了

MariaDB [wow]> select * from wow_info where id=666 \G;
Empty set (0.000 sec)
1
2

7-4-3. 项目实战总结

​ 这个 FastAPI 项目展示了一个完整的后端应用程序，用于管理魔兽世界角色的信息。通过这个项目，我们深入了解了 FastAPI 框架的使用，以及如何构建 RESTful API。

​ 首先，我们学习了如何定义数据模型，使用 Pydantic 创建了 WowInfo 模型来表示魔兽世界角色的信息。然后，我们编写了一系列处理 CRUD 操作的函数，用于获取、创建、更新和删除角色信息。这些函数与 MySQL 数据库进行交互，通过 pymysql 库执行 SQL 查询和操作。

​ 接下来，我们使用 FastAPI 创建了一个基于路由的 Web 服务，定义了不同的端点来处理不同的 HTTP 请求。通过最常用的装饰器 @app.get、@app.post、@app.put 和 @app.delete，我们将每个端点与对应的处理函数绑定在一起。

​ 最后，我们使用了日志记录功能和配置设置，使项目更具可维护性和可扩展性。我们将日志记录到文件中，并使用配置类来管理应用程序的设置。

​ 这个项目为我们提供了一个完整的 FastAPI 实战示例，展示了如何使用 FastAPI 框架构建一个简单但功能完备的后端应用程序。通过这个示例，我们不仅学习了 FastAPI 框架的核心概念和基本用法，还掌握了与数据库交互、路由定义、日志记录等高级功能。

8. 项目实战示例2 (前后端)

8-1. 项目目录结构

• app0416
  • server_monitor_frontend
    • index.html
  • server_monitor_backend.py

示例是一个前后端协作，仅展示逻辑demo，没有加入太多的功能模块

在 app0416 目录下有一个名为 server_monitor_backend.py 的文件，用于启动后端 FastAPI 服务器。在同一目录下还有一个名为 server_monitor_frontend 的文件夹，其中包含一个名为 index.html 的文件，用于显示服务器监控系统的界面。如果需要自定义样式，可以在 server_monitor_frontend 文件夹中创建一个名为 style.css 的文件，并在 index.html 中引入，此目录主要用来开发前端功能。最终实现打开http://127.0.0.1:8000/，可以5秒刷新展示一次系统的服务器资源情况。

8-2. 代码介绍

• server_monitor_backend.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# @Project  :app0416
# @File     :server_monitor_backend.py.py
# @Time     :2024/4/16 19:58
# @Author   :wangting_666

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from starlette.responses import HTMLResponse, FileResponse

app = FastAPI()

# 将静态文件目录设置为 server_monitor_frontend 目录
app.mount("/static", StaticFiles(directory="server_monitor_frontend"), name="static")


@app.get("/", response_class=HTMLResponse)
async def get_index_page():
    return FileResponse("server_monitor_frontend/index.html")


@app.get("/stats")
async def get_server_stats():
    # Your server stats logic here
    return {"cpu_percent": 50, "memory_percent": 60, "disk_percent": 70}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)
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

• index.html

<!DOCTYPE html>
<html>
<head>
    <title>服务器资源监控</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            text-align: center;
        }
    </style>
</head>
<body>
<h1>Server Monitor</h1>
<div id="stats"></div>

<script>
    function fetchStats() {
        fetch('http://127.0.0.1:8000/stats')
            .then(response => response.json())
            .then(data => {
                document.getElementById('stats').innerHTML = `
                        <p><h3>CPU使用率: ${data.cpu_percent}%</h3></p>
                        <p><h3>内存使用率: ${data.memory_percent}%</h3></p>
                        <p><h3>磁盘使用率: ${data.disk_percent}%</h3></p>
                    `;
            })
            .catch(error => console.error('Error fetching stats:', error));
    }

    fetchStats();
    setInterval(fetchStats, 5000); // Fetch stats every 5 seconds
</script>
</body>
</html>
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

8-3. 运行服务进行验证

运行server_monitor_backend.py

打开URL ： http://127.0.0.1:8000/

8-4. 前后端示例总结

这个案例演示了如何使用 FastAPI 构建一个简单的服务器监控系统。后端 FastAPI 提供了 /stats 路由，用于获取服务器的监控数据，而前端界面通过 JavaScript 定时获取数据并动态显示在页面上。这个示例结合了后端和前端技术，展示了如何利用 FastAPI 快速构建 API，以及如何使用 JavaScript 实现动态页面交互。