# 核心框架與文件上傳
pip install fastapi uvicorn python-multipart
# 異步任務(wù)與消息隊列
pip install celery redis
# Excel 解析
pip install pandas openpyxl  # openpyxl 用于讀取 .xlsx 格式
# 環(huán)境變量管理
pip install python-dotenv
# 數(shù)據(jù)庫（復(fù)用之前的 MySQL 配置，存儲導(dǎo)入的數(shù)據(jù)）
pip install sqlalchemy aiomysql

fastapi-async-file/
├── .env                # 環(huán)境變量（數(shù)據(jù)庫、Redis 配置）
├── main.py             # 主程序（FastAPI 實例、文件上傳/任務(wù)查詢接口）
├── celery_config.py    # Celery 配置（連接 Redis、任務(wù)設(shè)置）
├── tasks.py            # 異步任務(wù)（解析 Excel、批量導(dǎo)入數(shù)據(jù)庫）
├── database.py         # 數(shù)據(jù)庫配置（復(fù)用之前的 MySQL 異步配置）
├── models/             # 數(shù)據(jù)模型
│   ├── db.py           # SQLAlchemy 數(shù)據(jù)庫模型（用戶表）
│   └── schemas.py      # Pydantic 校驗?zāi)Ｐ停蛇x，用于數(shù)據(jù)校驗）
└── utils.py            # 工具函數(shù)（Excel 解析、數(shù)據(jù)校驗）

# .env
# MySQL 數(shù)據(jù)庫配置（批量導(dǎo)入的數(shù)據(jù)存這里）
DATABASE_URL=mysql+aiomysql://root:123456@localhost:3306/fastapi_demo
# Redis 配置（Celery 用）
REDIS_URL=redis://localhost:6379/0  # 0 是 Redis 的數(shù)據(jù)庫編號
# 任務(wù)相關(guān)配置
TASK_RESULT_EXPIRES=3600  # 任務(wù)結(jié)果保留 1 小時（3600 秒）

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import declarative_base
from dotenv import load_dotenv
import os

load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")

# 異步引擎
async_engine = create_async_engine(
    DATABASE_URL,
    echo=True,  # 開發(fā)時打印 SQL，便于調(diào)試
    future=True
)

# 異步會話工廠
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False
)

# 數(shù)據(jù)庫模型基類
Base = declarative_base()

# 依賴項：獲取數(shù)據(jù)庫會話
async def get_db():
    db = AsyncSessionLocal()
    try:
        yield db
    finally:
        await db.close()

# models/db.py
from sqlalchemy import Column, Integer, String, Boolean, func
from sqlalchemy.sql.sqltypes import DateTime
from database import Base

