fastapi

fastapi封装post、get请求

需要用到的相关python库

from pydantic import BaseModel
from typing import Generic, TypeVar, Type, Any
import httpx

1. 封装响应体

T = TypeVar("T")     
class ResponseTemplate(BaseModel, Generic[T]):
    flag_ret: bool
    data: T | list[T] | None
    msg: str | None

class Response:
    @staticmethod
    def success(*, data: T = None, msg: str = "操作成功") -> ResponseTemplate[T]:
        return ResponseTemplate(flag_ret=True, data=data, msg=msg)

    @staticmethod
    def fail(*, data: Any = None, msg: str = "操作失败") -> ResponseTemplate[T]:
        return ResponseTemplate(flag_ret=False, data=data, msg=msg)

2.封装post请求

ReqTemplate = TypeVar("ReqTemplate", bound=BaseModel)
RespTemplate = TypeVar("RespTemplate", bound=ResponseTemplate)
async def sendReqPost(ClsReq: Type[ReqTemplate], ClsResp: Type[RespTemplate], url, **kwargs: Any)->tuple[Type[RespTemplate]|None, str]:
    req = None
    try:
        req = ClsReq(**kwargs)
    except Exception as e:
        return None, f'make request body failed, {e}'

    try:
        async with httpx.AsyncClient(timeout=3*60) as client:
            dict_data = req.model_dump()
            response = await client.post(url, json=dict_data)
            try:
                ret = ClsResp.model_validate(response.json())
                if ret.flag_ret:
                    return ret, ''
                return None, f'{ret.msg}'
            except Exception as e:
                return None, f'{extractValidateErrFromDefaultJson(response.json())}'
    except Exception as e:
        return None, f'{e}'

3.封装get请求

RespTemplate = TypeVar("RespTemplate", bound=ResponseTemplate)
async def sendReqGet(ClsResp: Type[RespTemplate], url, **kwargs: Any)->tuple[Type[RespTemplate]|None, str]:
    try:
        async with httpx.AsyncClient(timeout=3*60) as client:
            response = await client.get(url, params=kwargs)
            try:
                ret = ClsResp.model_validate(response.json())
                if ret.flag_ret:
                    return ret, ''
                return None, f'{ret.msg}'
            except Exception as e:
                return None, f'{extractValidateErrFromDefaultJson(response.json())}'
    except Exception as e:
        return None, f'{e}'

4. 调用

get请求

• fastapi server

@app.get("/get")
def getIp(num:int, q:str):
    try:
        return Response.success(data={num,q})
    except Exception as e:
        return Response.fail(msg=f'{e}')

• client

ret, msg = asyncio.run(sendReqGet(ClsResp=ResponseTemplate, url='http://localhost:8000/get', num=1, q='123')) 

post请求

• post请求与get请求不同，需要用pydantic检验参数合法性，因此需要提前定义好结构体

from pydantic import BaseModel

class RequestBuildParam(BaseModel):
    url_ssh_repository:str
    checkout_target:str
    command:str
    list_to_pack:list[str]
    ip_dst:str
    path_folder_dst:str

• fastapi定义post接口

@app.post("/build")
def build(build_param:RequestBuildParam):
    try:
        clelery_app = Celery('auto_builder')
        clelery_app.config_from_object(celery_config)
        param_scheduler = ParamScheduler(NUM_WORKER, clelery_app, PATH_FOLDER_REPOS, build_param.url_ssh_repository)
        path_file_pack = genZipPath(PATH_FOLDER_PACK)
        param_task = ParamTask(path_file_pack, build_param.checkout_target, build_param.command, build_param.list_to_pack, build_param.ip_dst, build_param.path_folder_dst)
        work_scheduler = WorkScheduler()
        ret, msg = work_scheduler.init(param_scheduler, callbackInfo, callbackErr)
        id = work_scheduler.addTask(param_task)
        return Response.success(data=id)
    except Exception as e:
       return Response.fail(msg=f'{e}')

• 给fastapi发请求

        ret, msg = await sendReqPost(ClsReq=RequestBuildParam, ClsResp=ResponseTemplate, url="http://localhost:8000/build",
                                                    url_ssh_repository = text_ssh.value,
                                                    checkout_target = text_checkout.value,
                                                    command = text_command.value,
                                                    list_to_pack = list_pack,
                                                    ip_dst = text_ip_dst.value,
                                                    path_folder_dst = text_path_folder_dst.value
                                                    )
        if ret:
            id = ret.data