feat(api): implement FastAPI routes with JWT authentication
- jokes.py: Public joke listing, detail, and random endpoints - categories.py: Type and crowd listing endpoints - auth.py: Login endpoint with JWT token generation - admin.py: Full CRUD for jokes, types, crowds with JWT protection - Fix request body schemas to use Create DTOs instead of Response models Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
bwstudio
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
d59474f0a6
commit
c3a8e3e311
@@ -0,0 +1,6 @@
|
||||
from .jokes import router as jokes_router
|
||||
from .categories import router as categories_router
|
||||
from .auth import router as auth_router
|
||||
from .admin import router as admin_router
|
||||
|
||||
__all__ = ["jokes_router", "categories_router", "auth_router", "admin_router"]
|
||||
@@ -0,0 +1,272 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, Header
|
||||
from jose import JWTError, jwt
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import JWT_ALGORITHM, JWT_SECRET_KEY
|
||||
from app.database import get_db
|
||||
from app.models.joke import Joke
|
||||
from app.models.category import JokeCrowd, JokeType
|
||||
from app.models.user import AdminUser
|
||||
from app.schemas.category import JokeCrowdCreate, JokeCrowdResponse, JokeTypeCreate, JokeTypeResponse
|
||||
from app.schemas.joke import JokeCreate, JokeResponse, JokeUpdate, PaginatedJokeResponse
|
||||
|
||||
router = APIRouter(prefix="/admin", tags=["管理后台"])
|
||||
|
||||
|
||||
def get_current_admin_user(
|
||||
authorization: str = Header(None),
|
||||
db: Session = Depends(get_db),
|
||||
) -> AdminUser:
|
||||
"""Dependency to get current authenticated admin user."""
|
||||
if not authorization:
|
||||
raise HTTPException(status_code=401, detail="未提供认证令牌")
|
||||
parts = authorization.split()
|
||||
if len(parts) != 2 or parts[0].lower() != "bearer":
|
||||
raise HTTPException(status_code=401, detail="无效的认证格式")
|
||||
token = parts[1]
|
||||
try:
|
||||
payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
|
||||
user_id = payload.get("sub")
|
||||
if user_id is None:
|
||||
raise HTTPException(status_code=401, detail="无效的令牌")
|
||||
except JWTError:
|
||||
raise HTTPException(status_code=401, detail="无效或过期的令牌")
|
||||
|
||||
user = db.query(AdminUser).filter(AdminUser.id == int(user_id)).first()
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="用户不存在")
|
||||
return user
|
||||
|
||||
|
||||
def joke_to_response(joke: Joke) -> JokeResponse:
|
||||
"""Convert Joke model to JokeResponse schema."""
|
||||
return JokeResponse(
|
||||
id=joke.id,
|
||||
title=joke.title,
|
||||
content=joke.content,
|
||||
type_id=joke.type_id,
|
||||
crowd_id=joke.crowd_id,
|
||||
status=joke.status,
|
||||
view_count=joke.view_count,
|
||||
like_count=joke.like_count,
|
||||
created_at=joke.created_at,
|
||||
updated_at=joke.updated_at,
|
||||
type_name=joke.type.name if joke.type else None,
|
||||
crowd_name=joke.crowd.name if joke.crowd else None,
|
||||
)
|
||||
|
||||
|
||||
# Admin Jokes Routes
|
||||
@router.get("/jokes", response_model=PaginatedJokeResponse)
|
||||
def admin_list_jokes(
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
status: str = None,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""获取所有笑话（支持状态筛选）"""
|
||||
query = db.query(Joke)
|
||||
if status:
|
||||
query = query.filter(Joke.status == status)
|
||||
total = query.count()
|
||||
offset = (page - 1) * page_size
|
||||
jokes = query.order_by(Joke.created_at.desc()).offset(offset).limit(page_size).all()
|
||||
return PaginatedJokeResponse(
|
||||
items=[joke_to_response(j) for j in jokes],
|
||||
total=total,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/jokes", response_model=JokeResponse)
|
||||
def admin_create_joke(
|
||||
joke: JokeCreate,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""创建笑话"""
|
||||
db_joke = Joke(**joke.model_dump())
|
||||
db.add(db_joke)
|
||||
db.commit()
|
||||
db.refresh(db_joke)
|
||||
return joke_to_response(db_joke)
|
||||
|
||||
|
||||
@router.put("/jokes/{joke_id}", response_model=JokeResponse)
|
||||
def admin_update_joke(
|
||||
joke_id: int,
|
||||
joke: JokeUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""更新笑话"""
|
||||
db_joke = db.query(Joke).filter(Joke.id == joke_id).first()
|
||||
if not db_joke:
|
||||
raise HTTPException(status_code=404, detail="笑话不存在")
|
||||
update_data = joke.model_dump(exclude_unset=True)
|
||||
for key, value in update_data.items():
|
||||
setattr(db_joke, key, value)
|
||||
db.commit()
|
||||
db.refresh(db_joke)
|
||||
return joke_to_response(db_joke)
|
||||
|
||||
|
||||
@router.delete("/jokes/{joke_id}")
|
||||
def admin_delete_joke(
|
||||
joke_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""删除笑话"""
|
||||
db_joke = db.query(Joke).filter(Joke.id == joke_id).first()
|
||||
if not db_joke:
|
||||
raise HTTPException(status_code=404, detail="笑话不存在")
|
||||
db.delete(db_joke)
|
||||
db.commit()
|
||||
return {"message": "删除成功"}
|
||||
|
||||
|
||||
@router.put("/jokes/batch-approve")
|
||||
def admin_batch_approve(
|
||||
joke_ids: list[int],
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""批量审核通过笑话"""
|
||||
jokes = db.query(Joke).filter(Joke.id.in_(joke_ids)).all()
|
||||
for joke in jokes:
|
||||
joke.status = "approved"
|
||||
db.commit()
|
||||
return {"message": f"已审核通过 {len(jokes)} 条笑话"}
|
||||
|
||||
|
||||
@router.get("/stats")
|
||||
def admin_stats(
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""获取统计数据"""
|
||||
total_jokes = db.query(Joke).count()
|
||||
approved_jokes = db.query(Joke).filter(Joke.status == "approved").count()
|
||||
pending_jokes = db.query(Joke).filter(Joke.status == "pending").count()
|
||||
total_views = db.query(Joke).with_entities(func.sum(Joke.view_count)).scalar() or 0
|
||||
total_likes = db.query(Joke).with_entities(func.sum(Joke.like_count)).scalar() or 0
|
||||
return {
|
||||
"total_jokes": total_jokes,
|
||||
"approved_jokes": approved_jokes,
|
||||
"pending_jokes": pending_jokes,
|
||||
"total_views": total_views,
|
||||
"total_likes": total_likes,
|
||||
}
|
||||
|
||||
|
||||
# Categories CRUD
|
||||
@router.post("/types", response_model=JokeTypeResponse)
|
||||
def admin_create_type(
|
||||
joke_type: JokeTypeCreate,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""创建笑话类型"""
|
||||
db_type = JokeType(name=joke_type.name, icon=joke_type.icon, sort_order=joke_type.sort_order)
|
||||
db.add(db_type)
|
||||
db.commit()
|
||||
db.refresh(db_type)
|
||||
return JokeTypeResponse.model_validate(db_type)
|
||||
|
||||
|
||||
@router.put("/types/{type_id}", response_model=JokeTypeResponse)
|
||||
def admin_update_type(
|
||||
type_id: int,
|
||||
joke_type: JokeTypeResponse,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""更新笑话类型"""
|
||||
db_type = db.query(JokeType).filter(JokeType.id == type_id).first()
|
||||
if not db_type:
|
||||
raise HTTPException(status_code=404, detail="类型不存在")
|
||||
db_type.name = joke_type.name
|
||||
if joke_type.icon is not None:
|
||||
db_type.icon = joke_type.icon
|
||||
if joke_type.sort_order is not None:
|
||||
db_type.sort_order = joke_type.sort_order
|
||||
db.commit()
|
||||
db.refresh(db_type)
|
||||
return JokeTypeResponse.model_validate(db_type)
|
||||
|
||||
|
||||
@router.delete("/types/{type_id}")
|
||||
def admin_delete_type(
|
||||
type_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""删除笑话类型"""
|
||||
db_type = db.query(JokeType).filter(JokeType.id == type_id).first()
|
||||
if not db_type:
|
||||
raise HTTPException(status_code=404, detail="类型不存在")
|
||||
# Check if there are jokes using this type
|
||||
joke_count = db.query(Joke).filter(Joke.type_id == type_id).count()
|
||||
if joke_count > 0:
|
||||
raise HTTPException(status_code=400, detail=f"有 {joke_count} 条笑话使用此类型，无法删除")
|
||||
db.delete(db_type)
|
||||
db.commit()
|
||||
return {"message": "删除成功"}
|
||||
|
||||
|
||||
@router.post("/crowds", response_model=JokeCrowdResponse)
|
||||
def admin_create_crowd(
|
||||
crowd: JokeCrowdCreate,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""创建笑话人群分类"""
|
||||
db_crowd = JokeCrowd(name=crowd.name, icon=crowd.icon, sort_order=crowd.sort_order)
|
||||
db.add(db_crowd)
|
||||
db.commit()
|
||||
db.refresh(db_crowd)
|
||||
return JokeCrowdResponse.model_validate(db_crowd)
|
||||
|
||||
|
||||
@router.put("/crowds/{crowd_id}", response_model=JokeCrowdResponse)
|
||||
def admin_update_crowd(
|
||||
crowd_id: int,
|
||||
crowd: JokeCrowdResponse,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""更新笑话人群分类"""
|
||||
db_crowd = db.query(JokeCrowd).filter(JokeCrowd.id == crowd_id).first()
|
||||
if not db_crowd:
|
||||
raise HTTPException(status_code=404, detail="人群分类不存在")
|
||||
db_crowd.name = crowd.name
|
||||
if crowd.icon is not None:
|
||||
db_crowd.icon = crowd.icon
|
||||
if crowd.sort_order is not None:
|
||||
db_crowd.sort_order = crowd.sort_order
|
||||
db.commit()
|
||||
db.refresh(db_crowd)
|
||||
return JokeCrowdResponse.model_validate(db_crowd)
|
||||
|
||||
|
||||
@router.delete("/crowds/{crowd_id}")
|
||||
def admin_delete_crowd(
|
||||
crowd_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: AdminUser = Depends(get_current_admin_user),
|
||||
):
|
||||
"""删除笑话人群分类"""
|
||||
db_crowd = db.query(JokeCrowd).filter(JokeCrowd.id == crowd_id).first()
|
||||
if not db_crowd:
|
||||
raise HTTPException(status_code=404, detail="人群分类不存在")
|
||||
# Check if there are jokes using this crowd
|
||||
joke_count = db.query(Joke).filter(Joke.crowd_id == crowd_id).count()
|
||||
if joke_count > 0:
|
||||
raise HTTPException(status_code=400, detail=f"有 {joke_count} 条笑话使用此人群，无法删除")
|
||||
db.delete(db_crowd)
|
||||
db.commit()
|
||||
return {"message": "删除成功"}
|
||||
@@ -0,0 +1,39 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from jose import jwt
|
||||
from passlib.context import CryptContext
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import JWT_ALGORITHM, JWT_EXPIRATION_HOURS, JWT_SECRET_KEY
|
||||
from app.database import get_db
|
||||
from app.models.user import AdminUser
|
||||
from app.schemas.auth import LoginRequest, TokenResponse
|
||||
|
||||
router = APIRouter(prefix="/auth", tags=["认证"])
|
||||
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
|
||||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||
"""验证密码"""
|
||||
return pwd_context.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def create_access_token(data: dict) -> str:
|
||||
"""创建 JWT token"""
|
||||
to_encode = data.copy()
|
||||
expire = datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRATION_HOURS)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
@router.post("/login", response_model=TokenResponse)
|
||||
def login(req: LoginRequest, db: Session = Depends(get_db)):
|
||||
"""管理员登录"""
|
||||
user = db.query(AdminUser).filter(AdminUser.username == req.username).first()
|
||||
if not user or not verify_password(req.password, user.password_hash):
|
||||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||||
access_token = create_access_token(data={"sub": str(user.id), "username": user.username})
|
||||
return TokenResponse(access_token=access_token)
|
||||
@@ -0,0 +1,22 @@
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import get_db
|
||||
from app.models.category import JokeCrowd, JokeType
|
||||
from app.schemas.category import JokeCrowdResponse, JokeTypeResponse
|
||||
|
||||
router = APIRouter(prefix="/categories", tags=["分类"])
|
||||
|
||||
|
||||
@router.get("/types", response_model=list[JokeTypeResponse])
|
||||
def list_types(db: Session = Depends(get_db)):
|
||||
"""获取所有笑话类型"""
|
||||
types = db.query(JokeType).order_by(JokeType.sort_order, JokeType.id).all()
|
||||
return [JokeTypeResponse.model_validate(t) for t in types]
|
||||
|
||||
|
||||
@router.get("/crowds", response_model=list[JokeCrowdResponse])
|
||||
def list_crowds(db: Session = Depends(get_db)):
|
||||
"""获取所有笑话人群分类"""
|
||||
crowds = db.query(JokeCrowd).order_by(JokeCrowd.sort_order, JokeCrowd.id).all()
|
||||
return [JokeCrowdResponse.model_validate(c) for c in crowds]
|
||||
@@ -0,0 +1,78 @@
|
||||
import random
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import get_db
|
||||
from app.models.joke import Joke
|
||||
from app.schemas.joke import JokeResponse, PaginatedJokeResponse
|
||||
|
||||
router = APIRouter(prefix="/jokes", tags=["笑话"])
|
||||
|
||||
|
||||
def joke_to_response(joke: Joke) -> JokeResponse:
|
||||
"""Convert Joke model to JokeResponse schema."""
|
||||
return JokeResponse(
|
||||
id=joke.id,
|
||||
title=joke.title,
|
||||
content=joke.content,
|
||||
type_id=joke.type_id,
|
||||
crowd_id=joke.crowd_id,
|
||||
status=joke.status,
|
||||
view_count=joke.view_count,
|
||||
like_count=joke.like_count,
|
||||
created_at=joke.created_at,
|
||||
updated_at=joke.updated_at,
|
||||
type_name=joke.type.name if joke.type else None,
|
||||
crowd_name=joke.crowd.name if joke.crowd else None,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/", response_model=PaginatedJokeResponse)
|
||||
def list_jokes(
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
type_id: int | None = None,
|
||||
crowd_id: int | None = None,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""获取笑话列表（仅返回已审核通过的笑话）"""
|
||||
query = db.query(Joke).filter(Joke.status == "approved")
|
||||
|
||||
if type_id is not None:
|
||||
query = query.filter(Joke.type_id == type_id)
|
||||
if crowd_id is not None:
|
||||
query = query.filter(Joke.crowd_id == crowd_id)
|
||||
|
||||
total = query.count()
|
||||
offset = (page - 1) * page_size
|
||||
jokes = query.order_by(Joke.created_at.desc()).offset(offset).limit(page_size).all()
|
||||
|
||||
return PaginatedJokeResponse(
|
||||
items=[joke_to_response(j) for j in jokes],
|
||||
total=total,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{joke_id}", response_model=JokeResponse)
|
||||
def get_joke(joke_id: int, db: Session = Depends(get_db)):
|
||||
"""获取单条笑话详情"""
|
||||
joke = db.query(Joke).filter(Joke.id == joke_id).first()
|
||||
if not joke:
|
||||
raise HTTPException(status_code=404, detail="笑话不存在")
|
||||
# 增加浏览次数
|
||||
joke.view_count += 1
|
||||
db.commit()
|
||||
return joke_to_response(joke)
|
||||
|
||||
|
||||
@router.get("/random", response_model=JokeResponse)
|
||||
def get_random_joke(db: Session = Depends(get_db)):
|
||||
"""随机获取一条已审核通过的笑话"""
|
||||
joke = db.query(Joke).filter(Joke.status == "approved").order_by(func.random()).first()
|
||||
if not joke:
|
||||
raise HTTPException(status_code=404, detail="暂无笑话")
|
||||
return joke_to_response(joke)
|
||||
@@ -2,6 +2,7 @@ from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.config import API_TITLE, API_VERSION
|
||||
from app.routers import jokes_router, categories_router, auth_router, admin_router
|
||||
|
||||
# Create FastAPI application
|
||||
app = FastAPI(title=API_TITLE, version=API_VERSION)
|
||||
@@ -15,6 +16,12 @@ app.add_middleware(
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# Mount routers
|
||||
app.include_router(jokes_router)
|
||||
app.include_router(categories_router)
|
||||
app.include_router(auth_router)
|
||||
app.include_router(admin_router)
|
||||
|
||||
|
||||
@app.get("/")
|
||||
def root():
|
||||
|
||||
Reference in New Issue
Block a user