使用类作为依赖项的正确方法是什么（参数检查）？

Question

这里的主要目标是映射 UI 发送的参数以匹配我的数据库表。

---

现在你可以说我疯了，但是前一秒还可以，现在不行了。

我这里有几个类和一个函数（在

S_Sorting

中）

class ScanSort(str, Enum):
    ASSET_NAME = "assetName"
     CREATED_AT = "createdAt"
    ...

class BaseSorting(BaseModel):
    sort_order: Annotated[SortOrder, Query(...)] = SortOrder.ASCENDING
...

class S_Sorting(BaseSorting):
    sort_by: Annotated[ScanSort, Query(...)] = ScanSort.CREATED_AT
...

  @staticmethod
    def scan_table_mapping(passed_param: str) -> str:
        mapping = {
            ...,
            "createdAt": "scans.created_at",
            ..
        }
        return mapping.get(passed_param, "scans.created_at")

我的函数依赖于对数据进行排序（在 postgres 数据库中）。

 def build_query(
            self,
            ...,
            sorting: S_Sorting,


sort_by_mapped = S_Sorting.scan_table_mapping(sorting.sort_by)

direction = asc if sorting.sort_order == SortOrder.ASCENDING else desc
        query = query.order_by(direction(text(sort_by_mapped)))

所以如果没有

sort

参数，那么它默认按照

S_Sorting

排序，工作正常。

现在，当我传入

sort=createdAt

时，它应该对映射值进行排序。

但是，这是我得到的错误：

{
   "detail": [
       {
           "type": "enum",
           "loc": [
               "query",
               "sort_by"
           ],
           "msg": "Input should be '..., 'createdAt' or 'updatedAt'",
           "input": " createdAt",
           "ctx": {
               "expected": "'..., 'createdAt' or 'updatedAt'"
           }
       }
   ]
}

我确实给予了它所期望的东西。我将类作为依赖项做错了什么？

Answer 1

这个问题太笼统了，我仍然能想到的是：让您对依赖项的使用有一个基本且完整的了解。但在这种情况下，我将使用异步来更好地完成独立的并行流程。您需要自己生成此代码的查询并在需要的地方输入..

您可以使用类定义排序规则，类指定顺序（升序或降序）和条件（例如添加日期或上次玩的日期）。函数将这些标准转换为计算机可以理解的格式，确保数据正确排序。

当您通过网络路由发送排序请求时，代码会检查是否存在多余空格等错误，遵循定义的规则，然后返回排序后的数据。这可确保您的玩具（或数据）根据您选择的标准整齐地组织起来。

from enum import Enum
from pydantic import BaseModel, Query
from typing import Annotated

class SortOrder1(str, Enum):
    ASCENDING = "asc"
    DESCENDING = "desc"

class ScanSort1(str, Enum):
    ASSET_NAME = "assetName"
    CREATED_AT = "createdAt"
    UPDATED_AT = "updatedAt"

class BaseSorting2(BaseModel):
    sort_order: Annotated[SortOrder1, Query(sample)] = SortOrder1.ASCENDING

class S1Sorting(BaseSorting):
    sort_by: Annotated[ScanSort, Query(...)] = ScanSort.CREATED_AT

    @staticmethod
    def check_table_mapping(passed_param):
        mapping = {
            "assetName": "scans.asset_name",
            "createdAt": "scans.created_at",
            "updatedAt": "scans.updated_at",
            }
        return mapping.get(passed_param, "scans.created_at")

def generate_query(self, sorting: S1Sorting):
    sort_by_mapped = S1Sorting.scan_table_mapping(sorting.sort_by)
    direction = asc if sorting.sort_order == SortOrder.ASCENDING else desc
    query = query.order_by(direction(text(sort_by_mapped)))
    return query

from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/items/")
def get_items(sorting: S1Sorting = Depends()):
    query = generate_query(sorting=sorting)
    return {"query": str(query)}

from fastapi import Request

@app.middleware("http")
async def strip_whitespace_middleware(request: Request, call_next):
    query_params = request.query_params._dict
    stripped_query_params = {k: v.strip() for k, v in     query_params.items()}
    request._query_params = stripped_query_params
    response = await call_next(request)
    return response

它通过根据特定规则对数据进行排序来帮助组织数据库中的数据。可以将其想象为按不同类别（例如颜色、类型或尺寸）整理玩具盒。