04-28 周日 FastAPI Post请求同时传递文件和普通参数

时间	版本	修改人	描述
04-28 周日	V0.1	宋全恒	新建文档
2024年5月6日14:20:05	V1.0	宋全恒	完成文档的传递

简介

由于在重构FastBuild的时候，为了支持TLS是否启用，在接口中需要同时传递文件参数和其他参数，遇到了这个问题。结果发现由于HTTP的限制，不能同时传递JSON和文件参数。当时花费了较多的实践，因此记录了如下的过程。

代码示例

使用Form表单形式

使用Form表单参数，可以实现，同时结合使用UploadFile可以非常方便。

@router.post("/update-docker-server")
async def update_docker_server_config(host: str = Form(), port: int = Form(), tls_tar_file: UploadFile = File(None)):
    if not validate_host(host):
        return Response.error(f"请输入有效的ip或者域名,参数host: {host}")

    new_docker_server = DBDockerServer(
        host=host,
        port=port,
        tls_verify=False
    )
    tls_verify = False
    if tls_tar_file:
        tls_folder_name = get_ip_address_folder(host)

在相应的swagger页面上，显示如下所示:

同时传递文件和对象参数

如何在FastAPI POST请求中同时添加文件和JSON主体？

由于上述的HTTP限制，因此，无法在维持JSON的结构的同时传递文件参数。

如here所述，用户可以使用File和Form字段同时定义文件和表单数据。

@router.post("/update-docker-server-in-object")
async def update_docker_server_config_in_object(
        docker_server_request: DockerServerRequest = Depends(docker_server_request_checker),
        tls_tar_file: UploadFile = File(None)):
    print(docker_server_request.host)

docker_server_request_checker的定义如下:

from http.client import HTTPException

from fastapi import Form, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, ValidationError


class DockerServerRequest(BaseModel):
    host: str
    port: int = 2375


def docker_server_request_checker(data: str = Form(...)):
    try:
        model = DockerServerRequest.parse_raw(data)
    except ValidationError as e:
        raise HTTPException(
            detail=jsonable_encoder(e.errors()),
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        )
    return model

对于一个比较复杂的data参数，其中多层嵌套也是比较难写的。因为要手动生成这个结构，倒是不如直接使用Form表单化每个数据，第三方调用这个接口的时候，由于key是不用留心的，这样在传递参数的时候，不需要过多的注意力。

最后的实践

由于FastBuild工程在运行时需要提前配置好容器所在的宿主机IP，Harbor的用户名和密码，Docker服务器(两种情形，一种是Docker启用了TLS，则需要上传文件tls-client-certs-jenkins.tar.gz, 另外则是普通的Docker服务)，为了满足这多种情况，尤其是对于TLS的灵活性，因此最终选择了第一种方案，即使用多个Form参数以及文件参数的方式来完成对于FastBuild的统一配置接口

Controller

@router.post("/config-fastbuild")
async def update_docker_server_config_in_object(
        fastbuild_host: str = Form(), fastbuild_port: int = Form(default=48001),
        harbor_username: str = Form(), harbor_password: str = Form(), harbor_registry: str = Form(),
        harbor_registry_dns: str = Form(default=''),
        docker_host: str = Form(), docker_port: int = Form(default=2375), docker_tls_tar_file: UploadFile = File(None)):
    print("fastbuild: ", fastbuild_host, fastbuild_port)
    print("harbor: ", harbor_username, harbor_password, harbor_registry, harbor_registry_dns)

    if not all(map(validate_host, [fastbuild_host, harbor_registry, harbor_registry_dns, docker_host])):
        return Response.error(data="fastbuild_host, harbor_registry_host, harbor_registry_dns, docker_host均应为有效的ip或者域名")

    db_host = DBHost(host_ip=fastbuild_host, host_port=fastbuild_port)
    db_harbor = DBHarbor(username=harbor_username, password=harbor_password, registry=harbor_registry, registry_dns=harbor_registry_dns)
    db_docker = DBDockerServer(host=docker_host, port=docker_port, tls_verify=False)
    if docker_tls_tar_file:
        tls_folder_name = get_ip_address_folder(db_docker.host)
        tls_dir = save_and_extract_tar(docker_tls_tar_file, system_config.get_tls_dir(), tls_folder_name)
        tls_files = ["ca-jenkins.pem", "cert-jenkins.pem", "key-jenkins.pem"]
        if not all(file in get_files_in_directory(tls_dir) for file in tls_files):
            return Response.error(f"请上传正确的tls文件,当前上传的文件为{docker_tls_tar_file.filename},解压后不包含{' '.join(tls_files)}")

        db_docker.tls_verify = True
        db_docker.client_cert_path = os.path.join(tls_dir, "cert-jenkins.pem")
        db_docker.ca_path = os.path.join(tls_dir, "ca-jenkins.pem")
        db_docker.client_key_path = os.path.join(tls_dir, "key-jenkins.pem")
    try:
        image_utils = ImageUtils(db_docker, db_harbor)
    except DockerException as exe:
        print(f"发生异常: {exe}")
        raise FBException(code=123, message=f"使用提供的docker和harbor信息，进行登录测试，测试失败，请检查,错误信息为{str(exe)}")

    DBHostService.save(db_host)
    DBHarborService.save(db_harbor)
    DBDockerServerService.save(db_docker)

    return Response.success(data="成功完成为FastBuild配置需要的宿主机信息，Docker信息以及Harbor信息")

其中validate_host用于判断输入的字符串是一个有效的ip或者域名，具体定义如下：

def validate_host(host: str):
    # 匹配 IP 地址的正则表达式
    ip_pattern = r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$"
    # 匹配域名地址的正则表达式
    domain_pattern = r"^(?!:\/\/)([a-zA-Z0-9-_]+\.)*[a-zA-Z0-9][a-zA-Z0-9-_]+(\.[a-zA-Z]{2,11})$"

    # 尝试匹配 IP 地址和域名地址的正则表达式
    if re.match(ip_pattern, host) or re.fullmatch(domain_pattern, host):
        return True  # 主机地址合法
    else:
        return False  # 主机地址不合法

而get_ip_address_folder则是用来根据ip地址获取对应的一个目录，其中包含了一个6位的随机数字

def get_ip_address_folder(ip: str):
    """
    将 IP 地址或域名转换为文件夹名称,用于存储tls相关的文件
    :param ip: IP 地址或域名
    :return: 文件夹名称
    """

    random_number = ''.join(random.choices('0123456789', k=6))

    # 将 IP 或域名中的 . 替换为 -
    converted_ip = str(ip).replace(".", "-")

    # 将随机数添加到转换后的字符串中
    result = f"{converted_ip}-{random_number}"

    return result

swagger请求

如下，在配置时，传入多个参数。

由于按照Form参数类型传入值，因此需要再代码中重新组织这些参数，完成序列化。

    db_host = DBHost(host_ip=fastbuild_host, host_port=fastbuild_port)
    db_harbor = DBHarbor(username=harbor_username, password=harbor_password, registry=harbor_registry, registry_dns=harbor_registry_dns)
    db_docker = DBDockerServer(host=docker_host, port=docker_port, tls_verify=False)

总结

这主要是因为之前FastBuild系统的启动，需要依赖一个外部Docker服务器来进行系统镜像的构建，因此，在启动的时候，需要事先准备好TLS支持的配置文件"ca-jenkins.pem", “cert-jenkins.pem”, “key-jenkins.pem”，而这样优化之后，则可以先启动FastBuild，通过接口完成对于FastBuild的配置，从而减少FastBuild的依赖，这真的是很好的一种工程实践。