96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.dependencies import get_admin_user, get_db
|
|
from app.models.user import User
|
|
from app.schemas.user_oauth_app import UserOAuthAppCreate, UserOAuthAppOut, UserOAuthAppUpdate
|
|
from app.services.user_oauth_app_service import (
|
|
create_user_oauth_app,
|
|
delete_user_oauth_app,
|
|
get_user_oauth_app_by_id,
|
|
list_user_oauth_apps,
|
|
update_user_oauth_app,
|
|
)
|
|
|
|
router = APIRouter(prefix="/admin/user-oauth-apps", tags=["oauth"])
|
|
|
|
|
|
@router.get("/list", summary="获取用户授权应用列表")
|
|
async def list_apps(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
open_type: int | None = Query(None, ge=1, le=10, description="开户方式"),
|
|
status: int | None = Query(None, ge=1, le=2, description="应用状态，1=正常,2=禁用"),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
app_id: str | None = Query(None, max_length=255, description="应用id"),
|
|
):
|
|
result = await list_user_oauth_apps(db, page, page_size, open_type, status, admin.id, app_id)
|
|
return {
|
|
"total": result["total"],
|
|
"page": result["page"],
|
|
"page_size": result["page_size"],
|
|
"items": [UserOAuthAppOut.model_validate(item) for item in result["items"]],
|
|
}
|
|
|
|
|
|
@router.post("/create", summary="创建用户授权应用")
|
|
async def create_app(
|
|
req: UserOAuthAppCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
try:
|
|
app = await create_user_oauth_app(db, req.app_id, req.secret, req.open_type, admin.id)
|
|
return UserOAuthAppOut.model_validate(app)
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.get("/read/{id}", summary="获取用户授权应用详情")
|
|
async def get_app(
|
|
id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
app = await get_user_oauth_app_by_id(db, id)
|
|
if not app:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="应用不存在",
|
|
)
|
|
return UserOAuthAppOut.model_validate(app)
|
|
|
|
|
|
@router.post("/update/{id}", summary="更新用户授权应用")
|
|
async def update_app(
|
|
id: str,
|
|
req: UserOAuthAppUpdate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
app = await update_user_oauth_app(db, id, req.secret, req.open_type, req.status, admin.id)
|
|
if not app:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="应用不存在",
|
|
)
|
|
return UserOAuthAppOut.model_validate(app)
|
|
|
|
|
|
@router.get("/delete/{id}", summary="删除用户授权应用")
|
|
async def delete_app(
|
|
id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
success = await delete_user_oauth_app(db, id, admin.id)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="应用不存在",
|
|
)
|
|
return {"message": "删除成功"} |