时间戳

首页 > 解决方案 > 如何使用 mongo db(Motor) 为 fastapi 实现分页

mongodb - 如何使用 mongo db(Motor) 为 fastapi 实现分页

问题描述

我有一个简单的 REST api，它是一个使用 FastAPI 和 mongo db 作为后端创建的书店（我已经用作Motor库而不是Pymongo）。我有一个GET端点来获取数据库中的所有书籍，它也支持查询字符串（例如：用户可以搜索具有单个作者或流派类型等的书籍）。

以下是此端点的相应代码： routers.py


@router.get("/books", response_model=List[models.AllBooksResponse])
async def get_the_list_of_all_books(
    authors: Optional[str] = None,
    genres: Optional[str] = None,
    published_year: Optional[str] = None,
) -> List[Dict[str, Any]]:
    if authors is None and genres is None and published_year is None:
        all_books = [book for book in await mongo.BACKEND.get_all_books()]
    else:
        all_books = [
            book
            for book in await mongo.BACKEND.get_all_books(
                authors=authors.strip('"').split(",") if authors is not None else None,
                genres=genres.strip('"').split(",") if genres is not None else None,
                published_year=datetime.strptime(published_year, "%Y")
                if published_year is not None
                else None,
            )
        ]

    return all_books

对应型号：

class AllBooksResponse(BaseModel):
    name: str
    author: str
    link: Optional[str] = None

    def __init__(self, name, author, **data):
        super().__init__(
            name=name, author=author, link=f"{base_uri()}book/{data['book_id']}"
        )

以及用于获取数据的后端功能：

class MongoBackend:
    def __init__(self, uri: str) -> None:
        self._client = motor.motor_asyncio.AsyncIOMotorClient(uri)

    async def get_all_books(
        self,
        authors: Optional[List[str]] = None,
        genres: Optional[List[str]] = None,
        published_year: Optional[datetime] = None,
    ) -> List[Dict[str, Any]]:
        find_condition = {}
        if authors is not None:
            find_condition["author"] = {"$in": authors}
        if genres is not None:
            find_condition["genres"] = {"$in": genres}
        if published_year is not None:
            find_condition["published_year"] = published_year
        cursor = self._client[DB][BOOKS_COLLECTION].find(find_condition, {"_id": 0})
        return [doc async for doc in cursor]

现在我想为此端点实现分页。在这里我有几个问题：

  1. 在数据库级别或应用程序级别进行分页好吗？
  2. 我们是否有一些开箱即用的库可以帮助我在 fastapi 中做到这一点？我检查了https://pypi.org/project/fastapi-pagination/的文档，但这似乎更针对 SQL 数据库
  3. 我还查看了这个链接：https ://www.codementor.io/@arpitbhayani/fast-and-efficient-pagination-in-mongodb-9095flbqr讨论了在 Mongo db 中执行此操作的不同方法，但我认为只有第一个选项（使用limitand skip）对我有用，因为当我使用其他过滤器参数（例如作者和流派）时，我也想让它工作，除非我第一次查询到，否则我无法知道 ObjectId获取数据，然后我想做分页。

但是这个问题无处不在我看到使用limit并且skip不鼓励。

有人可以让我知道这里的最佳实践是什么，并且可以适用于我的要求和用例吗？

提前谢谢了。

标签： mongodbpaginationpymongorestfastapi

解决方案

这样的问题没有正确或错误的答案。很大程度上取决于您使用的技术堆栈以及您拥有的上下文，还要考虑您编写的软件以及您使用的软件 (mongo) 的未来方向。

回答您的问题：

  1. 这取决于您必须管理的负载和您使用的开发堆栈。通常它是在数据库级别完成的，因为检索前 110 个并删除前 100 个是非常愚蠢且耗费资源的（数据库会为您完成）。

  2. 对我来说，如何通过以下方式完成它似乎很简单fastapi：只需将get参数添加到您的函数中limit: int = 10，skip: int = 0然后在数据库的过滤功能中使用它们。Fastapi将为您检查数据类型，同时您可以检查限制不是负数或高于，例如 100。

  3. 它说没有灵丹妙药，而且由于skipmongo 的功能表现不佳。因此他认为第二种选择更好，只是为了表演。如果您有数十亿份文件（例如亚马逊），那么可能会使用不同的东西，尽管当您的网站发展得如此之快时，我想您将有钱支付整个团队的费用专家来解决问题并可能开发您自己的数据库。

TL;博士

最后，limitandskip方法是最常见的一种。它通常在数据库级别完成，以减少应用程序的工作量和带宽。

Mongo 在跳过和限制结果方面效率不高。如果您的数据库有，比如说一百万个文档，那么我认为您甚至不会注意到。您甚至可以将关系数据库用于此类工作负载。您始终可以对您拥有的选项进行基准测试并选择最合适的选项。

我对 mongo 了解不多，但我知道通常索引可以帮助限制和跳过记录（在这种情况下为文档），但我不确定 mongo 是否也是如此。

推荐阅读