82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Header, HTTPException, status
|
|
|
|
from app.core.constants import ERROR_CODE_OK
|
|
from app.core.time import format_now
|
|
from app.schemas.common import ApiResponse
|
|
from app.schemas.software_a import (
|
|
CreateDeployTaskData,
|
|
CreateDeployTaskRequest,
|
|
DeployTaskDetailData,
|
|
PermissionCheckData,
|
|
PermissionCheckRequest,
|
|
)
|
|
from app.services.software_a_service import SoftwareAService
|
|
|
|
router = APIRouter(prefix="/api/demo/software-a", tags=["demo-software-a"])
|
|
|
|
|
|
def build_request_id(header_value: str | None) -> str:
|
|
return header_value or f"req-{uuid4().hex[:12]}"
|
|
|
|
|
|
@router.post("/deploy-tasks", response_model=ApiResponse[CreateDeployTaskData])
|
|
def create_deploy_task(
|
|
payload: CreateDeployTaskRequest,
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[CreateDeployTaskData]:
|
|
request_id = build_request_id(x_request_id)
|
|
service = SoftwareAService("Asia/Shanghai")
|
|
task = service.create_deploy_task(payload)
|
|
return ApiResponse[CreateDeployTaskData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="deploy task created",
|
|
data=CreateDeployTaskData(
|
|
software_a_task_id=task["software_a_task_id"],
|
|
task_status=task["task_status"],
|
|
),
|
|
timestamp=format_now("Asia/Shanghai"),
|
|
)
|
|
|
|
|
|
@router.get("/deploy-tasks/{software_a_task_id}", response_model=ApiResponse[DeployTaskDetailData])
|
|
def get_deploy_task(
|
|
software_a_task_id: str,
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[DeployTaskDetailData]:
|
|
request_id = build_request_id(x_request_id)
|
|
task = SoftwareAService("Asia/Shanghai").get_deploy_task(software_a_task_id)
|
|
if not task:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail={"code": "NOT_FOUND", "message": "software_a task not found"})
|
|
return ApiResponse[DeployTaskDetailData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="success",
|
|
data=DeployTaskDetailData(**task),
|
|
timestamp=format_now("Asia/Shanghai"),
|
|
)
|
|
|
|
|
|
@router.post("/permissions/check", response_model=ApiResponse[PermissionCheckData])
|
|
def check_permission(
|
|
payload: PermissionCheckRequest,
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[PermissionCheckData]:
|
|
request_id = build_request_id(x_request_id)
|
|
allowed, reason = SoftwareAService("Asia/Shanghai").check_permission(payload.action_type, payload.env)
|
|
return ApiResponse[PermissionCheckData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="success",
|
|
data=PermissionCheckData(allowed=allowed, reason=reason),
|
|
timestamp=format_now("Asia/Shanghai"),
|
|
)
|