說明：
我計劃用python的fastapi框架，實現操作MySQL數據庫的表，實現增刪改查的操作，并且在postman里面測試
step1: 安裝數據庫依賴

pip install fastapi uvicorn pymysql

step2:C:\Users\Administrator\PycharmProjects\FastAPIProject\main.py

from fastapi import FastAPI, HTTPException, Body, Path
import pymysql.cursors
from typing import Optional, Dict
app = FastAPI()
# 數據庫連接配置
DB_CONFIG = {'host': 'localhost','user': 'root','password': '123456','db': 'db_spring','charset': 'utf8mb4','cursorclass': pymysql.cursors.DictCursor
}
# 查詢數據庫的函數
def query_database(query: str, params=None):try:connection = pymysql.connect(**DB_CONFIG)with connection.cursor() as cursor:cursor.execute(query, params)result = cursor.fetchall()connection.close()return resultexcept Exception as e:raise HTTPException(status_code=500, detail=str(e))
# 查詢表數據的 API 端點
@app.get("/query")
async def query_table(table_name: str):query = f"SELECT * FROM {table_name}"try:data = query_database(query)return {"status": "success", "data": data}except HTTPException as e:return {"status": "error", "message": e.detail}# 新增用戶 (POST)
@app.post("/users")
async def create_user(user_data: Dict = Body(..., example={"name": "諸葛亮","email": "zhugeliang@shu.com","age": 54})
):try:# 檢查郵箱唯一性check_query = "SELECT id FROM users WHERE email = %s"exist = query_database(check_query, (user_data["email"],))if exist:raise HTTPException(409, "郵箱已存在")# 執行插入操作insert_query = """INSERT INTO users (name, email, age)VALUES (%s, %s, %s)"""with pymysql.connect(**DB_CONFIG) as conn:with conn.cursor() as cursor:cursor.execute(insert_query, (user_data["name"],user_data["email"],user_data["age"]))new_id = cursor.lastrowidconn.commit()return {"status": "success", "id": new_id}except KeyError as e:raise HTTPException(400, f"缺失必要字段: {e}")# 更新用戶 (PUT)
@app.put("/users/{user_id}")
async def update_user(user_id: int = Path(..., gt=0),update_data: Dict = Body(..., example={"name": "更新名稱","age": 99})
):try:# 檢查用戶是否存在exist = query_database("SELECT id FROM users WHERE id = %s", (user_id,))if not exist:raise HTTPException(404, "用戶不存在")# 動態生成更新語句set_clause = ", ".join([f"{k}=%s" for k in update_data.keys()])update_query = f"UPDATE users SET {set_clause} WHERE id = %s"with pymysql.connect(**DB_CONFIG) as conn:with conn.cursor() as cursor:cursor.execute(update_query, (*update_data.values(),user_id))conn.commit()return {"status": "success", "affected_rows": cursor.rowcount}except pymysql.err.IntegrityError:raise HTTPException(409, "郵箱沖突或數據約束失敗")
# 刪除用戶 (DELETE)
@app.delete("/users/{user_id}")
async def delete_user(user_id: int = Path(..., gt=0)):try:with pymysql.connect(**DB_CONFIG) as conn:with conn.cursor() as cursor:cursor.execute("DELETE FROM users WHERE id = %s",(user_id,))conn.commit()if cursor.rowcount == 0:raise HTTPException(404, "用戶不存在")return {"status": "success", "deleted_id": user_id}except pymysql.err.Error as e:raise HTTPException(500, f"數據庫錯誤: {e}")
# 啟動應用
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

step3:運行

C:\Users\Administrator\PycharmProjects\FastAPIProject\.venv\Scripts\python.exe -m uvicorn main:app --reload 
INFO:     Will watch for changes in these directories: ['C:\\Users\\Administrator\\PycharmProjects\\FastAPIProject']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [12132] using StatReload
INFO:     Started server process [2000]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:53269 - "DELETE /users/16 HTTP/1.1" 200 OK

在postman里面，測試驗證成功

end