FastAPI 学习指南 - 为什么它代表了未来？

pip install fastapi uvicorn

# main.py
from fastapi import FastAPI

# 初始化 FastAPI 应用实例
app = FastAPI(
    title="我的第一个 FastAPI 应用",  # 文档标题
    description="FastAPI 入门示例",    # 文档描述
    version="1.0.0"                   # 版本号
)

# 定义 GET 请求路由，路径为根路径 "/"
@app.get("/")
def read_root():
    """根路径接口，返回简单的问候信息"""
    return {"Hello": "World"}

# 带路径参数的 GET 接口
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    """根据 item_id 查询物品，q 为可选查询参数"""
    return {"item_id": item_id, "q": q}

# 启动命令（终端执行）：uvicorn main:app --reload

# main.py
import asyncio
from fastapi import FastAPI

app = FastAPI()

# 同步接口（对比用）
@app.get("/sync")
def sync_operation():
    # 模拟同步IO等待（会阻塞整个线程）
    import time
    time.sleep(1)  # 等待1秒
    return {"status": "completed", "type": "sync"}

# 异步接口
@app.get("/async")
async def async_operation():
    # 模拟异步IO等待（不会阻塞线程）
    await asyncio.sleep(1)  # 异步等待1秒
    return {"status": "completed", "type": "async"}

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field

app = FastAPI()

# 定义数据模型（Pydantic BaseModel）
class User(BaseModel):
    username: str = Field(min_length=3, max_length=50, description="用户名，3-50个字符")
    email: EmailStr  # 自动验证邮箱格式
    age: int = Field(ge=18, le=120, description="年龄，18-120岁")
    is_active: bool = True  # 默认值

# 接收并验证 POST 请求体
@app.post("/users/")
async def create_user(user: User):
    # user 已自动验证并转换为对象，可直接访问属性
    return {
        "message": f"用户 {user.username} 创建成功",
        "user_data": user.dict()  # 序列化为字典返回
    }

from fastapi import FastAPI, Depends
import sqlite3

app = FastAPI()

# 定义依赖函数：获取数据库连接
def get_db():
    conn = sqlite3.connect("mydb.db")
    try:
        yield conn  # 提供连接给接口使用
    finally:
        conn.close()  # 接口执行完后关闭连接

# 接口使用依赖
@app.get("/db-items/")
async def get_db_items(db: sqlite3.Connection = Depends(get_db)):
    cursor = db.cursor()
    cursor.execute("SELECT * FROM items")
    items = cursor.fetchall()
    return {"items": items}

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

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

# 配置
SECRET_KEY = "your-secret-key-keep-it-safe"  # 生产环境需用环境变量
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 模拟用户数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "hashed_password": pwd_context.hash("secret"),
        "email": "johndoe@example.com",
    }
}

# 生成 Token
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 验证 Token 获取当前用户
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return user

# 获取 Token 接口
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or not pwd_context.verify(form_data.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )
    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护的接口
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

# todo_app.py
from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# 1. 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./todos.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 2. 数据库模型
class TodoDB(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, nullable=True)
    completed = Column(Boolean, default=False)

# 创建数据库表
Base.metadata.create_all(bind=engine)

# 3. Pydantic 模型（数据验证/序列化）
class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)

class TodoUpdate(BaseModel):
    title: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)
    completed: bool | None = None

class Todo(TodoCreate):
    id: int
    completed: bool

    class Config:
        orm_mode = True  # 支持从 ORM 对象转换为 Pydantic 模型

# 4. FastAPI 应用初始化
app = FastAPI(title="Todo API", version="1.0")

# 5. 依赖：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 6. CRUD 接口
# 创建 Todo
@app.post("/todos/", response_model=Todo, status_code=status.HTTP_201_CREATED)
def create_todo(todo: TodoCreate, db: Session = Depends(get_db)):
    db_todo = TodoDB(**todo.dict())
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo

# 获取所有 Todo
@app.get("/todos/", response_model=list[Todo])
def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    todos = db.query(TodoDB).offset(skip).limit(limit).all()
    return todos

# 获取单个 Todo
@app.get("/todos/{todo_id}", response_model=Todo)
def read_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo 不存在")
    return todo

# 更新 Todo
@app.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, todo: TodoUpdate, db: Session = Depends(get_db)):
    db_todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo 不存在")
    # 只更新传入的字段
    update_data = todo.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_todo, key, value)
    db.commit()
    db.refresh(db_todo)
    return db_todo

# 删除 Todo
@app.delete("/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    db_todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo 不存在")
    db.delete(db_todo)
    db.commit()
    return None

gunicorn todo_app:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000