类型提示#

从4.1版开始,PyMongo附带 type hints 。有了类型提示,Python类型检查器可以很容易地在代码中发现错误之前发现它们。

如果您的IDE配置为使用类型提示,它可以建议更适当的完成并突出显示代码中的错误。一些例子包括 PyCharmSublime Text ,以及 Visual Studio Code

您也可以使用 mypy 工具从您的命令行或在持续集成测试中。

PyMongo中的所有公共API都是完全类型提示的,其中几个支持解码BSON文档时返回的Document对象类型的泛型参数。

由于 limitations in mypy ,尚未提供通用文档类型的缺省值(它们最终将 Dict[str, any] )。

有关使用类型的更多示例,请参见 test_typing module

如果您希望选择不使用提供的类型,请将以下内容添加到您的 mypy config **

[mypy-pymongo]
follow_imports = False

基本用法#

请注意,类型为 MongoClient 必须指定。在这里,我们使用默认的未指定文档类型:

>>> from pymongo import MongoClient
>>> client: MongoClient = MongoClient()
>>> collection = client.test.test
>>> inserted = collection.insert_one({"x": 1, "tags": ["dog", "cat"]})
>>> retrieved = collection.find_one({"x": 1})
>>> assert isinstance(retrieved, dict)

要更准确地键入文档类型,您可以使用:

>>> from typing import Any, Dict
>>> from pymongo import MongoClient
>>> client: MongoClient[Dict[str, Any]] = MongoClient()
>>> collection = client.test.test
>>> inserted = collection.insert_one({"x": 1, "tags": ["dog", "cat"]})
>>> retrieved = collection.find_one({"x": 1})
>>> assert isinstance(retrieved, dict)

类型化客户端#

MongoClient 是用于解码BSON文档的文档类型的泛型。

您可以指定一个 RawBSONDocument 文档类型:

>>> from pymongo import MongoClient
>>> from bson.raw_bson import RawBSONDocument
>>> client = MongoClient(document_class=RawBSONDocument)
>>> collection = client.test.test
>>> inserted = collection.insert_one({"x": 1, "tags": ["dog", "cat"]})
>>> result = collection.find_one({"x": 1})
>>> assert isinstance(result, RawBSONDocument)

的子类 collections.abc.Mapping 也可以使用,例如 SON

>>> from bson import SON
>>> from pymongo import MongoClient
>>> client = MongoClient(document_class=SON[str, int])
>>> collection = client.test.test
>>> inserted = collection.insert_one({"x": 1, "y": 2})
>>> result = collection.find_one({"x": 1})
>>> assert result is not None
>>> assert result["x"] == 1

请注意,在使用 SON ,必须给出键和值类型,例如 SON[str, Any]

类型化集合#

您可以使用 TypedDict (Python3.8+)在使用明确定义的架构处理 Collection 。请注意,所有 schema validation 插入和更新是在服务器上完成的。这些方法会自动添加“_id”字段。

>>> from typing import TypedDict
>>> from pymongo import MongoClient
>>> from pymongo.collection import Collection
>>> class Movie(TypedDict):
...     name: str
...     year: int
...
>>> client: MongoClient = MongoClient()
>>> collection: Collection[Movie] = client.test.test
>>> inserted = collection.insert_one(Movie(name="Jurassic Park", year=1993))
>>> result = collection.find_one({"name": "Jurassic Park"})
>>> assert result is not None
>>> assert result["year"] == 1993
>>> # This will raise a type-checking error, despite being present, because it is added by PyMongo.
>>> assert result["_id"]  # type:ignore[typeddict-item]

相同的类型化方案适用于所有的插入方法 (insert_one()insert_many() ,以及 bulk_write() )。为 bulk_write 两者都有 InsertOneReplaceOne 运算符是通用的。

>>> from typing import TypedDict
>>> from pymongo import MongoClient
>>> from pymongo.operations import InsertOne
>>> from pymongo.collection import Collection
>>> client: MongoClient = MongoClient()
>>> collection: Collection[Movie] = client.test.test
>>> inserted = collection.bulk_write([InsertOne(Movie(name="Jurassic Park", year=1993))])
>>> result = collection.find_one({"name": "Jurassic Park"})
>>> assert result is not None
>>> assert result["year"] == 1993
>>> # This will raise a type-checking error, despite being present, because it is added by PyMongo.
>>> assert result["_id"]  # type:ignore[typeddict-item]

使用TyedDict对文档类型进行建模#

您可以使用 TypedDict (Python3.8+)来为结构化数据建模。如上所述,PyMongo将自动添加一个 _id 字段(如果不存在)。这也适用于TypeDict。有三种方法可以做到这一点:

  1. 不指定 _id 完全没有。它将自动插入,并可在运行时检索,但除非显式忽略,否则将产生类型检查错误。

  2. 指定 _id explicitly. This will mean that every instance of your custom TypedDict class will have to pass a value for _ Id

  3. 利用…… NotRequired 。这具有选项1的灵活性,但能够访问 _id 字段,而不会导致类型检查错误。

注:要使用 TypedDictNotRequired 在早期版本的Python(<3.8、<3.11)中,使用 typing_extensions 包裹。

