將 FastAPI 結合進來通過 API 來執行 PowerShell 命令，為了之後在寫前端時 response 都能有個統一的格式，先定義回傳的格式如下:

{
code: 20000,
message: "success",
    data: {
    msg: "Hello Wolrd!
    "
    }
}

封裝統一的 response 的函式

def reponse(*, code=20000,data: Union[list, dict, str],message="success") -> Response:
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content={
            'code': code,
            'message': message,
            'data': data,
        }
    )

將 subprocess run 函數參數調整為 stdout=PIPE 和 stderr=STDOUT 可以將 stdout 和 stderr 合併，執行預先定義好的命令搭配 response 的函式輸出結果

@app.get("/test_command")
def test_command():
    command = "powershell -Command Write-Host 'Hello Wolrd!'"
    ret = subprocess.run(command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,encoding="utf-8")
    if ret.returncode < 0:
        return reponse(code=10000,data={'msg':"命令中斷"},message="error")
    else: 
        return reponse(data={'msg':ret.stdout},message="success")

通過瀏覽器訪問 API，可以看到 JSON 格式的回覆

接者將命令調整可以參數傳入，定義請求體

class Item(BaseModel):
    command: str

新建一個 API 並使用 HTTP 方法中的 POST 操作

@app.post("/command/")
async def execute_command(item: Item):
    ret = subprocess.run(item.command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,encoding="utf-8")
    if ret.returncode < 0:
        return reponse(code=10000,data={'msg':"命令中斷"},message="error")
    else: 
        return reponse(data={'msg':ret.stdout},message="success")

測試使用 Postman

參數傳入使用 Body 中的 raw 並選取 JSON 格式

參考資源

FastAPI 學習之路（五十九）封裝統一的json返回處理工具

請求體

subprocess --- 子进程管理