payment-platform/backend/app/plugin/module_payment/apikey/controller.py

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Path
from fastapi.responses import JSONResponse

from app.api.v1.module_system.auth.schema import AuthSchema
from app.plugin.module_payment.apikey.schema import (
    TenantApiKeyCreate,
    TenantApiKeyResponse,
    TenantApiKeyListResponse,
    TenantApiKeyUpdate,
    ApiKeyQueryParam,
)
from app.plugin.module_payment.apikey.service import TenantApiKeyService
from app.common.response import ResponseSchema, SuccessResponse
from app.core.base_params import PaginationQueryParam
from app.core.dependencies import AuthPermission
from app.core.logger import log

ApiKeyRouter = APIRouter(prefix="/api-key", tags=["租户API Key管理"])


@ApiKeyRouter.post(
    "",
    summary="创建API Key",
    description="为租户创建新的API Key",
    response_model=ResponseSchema[TenantApiKeyResponse],
)
async def create_api_key_controller(
    data: TenantApiKeyCreate,
    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:tenant:api-key:create"]))],
) -> JSONResponse:
    tenant_id = data.tenant_id or auth.tenant_id
    api_key_obj = await TenantApiKeyService.create_api_key(
        auth=auth,
        tenant_id=tenant_id,
        expired_days=data.expired_days,
        description=data.description,
    )
    log.info(f"创建API Key成功: 租户ID={tenant_id}")
    return SuccessResponse(
        data=TenantApiKeyResponse(
            id=api_key_obj.id,
            api_key=api_key_obj.api_key,
            api_secret=api_key_obj.api_secret,
            status=api_key_obj.status,
            expired_at=api_key_obj.expired_at,
            created_time=api_key_obj.created_time,
        )
    )


@ApiKeyRouter.get(
    "/list",
    summary="查询API Key",
    description="查询API Key",
    response_model=ResponseSchema[list[TenantApiKeyListResponse]],
)
async def get_api_key_list_controller(
    page: Annotated[PaginationQueryParam, Depends()],
    search: Annotated[ApiKeyQueryParam, Depends()],
    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:tenant:api-key:query"]))],
) -> JSONResponse:
    target_tenant_id = search.tenant_id or auth.tenant_id
    result_dict = await TenantApiKeyService.get_api_key_page_service(
        auth=auth,
        page_no=page.page_no,
        page_size=page.page_size,
        tenant_id=target_tenant_id,
        status=search.status,
    )
    log.info("查询API Key成功")
    return SuccessResponse(data=result_dict)


@ApiKeyRouter.put(
    "/{api_key_id}",
    summary="更新API Key状态",
    description="更新API Key的状态",
    response_model=ResponseSchema[TenantApiKeyListResponse],
)
async def update_api_key_status_controller(
    api_key_id: Annotated[int, Path(description="API Key ID")],
    data: TenantApiKeyUpdate,
    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:tenant:api-key:update"]))],
) -> JSONResponse:
    api_key_obj = await TenantApiKeyService.update_api_key_status(auth=auth, api_key_id=api_key_id, status=data.status)
    if not api_key_obj:
        raise HTTPException(status_code=404, detail="API Key不存在")
    log.info(f"更新API Key状态成功: ID={api_key_id}, 状态={data.status}")
    return SuccessResponse(
        data=TenantApiKeyListResponse(
            id=api_key_obj.id,
            api_key=api_key_obj.api_key,
            status=api_key_obj.status,
            expired_at=api_key_obj.expired_at,
            last_used_at=api_key_obj.last_used_at,
            created_time=api_key_obj.created_time,
            description=api_key_obj.description,
        )
    )


@ApiKeyRouter.delete(
    "/{api_key_id}",
    summary="删除API Key",
    description="删除API Key",
    response_model=ResponseSchema[dict],
)
async def delete_api_key_controller(
    api_key_id: Annotated[int, Path(description="API Key ID")],
    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:tenant:api-key:delete"]))],
) -> JSONResponse:
    success = await TenantApiKeyService.delete_api_key(auth=auth, api_key_id=api_key_id)
    if not success:
        raise HTTPException(status_code=404, detail="API Key不存在")
    log.info(f"删除API Key成功: ID={api_key_id}")
    return SuccessResponse(data={"success": True})