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

Question

我有一个简单的 REST api，这是一个使用 FastAPI 和 mongo 数据库作为后端创建的书店（我使用 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]

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

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

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

有人可以告诉我这里的最佳实践是什么吗？有什么可以适用于我的要求和用例吗？

非常感谢。

Answer 1

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

回答您的问题：

这取决于您必须管理的负载和您使用的开发堆栈。通常它是在数据库级别完成的，因为检索前 110 个和删除前 100 个是非常愚蠢和消耗资源的（数据库会为你做）。
对我来说，如何通过 fastapi 实现它似乎很简单：只需将参数 limit: int = 10 和 [=13= 添加到您的 get 函数中] 并在数据库的过滤功能中使用它们。 Fastapi 将为您检查数据类型，同时您可以检查限制是否为负数或大于 100。
它说没有灵丹妙药，因为 mongo 的 skip 功能表现不佳。因此，他认为第二种选择更好，只是为了表演。如果您有数十亿份文件（例如亚马逊），那么，可能需要使用不同的东西，尽管当您的网站增长到那么多时，我想您将有钱支付整个团队的费用专家来解决问题，并可能开发您自己的数据库。

TL;DR

总而言之，limit 和 skip 方法是最常见的方法。它通常在数据库级别完成，以减少应用程序和带宽的工作量。

Mongo 在跳过和限制结果方面不是很有效。如果您的数据库有一百万个文档，那么我认为您甚至不会注意到。您甚至可以为此类工作负载使用关系数据库。您可以随时对现有选项进行基准测试，然后选择最合适的选项。

我不太了解mongo，但我知道一般来说，索引可以帮助限制和跳过记录（在这种情况下是文档），但我不确定mongo 还有。

Answer 2

您可以使用这个包来分页： https://pypi.org/project/fastapi-paginate

使用方法： https://github.com/nazmulnnb/fastapi-paginate/blob/main/examples/pagination_motor.py

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

How to implement pagination for fastapi with mongo db(Motor)

rest

pagination

mongodb

pymongo

fastapi

TL;DR