#!/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

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="查询失败")code>

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不存在")code>

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 不存在")code>

return { "status": "success", "message": "删除魔兽世界角色 id {id}成功"}

logger.py

get_wowinfo_all 对应了<code>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()

/get_wowinfo/{role}

/get_wowinfo/{role} 对应了<code>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="查询失败")code>

return info

create_wowinfo 对应了<code>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"

}

@app.post("/create_wowinfo/", response_model=WowInfo)

async def create_new_wowinfo(info: WowInfo):

logger.info("创建魔兽世界角色")

return create_wowinfo(info)

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

update_wowinfo/{id}

update_wowinfo/{id} 对应了<code>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"

}

@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不存在")code>

return existing_info

<code>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

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 不存在")code>

return { "status": "success", "message": "删除魔兽世界角色 id {id}成功"}

<code>MariaDB [wow]> select * from wow_info where id=666 \G;

Empty set (0.000 sec)

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")code>

@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)code>

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>code>

<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>

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