class DBUser(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    full_name = Column(String, index=True, nullable=True)
    disabled = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

# models/schemas.py
from pydantic import BaseModel, Field, validator
from typing import Optional

class UserImport(BaseModel):
    """Excel 導(dǎo)入數(shù)據(jù)的校驗?zāi)Ｐ?""
    username: str = Field(..., min_length=3, max_length=20, description="用戶名3-20字符")
    email: str = Field(..., pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", description="郵箱格式正確")
    full_name: Optional[str] = Field(None, description="真實姓名可選")
    
    # 可選：自定義校驗，比如用戶名不能包含特殊字符
    @validator("username")
    def username_no_special_chars(cls, v):
        if not v.isalnum():  # 只允許字母和數(shù)字
            raise ValueError("用戶名只能包含字母和數(shù)字")
        return v

from dotenv import load_dotenv
import os

load_dotenv()

# Celery 配置
CELERY_CONFIG = {
    # 連接 Redis（消息隊列）
    "broker_url": os.getenv("REDIS_URL"),
    # 存儲任務(wù)結(jié)果（便于查詢?nèi)蝿?wù)狀態(tài)）
    "result_backend": os.getenv("REDIS_URL"),
    # 任務(wù)結(jié)果保留時間（秒）
    "result_expires": int(os.getenv("TASK_RESULT_EXPIRES", 3600)),
    # 任務(wù)序列化方式
    "task_serializer": "json",
    # 結(jié)果序列化方式
    "result_serializer": "json",
    # 接受的內(nèi)容類型
    "accept_content": ["json"],
}

from celery import Celery
from celery_config import CELERY_CONFIG
import pandas as pd
from sqlalchemy.ext.asyncio import AsyncSession
from database import AsyncSessionLocal
from models.db import DBUser
from models.schemas import UserImport
from typing import List, Dict
import traceback

# 初始化 Celery 實例
celery_app = Celery("fastapi_tasks", config_source=CELERY_CONFIG)

# 定義異步任務(wù)：解析 Excel 并批量導(dǎo)入用戶
@celery_app.task(bind=True, name="import_users_from_excel")
def import_users_from_excel(self, file_path: str) -> Dict[str, any]:
    """
    異步任務(wù)：從 Excel 文件批量導(dǎo)入用戶
    :param self: 任務(wù)實例（用于記錄任務(wù)狀態(tài)）
    :param file_path: 上傳的 Excel 文件路徑
    :return: 任務(wù)結(jié)果（成功數(shù)量、失敗數(shù)量、失敗原因）
    """
    result = {
        "success_count": 0,
        "fail_count": 0,
        "failures": [],  # 存儲失敗的數(shù)據(jù)和原因
        "status": "completed"
    }
    
    try:
        # 1. 解析 Excel 文件（pandas 讀取 .xlsx）
        df = pd.read_excel(file_path, engine="openpyxl")
        # 確保列名正確（Excel 第一行必須是 username、email、full_name）
        required_columns = ["username", "email", "full_name"]
        if not all(col in df.columns for col in required_columns):
            raise ValueError(f"Excel 缺少必需列！需要：{required_columns}")
        
        # 2. 數(shù)據(jù)校驗（用 Pydantic 過濾臟數(shù)據(jù)）
        valid_users: List[Dict] = []
        for idx, row in df.iterrows():
            row_data = row.to_dict()
            try:
                # 校驗數(shù)據(jù)格式
                user = UserImport(**row_data)
                valid_users.append(user.dict())
            except Exception as e:
                # 記錄失敗數(shù)據(jù)和原因
                result["fail_count"] += 1
                result["failures"].append({
                    "row": idx + 2,  # Excel 行號從 2 開始（第一行是表頭）
                    "data": row_data,
                    "reason": str(e)
                })
        
        # 3. 批量導(dǎo)入數(shù)據(jù)庫（異步會話，需要用 asyncio.run 執(zhí)行）
        if valid_users:
            import asyncio
            success_num = asyncio.run(_batch_insert_users(valid_users))
            result["success_count"] = success_num
            result["fail_count"] += len(valid_users) - success_num  # 可能有重復(fù)用戶名/郵箱導(dǎo)致插入失敗
    
    except Exception as e:
        # 任務(wù)執(zhí)行失敗（如文件損壞、數(shù)據(jù)庫連接錯誤）
        result["status"] = "failed"
        result["error"] = str(e)
        result["traceback"] = traceback.format_exc()
    
    return result

async def _batch_insert_users(users: List[Dict]) -> int:
    """批量插入用戶到數(shù)據(jù)庫（異步函數(shù)）"""
    async with AsyncSessionLocal() as db:
        success_count = 0
        for user_data in users:
            try:
                # 檢查用戶名/郵箱是否已存在
                from sqlalchemy import select
                result = await db.execute(
                    select(DBUser).where(
                        (DBUser.username == user_data["username"]) | 
                        (DBUser.email == user_data["email"])
                    )
                )
                if not result.scalar_one_or_none():
                    # 不存在則插入
                    db_user = DBUser(**user_data)
                    db.add(db_user)
                    success_count += 1
            except Exception as e:
                print(f"插入用戶失敗：{user_data}，原因：{e}")
        await db.commit()
        return success_count

from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
from fastapi.responses import JSONResponse
from celery.result import AsyncResult
from tasks import celery_app
from database import get_db, Base, async_engine
from sqlalchemy.ext.asyncio import AsyncSession
import os
from contextlib import asynccontextmanager
from typing import AsyncGenerator

# 定義生命周期：啟動時創(chuàng)建數(shù)據(jù)庫表
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

# 初始化 FastAPI 實例
app = FastAPI(
    title="FastAPI 異步任務(wù) + 文件上傳實戰(zhàn)",
    description="Excel 批量導(dǎo)入用戶數(shù)據(jù)全流程",
    lifespan=lifespan
)

# 配置文件上傳路徑（確保目錄存在）
UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)  # 不存在則創(chuàng)建

# --- 1. 文件上傳接口：上傳 Excel 并觸發(fā)異步任務(wù) ---
@app.post("/upload/excel/", summary="上傳 Excel 批量導(dǎo)入用戶")
async def upload_excel(
    file: UploadFile = File(..., description="上傳 .xlsx 格式的 Excel 文件"),
    db: AsyncSession = Depends(get_db)
):
    # 第一步：校驗文件格式和大小
    # 校驗格式：只允許 .xlsx
    if not file.filename.endswith(".xlsx"):
        raise HTTPException(
            status_code=400,
            detail="只支持 .xlsx 格式的 Excel 文件！"
        )
    # 校驗大小：限制 10MB 以內(nèi)（可根據(jù)需求調(diào)整）
    file_size = 0
    contents = await file.read()
    file_size = len(contents)
    MAX_SIZE = 10 * 1024 * 1024  # 10MB
    if file_size > MAX_SIZE:
        raise HTTPException(
            status_code=400,
            detail=f"文件大小超過限制！最大支持 10MB，當(dāng)前文件 {file_size/1024/1024:.2f}MB"
        )
    
    # 第二步：保存文件到本地（uploads 目錄）
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    with open(file_path, "wb") as f:
        f.write(contents)
    
    # 第三步：觸發(fā)異步任務(wù)（celery_app.send_task 提交任務(wù)）
    task = celery_app.send_task(
        "import_users_from_excel",  # 任務(wù)名稱（和 tasks.py 中定義的一致）
        args=[file_path]  # 傳遞給任務(wù)的參數(shù)（文件路徑）
    )
    
    # 第四步：返回任務(wù) ID，讓用戶后續(xù)查詢狀態(tài)
    return JSONResponse(
        status_code=200,
        content={
            "message": "文件上傳成功，異步導(dǎo)入任務(wù)已啟動！",
            "task_id": task.id,  # 任務(wù)唯一 ID
            "file_name": file.filename,
            "file_size": f"{file_size/1024:.2f}KB"
        }
    )

# --- 2. 任務(wù)狀態(tài)查詢接口 ---
@app.get("/task/{task_id}/", summary="查詢異步任務(wù)狀態(tài)和結(jié)果")
async def get_task_status(task_id: str):
    # 獲取任務(wù)結(jié)果對象
    task_result = AsyncResult(task_id, app=celery_app)
    
    # 構(gòu)造任務(wù)狀態(tài)信息
    result = {
        "task_id": task_id,
        "status": task_result.status,  # PENDING（待執(zhí)行）/ RUNNING（執(zhí)行中）/ SUCCESS（成功）/ FAILURE（失敗）
        "result": None,
        "error": None
    }
    
    # 根據(jù)任務(wù)狀態(tài)返回不同信息
    if task_result.status == "SUCCESS":
        result["result"] = task_result.result  # 任務(wù)執(zhí)行成功的結(jié)果（成功數(shù)量、失敗數(shù)量等）
    elif task_result.status == "FAILURE":
        result["error"] = str(task_result.result)  # 任務(wù)失敗的原因
    
    return result

# 運行程序
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

celery -A tasks.celery_app worker --loglevel=info