使用Pydantic的带有可选元素的API JSON架构验证

问题描述

我正在使用pydantic中的fastapi和BaseModel来验证和记录JSON模式以获取API返回。

这对于固定的返回值效果很好,但是我有一些可更改返回值的可选参数,因此我希望将其包括在验​​证中,但是当参数丢失并且API中未返回该字段时,它不会失败。

例如:我有一个可选的布尔参数transparency,当它设置为true时,我返回一个名为search_transparency的块,并返回了弹性查询。

{
  "info": {
    "totalrecords": 52
  },"records": [],"search_transparency": {"full_query": "blah blah"}
}

如果未设置参数transparency=true,我希望返回值是:

{
  "info": {
    "totalrecords": 52
  },"records": []
}

但是,当我在pydantic中将该元素设置为Optional时,却得到了以下返回:

{
  "info": {
    "totalrecords": 52
  },"search_transparency": None
}

对于记录下的字段,我也有类似的内容。默认值为最小字段返回,但是如果您设置参数full=true,则将返回更多字段。我想以类似的方式处理此问题,只是缺少字段,而不显示值为None

这就是我用pydantic处理它的方式:

class Info(BaseModel):
    totalrecords: int

class Transparency(BaseModel):
    full_query: str

class V1Place(BaseModel):
    name: str

class V1PlaceAPI(BaseModel):
    info: Info
    records: List[V1Place] = []
    search_transparency: Optional[Transparency]

这就是我使用fastapi强制执行验证的方式:

@app.get("/api/v1/place/search",response_model=V1PlaceAPI,tags=["v1_api"])

我怀疑也许我试图实现的是不良的API实践,也许我不应该获得可变的回报。

我应该创建多个单独的端点来处理吗?

例如api/v1/place/search?q=testapi/v1/place/full/transparent/search?q=test

编辑

我的API函数的更多细节:

@app.get("/api/v1/place/search",tags=["v1_api"])

def v1_place_search(q: str = Query(None,min_length=3,max_length=500,title="search through all place fields"),transparency: Optional[bool] = None,offset: Optional[int] = Query(0),limit: Optional[int] = Query(15)):

    search_limit = offset + limit

    results,transparency_query = ESQuery(client=es_client,index='places',transparency=transparency,track_hits=True,offset=offset,limit=search_limit)

    return v1_place_parse(results.to_dict(),show_transparency=transparency_query)

其中ESQuery只是返回一个Elasticsearch响应。 这是我的解析函数:

def v1_place_parse(resp,show_transparency=None):
    """This takes a response from elasticsearch and parses it for our legacy V1 elasticapi

    Args:
        resp (dict): This is the response from Search.execute after passing to_dict()

    Returns:
        dict: A dictionary that is passed to API
    """

    new_resp = {}
    total_records = resp['hits']['total']['value']
    query_records = len(resp.get('hits',{}).get('hits',[]))

    new_resp['info'] = {'totalrecords': total_records,'totalrecords_relation': resp['hits']['total']['relation'],'totalrecordsperquery': query_records,}
    if show_transparency is not None:
        search_string = show_transparency.get('query','')
        new_resp['search_transparency'] = {'full_query': str(search_string),'components': {}}
    new_resp['records'] = []
    for hit in resp.get('hits',[]):
        new_record = hit['_source']
        new_resp['records'].append(new_record)

    return new_resp

解决方法

暂无找到可以解决该程序问题的有效方法,小编努力寻找整理中!

如果你已经找到好的解决方法,欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@)