>>> from typing import TypedDict, NotRequired
>>> from pymongo import MongoClient
>>> from pymongo.collection import Collection
>>> from bson import ObjectId
>>> class Movie(TypedDict):
...     name: str
...     year: int
...
>>> class ExplicitMovie(TypedDict):
...     _id: ObjectId
...     name: str
...     year: int
...
>>> class NotRequiredMovie(TypedDict):
...     _id: NotRequired[ObjectId]
...     name: str
...     year: int
...
>>> client: MongoClient = MongoClient()
>>> collection: Collection[Movie] = client.test.test
>>> inserted = collection.insert_one(Movie(name="Jurassic Park", year=1993))
>>> result = collection.find_one({"name": "Jurassic Park"})
>>> assert result is not None
>>> # This will yield a type-checking error, despite being present, because it is added by PyMongo.
>>> assert result["_id"]  # type:ignore[typeddict-item]
>>> collection: Collection[ExplicitMovie] = client.test.test
>>> # Note that the _id keyword argument must be supplied
>>> inserted = collection.insert_one(
...     ExplicitMovie(_id=ObjectId(), name="Jurassic Park", year=1993)
... )
>>> result = collection.find_one({"name": "Jurassic Park"})
>>> assert result is not None
>>> # This will not raise a type-checking error.
>>> assert result["_id"]
>>> collection: Collection[NotRequiredMovie] = client.test.test
>>> # Note the lack of _id, similar to the first example
>>> inserted = collection.insert_one(NotRequiredMovie(name="Jurassic Park", year=1993))
>>> result = collection.find_one({"name": "Jurassic Park"})
>>> assert result is not None
>>> # This will not raise a type-checking error, despite not being provided explicitly.
>>> assert result["_id"]

类型化数据库#

虽然不太常见,但可以使用以下命令指定整个数据库中的文档与定义良好的模式匹配 TypedDict (Python3.8+)。

>>> from typing import TypedDict
>>> from pymongo import MongoClient
>>> from pymongo.database import Database
>>> class Movie(TypedDict):
...     name: str
...     year: int
...
>>> client: MongoClient = MongoClient()
>>> db: Database[Movie] = client.test
>>> collection = db.test
>>> inserted = collection.insert_one({"name": "Jurassic Park", "year": 1993})
>>> result = collection.find_one({"name": "Jurassic Park"})
>>> assert result is not None
>>> assert result["year"] == 1993

键入的命令#

在使用 command() ,您可以通过提供自定义的 CodecOptions

>>> from pymongo import MongoClient
>>> from bson.raw_bson import RawBSONDocument
>>> from bson import CodecOptions
>>> client: MongoClient = MongoClient()
>>> options = CodecOptions(RawBSONDocument)
>>> result = client.admin.command("ping", codec_options=options)
>>> assert isinstance(result, RawBSONDocument)

自定义 collections.abc.Mapping 子类和 TypedDict (Python3.8+)也受支持。为 TypedDict ,请使用以下形式: options: CodecOptions[MyTypedDict] = CodecOptions(...)

打字BSON译码#

可以指定由返回的文档类型 bson 通过提供以下功能来解码函数 CodecOptions

>>> from typing import Any, Dict
>>> from bson import CodecOptions, encode, decode
>>> class MyDict(Dict[str, Any]):
...     def foo(self):
...         return "bar"
...
>>> options = CodecOptions(document_class=MyDict)
>>> doc = {"x": 1, "y": 2}
>>> bsonbytes = encode(doc, codec_options=options)
>>> rt_document = decode(bsonbytes, codec_options=options)
>>> assert rt_document.foo() == "bar"

RawBSONDocumentTypedDict (Python3.8+)也受支持。为 TypedDict ,请使用以下形式: options: CodecOptions[MyTypedDict] = CodecOptions(...)

故障排除#

客户端类型注释#

如果您忘记为 MongoClient 对象,您可能会得到以下信息 mypy 错误::

from pymongo import MongoClient
client = MongoClient()  # error: Need type annotation for "client"

解决方案是将该类型注释为 client: MongoClientclient: MongoClient[Dict[str, Any]] 。看见 Basic Usage

不兼容的类型#

如果使用泛型形式的 MongoClient 您可能会遇到 mypy 错误如下::

from pymongo import MongoClient

client: MongoClient = MongoClient()
client.test.test.insert_many(
    {"a": 1}
)  # error: Dict entry 0 has incompatible type "str": "int";
   # expected "Mapping[str, Any]": "int"

解决方案是使用 client: MongoClient[Dict[str, Any]] 中使用的 Basic Usage

实际类型错误#

其他时间 mypy 将捕获实际错误,如以下代码所示:

from pymongo import MongoClient
from typing import Mapping
client: MongoClient = MongoClient()
client.test.test.insert_one(
    [{}]
)  # error: Argument 1 to "insert_one" of "Collection" has
   # incompatible type "List[Dict[<nothing>, <nothing>]]";
   # expected "Mapping[str, Any]"

在这种情况下,解决方案是使用 insert_one({}) ,传递文档而不是列表。

另一个示例是尝试在 RawBSONDocument ,为只读。::

from bson.raw_bson import RawBSONDocument
from pymongo import MongoClient

client = MongoClient(document_class=RawBSONDocument)
coll = client.test.test
doc = {"my": "doc"}
coll.insert_one(doc)
retrieved = coll.find_one({"_id": doc["_id"]})
assert retrieved is not None
assert len(retrieved.raw) > 0
retrieved[
    "foo"
] = "bar"  # error: Unsupported target for indexed assignment
           # ("RawBSONDocument")  [index]