📚 简介

FastAPI是一个现代、快速的Web框架，用于构建API。

一个完整的FastAPI项目结构示例，包含路由、数据库、中间件等。

🚀 主入口文件

import uvicorn
from fastapi import FastAPI

from core.config import settings
from core.init_db import init_table, init_data
from register import register_mount, register_exception, register_router, register_cors
from common.custom_log import my_logger

app = FastAPI(
    description=settings.PROJECT_DESC, 
    version=settings.PROJECT_VERSION
)

register_cors(app)        # 注册跨域请求
register_mount(app)       # 挂载静态文件
register_router(app)      # 注册路由
register_exception(app)   # 注册异常捕获

init_table()              # 初始化表
init_data()               # 初始化表数据

my_logger.info("项目启动成功！！！")

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

📁 项目结构

backend/
├── apis/           # API路由
├── common/         # 公共模块
├── core/           # 核心配置
├── crud/           # 数据库操作
├── models/         # 数据模型
├── register/       # 注册模块
├── schemas/        # Pydantic模型
├── utils/          # 工具函数
└── main.py         # 入口文件

📦 主要模块

🚀 基本示例

from datetime import datetime, time, timedelta
from typing import Union, Dict
from fastapi import FastAPI, Query, Path, Body, Cookie, Header, Response, status, Form, File, UploadFile
from pydantic import BaseModel, Field
from uuid import UUID

app = FastAPI()


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


# 路径参数
@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}


# 查询参数
@app.get("/query_items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip: skip + limit]


# POST请求 - Pydantic模型
class Item(BaseModel):
    name: str
    description: Union[str, None] = Field(default=None, title="The description", max_length=300)
    price: float
    tax: Union[float, None] = None


@app.post("/create_items/")
async def create_item(item: Item):
    return item


# 字符串校验
@app.get("/limit_query/")
async def read_items(
    q: Union[str, None] = Query(default=None, min_length=3, max_length=50)
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results


# 路径参数校验
@app.get("/items_path/{item_id_path}")
async def read_items(
    item_id_path: int = Path(title="The ID of the item to get"),
    q: Union[str, None] = Query(default=None, alias="item-query"),
):
    results = {"item_id_path": item_id_path}
    if q:
        results.update({"q": q})
    return results


# PUT请求
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results


# Cookie示例
@app.get("/read_cookie/")
async def read_cookie_items(ads_id: Union[str, None] = Cookie(default=None)):
    return {"ads_id": ads_id}


# Header示例
@app.get("/read_header/")
async def read_header_items(user_agent: Union[str, None] = Header(default=None)):
    return {"User-Agent": user_agent}


# 表单对象
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username, "password": password}


# 文件上传
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    if not file:
        return {"message": "No upload file sent"}
    else:
        return {"filename": file.filename}

📦 主要特性

📚 参考资源

• 官方文档: https://fastapi.tiangolo.com/