了解FastAPI程序結構

編寫一個簡單的FastAPI程序需要五個小步驟，先看一個完整例子

from fastapi import FastAPIapp = FastAPI()@app.get("/")
def root():return {"message": "Hello World"}

第一步，導入FastAPI

from fastapi import FastAPI

第二步，創建一個app實例

app = FastAPI()

第三步，編寫一個 路徑操作裝飾器

@app.get("/")

需要注意的兩點是：
● 你可以將get操作方法更改成@app.post()、@app.put()、@app.delete()等方法
● 你可以更改相應的路徑（“/”）為自己想要的，例如我更改為（“/hello_word/”）
第四步，編寫一個路徑操作函數，例如下面代碼中的root函數。它位于路徑操作裝飾器下方（見上方例子）
def root():
return {“message”: “Hello World”}
這個函數的返回值可以是
dict，list，單獨的值，比如str，int，或者是Pydantic模型

第五步、運行開發服務器uvicorn main:app --reload即可訪問api鏈接。

也可以安裝uvicorn，啟動

pip install uvicornimport uvicorn
from fastapi import FastAPI
app=FastAPI()
if __name__ == '__main__':uvicorn.run(app)

1. 例如我在終端運行uvicorn main:app --reload之后，在瀏覽器輸入127.0.0.1:8000，出現"message": "Hello World"這句話。
2. 在這里可以自己指定要運行的服務器ip和端口號。
3. 例如：uvicorn main:app --host 127.0.0.1 --port 8001 --reload表示指定本地電腦為服務器，端口號為8001。下面所有的代碼演示都默認這個本機ip地址和8001端口號。

符案例

from fastapi import FastAPI
import uvicorn
from service.knowledge_service import router as kno_router
from service.session_service import router as session_router
app = FastAPI()
# 聲明多應用
app.include_router(kno_router)
app.include_router(session_router)@app.get("/")
def root():return {"message": "Hello World"}if __name__ == '__main__':uvicorn.run(app="main:app", host="127.0.0.1", port=8000, reload=True)

聲明路徑參數

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}") 
def read_item(item_id):return {"item_id": item_id}

聲明路徑參數的類型

from fastapi import FastAPIapp = FastAPI()@app.get1("/items/{item_id}")
async def read_item1(item_id: int):return {"item_id": item_id}@app.get2("/items/{item_name}")
def read_item2(item_name: str):return {"item_id": item_name}

get請求查詢參數

from fastapi import FastAPIapp = FastAPI()@app.get("/files/")
def add(num1: int=2, num2: int=8):return {"num1 + num2 = ": num1 + num2}

請求體

請求體是客戶端發送到您的API的數據。 響應體是您的API發送給客戶端的數據。
API幾乎總是必須發送一個響應體，但是客戶端并不需要一直發送請求體。
定義請求體，需要使用 Pydantic 模型。注意以下幾點
不能通過GET請求發送請求體
發送請求體數據，必須使用以下幾種方法之一：POST（最常見）、PUT、DELETE、PATCH

如何實現請求體

第一步，從pydantic中導入BaseModel

from pydantic import BaseModel

第二步，創建請求體數據模型
聲明請求體數據模型為一個類，且該類繼承 BaseModel。所有的屬性都用標準Python類。和查詢參數一樣：數據類型的屬性如果不是必須的話，可以擁有一個默認值或者是可選None。否則，該屬性就是必須的。

from pydantic import BaseModelclass Item(BaseModel):name: str description: str = Noneprice: float tax: float = None 

所以訪問鏈接的時候傳入的請求體可以是下面兩種

第一種
{"name": "Foo","description": "An optional description","price": 45.2,"tax": 3.5
}
第二種
{"name": "Foo","price": 45.2
}

第三步、將模型定義為參數
將上面定義的模型添加到你的路徑操作中，就和定義Path和Query參數一樣的方式：

from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str = Noneprice: floattax: float = Noneapp = FastAPI()@app.post("/items/")
async def create_item(item: Item):return item

【注】
(1)

class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):results = {"item_id": item_id, "item": item}return results

async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
item: Annotated[Item, Body(embed=True)

item: Annotated[Item, Body(embed=True)]
● 這部分定義了請求體中的數據。它使用了 Annotated 和 Body 來為 item 參數指定詳細的信息。
● Annotated[Item, Body(embed=True)]：
○ Item 是一個 Pydantic 模型類，表示請求體的數據結構。Pydantic 會驗證請求體中的數據是否符合 Item 類的要求，自動進行類型檢查。
○ Item 類的字段包括：
■ name（字符串類型，必填）
■ description（可選的字符串）
■ price（浮動類型，必填）
■ tax（可選的浮動類型）
○ Body(embed=True) 是 FastAPI 提供的一個功能，表示請求體數據應該“嵌套”在一個對象內，而不是直接以屬性的形式存在。這意味著，客戶端請求的 JSON 數據結構應該像這樣：
{
“item”: {
“name”: “item_name”,
“description”: “item_description”,
“price”: 99.99,
“tax”: 5.0
}
}

(2)

from pydantic import BaseModel, Field
description: str | None = Field(default=None, title="The description of the item", max_length=300
)

= Field(…)
● 這里使用了 Pydantic 的 Field 函數來為 description 字段提供更多的元數據和驗證規則。
● Field 是 Pydantic 用來定義字段的函數，可以用來設置字段的默認值、驗證條件、描述信息等。

跨域配置

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()# 在這里換上前端的源請求
origins = ["http://localhost.tiangolo.com","https://localhost.tiangolo.com","http://localhost","http://localhost:8080",
]app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)@app.get("/")
async def main():return {"message": "Hello World"}

