FastAPI 基础教程

下面给你一份真正实用、可直接上手项目的 FastAPI 教程。

不是那种只讲概念的，而是结合你现在的技术栈（FastAPI + SQLAlchemy ORM + PostgreSQL + JWT + Casbin + UMI + AntD）去写的项目级教程。

内容分为 基础 -> 实战 -> 工程化最佳实践 三个部分。

🚀 一、FastAPI 基础入门（10 分钟掌握）

1.1 安装

pip install fastapi uvicorn[standard]

1.2 最简单的示例

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def index():
    return {"msg": "Hello FastAPI"}

运行：

uvicorn main:app --reload

访问：http://127.0.0.1:8000/docs

Swagger 文档自动生成。

📌 二、路由（Router）与项目结构

推荐模块化结构（你已经在使用）：

app/
    api/
        users.py
        auth.py
        roles.py
        permissions.py
    core/
        config.py
        security.py
    db/
        base.py
        session.py
        models/
            user.py
            role.py
            permission.py
    schemas/
        user.py
        role.py
    main.py

在 main.py 注册路由：

from fastapi import FastAPI
from api import users, auth

app = FastAPI()

app.include_router(auth.router, prefix="/api/auth")
app.include_router(users.router, prefix="/api/users")

🧱 三、Pydantic 模型（Schemas）

请求模型 + 响应模型：

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

class UserOut(BaseModel):
    id: int
    username: str
    email: str

    class Config:
        from_attributes = True

FastAPI 会自动验证输入。

🗄 四、数据库：SQLAlchemy ORM（Async 版本）

你项目中使用 Async SQLAlchemy 2.0 是最佳方案。

4.1 数据库连接

# db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost:5432/dbname",
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    engine, expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

4.2 ORM 模型

# db/models/user.py
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Integer
from db.session import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    password: Mapped[str] = mapped_column(String(255))

🔐 五、JWT 登录鉴权（你项目正在使用）

5.1 安装依赖

pip install python-jose passlib[bcrypt] python-multipart

5.2 创建 JWT 工具（你已要求集成 bcrypt）

# core/security.py
from datetime import datetime, timedelta
from jose import jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret"
ALGORITHM = "HS256"
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str):
    return pwd_context.hash(password)

def verify_password(password: str, hashed: str):
    return pwd_context.verify(password, hashed)

def create_access_token(data: dict, expires_minutes=60):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=expires_minutes)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

5.3 登录接口

# api/auth.py
from fastapi import APIRouter, Depends, HTTPException
from core.security import verify_password, create_access_token
from db.session import get_db
from sqlalchemy import select
from db.models.user import User

router = APIRouter()

@router.post("/login")
async def login(data: dict, db=Depends(get_db)):
    stmt = select(User).where(User.username == data["username"])
    result = await db.execute(stmt)
    user = result.scalar_one_or_none()

    if not user or not verify_password(data["password"], user.password):
        raise HTTPException(401, "用户名或密码错误")

    token = create_access_token({"id": user.id})

    return {"token": token}

⚖ 六、权限控制：Casbin（你项目正在使用）

安装：

pip install casbin casbin_sqlalchemy_adapter

集成示例（核心代码）

from fastapi import Depends, HTTPException
from casbin import Enforcer
from core.casbin import enforcer

def check_permission(sub: str, obj: str, act: str):
    if not enforcer.enforce(sub, obj, act):
        raise HTTPException(status_code=403, detail="权限不足")

可在路由中使用：

@router.get("/roles")
async def get_roles(user=Depends(get_current_user)):
    check_permission(user.role_code, "roles", "read")

🧩 七、返回 JSON：通用 ORM -> JSON 转换工具（你要求过）

示例：

from datetime import date, datetime

def to_dict_list(result, fields: list[str] | None = None):
    data = []
    for item in result:
        row = {}
        for key in fields or item.__dict__.keys():
            val = getattr(item, key, None)
            if isinstance(val, (datetime, date)):
                val = val.strftime("%Y-%m-%d %H:%M:%S") if isinstance(val, datetime) else val.strftime("%Y-%m-%d")
            elif isinstance(val, str):
                val = val.strip()
            row[key] = val
        data.append(row)
    return data

📦 八、上传、下载文件

上传：

from fastapi import UploadFile

@router.post("/upload")
async def upload(file: UploadFile):
    content = await file.read()
    with open(f"files/{file.filename}", "wb") as f:
        f.write(content)
    return {"msg": "ok"}

下载：

from fastapi.responses import FileResponse

@router.get("/download")
async def download():
    return FileResponse("files/test.pdf")

📡 九、FastAPI 中间件示例（日志、跨域）

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

⚙ 十、工程化最佳实践（强烈推荐）

你构建后端系统时，需要以下规范：

✅ 配置文件 Config（区分 dev/prod）

使用 pydantic-settings。

✅ 异步 SQLAlchemy 工具函数

你正在封装 async insert/update/delete，可采用：

async def db_insert(db: AsyncSession, obj):
    db.add(obj)
    await db.commit()
    await db.refresh(obj)
    return obj

✅ 错误统一处理

统一输出：

{"code": 400, "msg": "xxx", "data": null}

✅ 日志

loguru + 旋转日志。