數據庫連接

1.配置

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager
# MySQL所在主機名
HOSTNAME = "127.0.0.1"
# MySQL監聽的端口號，默認3306
PORT = 3306
# 連接MySQL的用戶名，自己設置
USERNAME = "root"
# 連接MySQL的密碼，自己設置
PASSWORD = "root"
# MySQL上創建的數據庫名稱
DATABASE = "fadiantest"SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOSTNAME}:{PORT}/{DATABASE}?charset=utf8mb4"engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True
)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)@contextmanager
def get_db():db = SessionLocal()try:yield dbfinally:db.close()Base = declarative_base()

2.定義實體

from sqlalchemy import Column, String, Integer, Float ,DateTime
from config.mysql_config import Baseclass PowerPlant(Base):'''發電量'''__tablename__ = 'power_plant'id = Column(Integer, primary_key=True, autoincrement=True)power_plant_date = Column(String(128))month_power_generation = Column(Float)unit = Column(String(128))

3.應用

from config.mysql_config import get_db
from pojo.entities import PowerPlant@app.get("/hi")
async def say_hello():with get_db() as db:  # 確保會話被正確管理powerPlant = PowerPlant(power_plant_date="2024-11-11", month_power_generation=12.21, unit="機組")db.add(powerPlant)db.commit()return {"message": f"添加成功"}

CRUD

from fastapi import FastAPI,Query
import uvicorn
from sqlalchemy.orm import Session
from sqlalchemy import and_
app = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}@app.get("/hello/{name}")
async def say_hello(name: str):return {"message": f"Hello {name}"}from config.mysql_config import get_db
from models.entities import PowerPlant
from pojo.entities import PowerPlantPo
@app.get("/hi")
async def say_hello():with get_db() as db:  # 確保會話被正確管理powerPlant = PowerPlant(power_plant_date="2024-11-11", month_power_generation=12.21, unit="機組")db.add(powerPlant)db.commit()return {"message": f"添加成功"}@app.get("/getAll")
async def get_all_power_plants():with get_db() as db:powerPlants = db.query(PowerPlant).all()  # 查詢所有記錄return [{"id": plant.id,"power_plant_date": plant.power_plant_date,"month_power_generation": plant.month_power_generation,"unit": plant.unit} for plant in powerPlants]@app.put("/update_power_plant")
async def update_power_plant(powerPlant: PowerPlantPo):with get_db() as db:db_power_plant = db.query(PowerPlant).filter(PowerPlant.id == powerPlant.id).first()if db_power_plant:db_power_plant.power_plant_date = powerPlant.power_plant_datedb_power_plant.month_power_generation = powerPlant.month_power_generationdb_power_plant.unit = powerPlant.unitdb.commit()  # 提交事務db.refresh(db_power_plant)  # 確保數據已經被提交并刷新到數據庫print(f"Updated power plant: {db_power_plant}")  # 打印確認是否更新return {"message": "更新成功"}return {"message": "未找到該電廠信息"}@app.post("/addpow")
async def add_power_plant(powerPlant: PowerPlantPo):with get_db() as db:powerPlant = PowerPlant(power_plant_date=powerPlant.power_plant_date,month_power_generation=powerPlant.month_power_generation,unit=powerPlant.unit)db.add(powerPlant)  # 將實體添加到會話db.commit()  # 提交事務return {"message": "添加成功"}@app.delete("/deletepower/{plant_id}")
async def delete_power_plant(plant_id: int):with get_db() as db:powerPlant = db.query(PowerPlant).filter(PowerPlant.id == plant_id).first()if powerPlant:db.delete(powerPlant)  # 刪除數據db.commit()  # 提交事務return {"message": "刪除成功"}return {"message": "未找到該電廠信息"}@app.get("/powerplants")
async def get_power_plants(page: int = 1,  # 默認第一頁size: int = 10,  # 默認每頁10條name: str = Query(None),  # 過濾條件：電廠名稱location: str = Query(None)  # 過濾條件：電廠位置
):offset = (page - 1) * size  # 計算偏移量with get_db() as db:# 構造查詢過濾條件query_filters = []if name:query_filters.append(PowerPlant.unit.like(f"%{name}%"))if location:query_filters.append(PowerPlant.power_plant_date.like(f"%{location}%"))# 使用and_將多個過濾條件組合query = db.query(PowerPlant).filter(and_(*query_filters)) if query_filters else db.query(PowerPlant)# 獲取總數total_count = query.count()# 獲取分頁后的數據power_plants = query.offset(offset).limit(size).all()# 計算總頁數total_pages = (total_count + size - 1) // sizereturn {"total_count": total_count,"total_pages": total_pages,"page": page,"size": size,"power_plants": power_plants}if __name__ == '__main__':uvicorn.run(app)
``## 多個文件對應flask的藍圖
### 注冊多個文件```python
from fastapi import APIRouter
from config.mysql_config import get_db
from models.entities import PowerPlantrouter = APIRouter(prefix="/user",tags=["user"],responses={404: {"description": "Not found"}}
)@router.get('/getAll',tags=["users"])
def read_pow():with get_db() as db:powerPlants = db.query(PowerPlant).all()  # 查詢所有記錄return [{"id": plant.id,"power_plant_date": plant.power_plant_date,"month_power_generation": plant.month_power_generation,"unit": plant.unit} for plant in powerPlants]

在main.py中聲明

from service.user_service import router as user_router
app = FastAPI()app.include_router(user_router)

路由就已經注冊好了，可以正常訪問