File: type_hints.rst

package info (click to toggle)
python-motor 3.7.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,572 kB
  • sloc: python: 12,252; javascript: 137; makefile: 74; sh: 8
file content (391 lines) | stat: -rw-r--r-- 14,363 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391

.. _type_hints-example:

Type Hints
==========

.. warning:: Motor will be deprecated on May 14th, 2026, one year after the production release of the PyMongo Async driver. Critical bug fixes will be made until May 14th, 2027.
  We strongly recommend that Motor users migrate to the PyMongo Async driver while Motor is still supported.
  To learn more, see `the migration guide <https://www.mongodb.com/docs/languages/python/pymongo-driver/current/reference/migration/>`_.


As of version 3.3.0, Motor ships with `type hints`_. With type hints, Python
type checkers can easily find bugs before they reveal themselves in your code.

If your IDE is configured to use type hints,
it can suggest more appropriate completions and highlight errors in your code.
Some examples include `PyCharm`_,  `Sublime Text`_, and `Visual Studio Code`_.

You can also use the `mypy`_ tool from your command line or in Continuous Integration tests.

All of the public APIs in Motor are fully type hinted, and
several of them support generic parameters for the
type of document object returned when decoding BSON documents.

Due to `limitations in mypy`_, the default
values for generic document types are not yet provided (they will eventually be ``Dict[str, any]``).

For a larger set of examples that use types, see the Motor `test_typing module`_.

If you would like to opt out of using the provided types, add the following to
your `mypy config`_: ::

    [mypy-motor]
    follow_imports = False


Basic Usage
-----------

Note that a type for :class:`~motor.motor_asyncio.AsyncIOMotorClient` must be specified.  Here we use the
default, unspecified document type:

.. code-block:: python

   from motor.motor_asyncio import AsyncIOMotorClient


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       collection = client.test.test
       inserted = await collection.insert_one({"x": 1, "tags": ["dog", "cat"]})
       retrieved = await collection.find_one({"x": 1})
       assert isinstance(retrieved, dict)

For a more accurate typing for document type you can use:

.. code-block:: python

   from typing import Any, Dict
   from motor.motor_asyncio import AsyncIOMotorClient


   async def main():
       client: AsyncIOMotorClient[Dict[str, Any]] = AsyncIOMotorClient()
       collection = client.test.test
       inserted = await collection.insert_one({"x": 1, "tags": ["dog", "cat"]})
       retrieved = await collection.find_one({"x": 1})
       assert isinstance(retrieved, dict)

Typed Client
------------

:class:`~motor.motor_asyncio.AsyncIOMotorClient` is generic on the document type used to decode BSON documents.

You can specify a :class:`~bson.raw_bson.RawBSONDocument` document type:

.. code-block:: python

   from motor.motor_asyncio import AsyncIOMotorClient
   from bson.raw_bson import RawBSONDocument


   async def main():
       client = AsyncIOMotorClient(document_class=RawBSONDocument)
       collection = client.test.test
       inserted = await collection.insert_one({"x": 1, "tags": ["dog", "cat"]})
       result = await collection.find_one({"x": 1})
       assert isinstance(result, RawBSONDocument)

Subclasses of :py:class:`collections.abc.Mapping` can also be used, such as :class:`~bson.son.SON`:

.. code-block:: python

   from bson import SON
   from motor.motor_asyncio import AsyncIOMotorClient


   async def main():
       client = AsyncIOMotorClient(document_class=SON[str, int])
       collection = client.test.test
       inserted = await collection.insert_one({"x": 1, "y": 2})
       result = await collection.find_one({"x": 1})
       assert result is not None
       assert result["x"] == 1

Note that when using :class:`~bson.son.SON`, the key and value types must be given, e.g. ``SON[str, Any]``.


Typed Collection
----------------

You can use :py:class:`~typing.TypedDict` when using a well-defined schema for the data in a
:class:`~motor.motor_asyncio.AsyncIOMotorClient`. Note that all `schema validation`_ for inserts and updates is done on the server.
These methods automatically add an "_id" field.

.. code-block:: python

   from typing import TypedDict
   from motor.motor_asyncio import AsyncIOMotorClient
   from motor.motor_asyncio import AsyncIOMotorCollection


   class Movie(TypedDict):
       name: str
       year: int


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       collection: AsyncIOMotorCollection[Movie] = client.test.test
       inserted = await collection.insert_one(Movie(name="Jurassic Park", year=1993))
       result = await 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 Motor.
       assert result["_id"]  # type:ignore[typeddict-item]

This same typing scheme works for all of the insert methods (:meth:`~motor.motor_asyncio.AsyncIOMotorCollection.insert_one`,
:meth:`~motor.motor_asyncio.AsyncIOMotorCollection.insert_many`, and :meth:`~motor.motor_asyncio.AsyncIOMotorCollection.bulk_write`).
For ``bulk_write`` both :class:`~pymongo.operations.InsertOne` and :class:`~pymongo.operations.ReplaceOne` operators are generic.

.. code-block:: python

   from typing import TypedDict
   from motor.motor_asyncio import AsyncIOMotorClient
   from motor.motor_asyncio import AsyncIOMotorCollection
   from pymongo.operations import InsertOne


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       collection: AsyncIOMotorCollection[Movie] = client.test.test
       inserted = await collection.bulk_write(
           [InsertOne(Movie(name="Jurassic Park", year=1993))]
       )
       result = await 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 Motor.
       assert result["_id"]  # type:ignore[typeddict-item]

Modeling Document Types with TypedDict
--------------------------------------

You can use :py:class:`~typing.TypedDict` to model structured data.
As noted above, Motor will automatically add an ``_id`` field if it is not present. This also applies to TypedDict.
There are three approaches to this:

1. Do not specify ``_id`` at all. It will be inserted automatically, and can be retrieved at run-time, but will yield a type-checking error unless explicitly ignored.

2. Specify ``_id`` explicitly. This will mean that every instance of your custom TypedDict class will have to pass a value for ``_id``.

3. Make use of :py:class:`~typing.NotRequired`. This has the flexibility of option 1, but with the ability to access the ``_id`` field without causing a type-checking error.

Note: to use :py:class:`~typing.NotRequired` in earlier versions of Python (<3.11), use the ``typing_extensions`` package.

.. code-block:: python

   from typing import TypedDict, NotRequired
   from motor.motor_asyncio import AsyncIOMotorClient
   from motor.motor_asyncio import AsyncIOMotorCollection
   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


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       collection: AsyncIOMotorCollection[Movie] = client.test.test
       inserted = await collection.insert_one(Movie(name="Jurassic Park", year=1993))
       result = await 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 Motor.
       assert result["_id"]  # type:ignore[typeddict-item]

       collection: AsyncIOMotorCollection[ExplicitMovie] = client.test.test
       # Note that the _id keyword argument must be supplied
       inserted = await collection.insert_one(
           ExplicitMovie(_id=ObjectId(), name="Jurassic Park", year=1993)
       )
       result = await collection.find_one({"name": "Jurassic Park"})
       assert result is not None
       # This will not raise a type-checking error.
       assert result["_id"]

       collection: AsyncIOMotorCollection[NotRequiredMovie] = client.test.test
       # Note the lack of _id, similar to the first example
       inserted = await collection.insert_one(
           NotRequiredMovie(name="Jurassic Park", year=1993)
       )
       result = await 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"]


Typed Database
--------------

While less common, you could specify that the documents in an entire database
match a well-defined schema using :py:class:`~typing.TypedDict`.

.. code-block:: python

   from typing import TypedDict
   from motor.motor_asyncio import AsyncIOMotorClient
   from motor.motor_asyncio import AsyncIOMotorDatabase


   class Movie(TypedDict):
       name: str
       year: int


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       db: AsyncIOMotorDatabase[Movie] = client.test
       collection = db.test
       inserted = await collection.insert_one({"name": "Jurassic Park", "year": 1993})
       result = await collection.find_one({"name": "Jurassic Park"})
       assert result is not None
       assert result["year"] == 1993

Typed Command
-------------
When using the :meth:`~motor.motor_asyncio.AsyncIOMotorDatabase.command`, you can specify the document type by providing a custom :class:`~bson.codec_options.CodecOptions`:

.. code-block:: python

   from motor.motor_asyncio import AsyncIOMotorClient
   from bson.raw_bson import RawBSONDocument
   from bson import CodecOptions


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       options = CodecOptions(RawBSONDocument)
       result = await client.admin.command("ping", codec_options=options)
       assert isinstance(result, RawBSONDocument)

Custom :py:class:`collections.abc.Mapping` subclasses and :py:class:`~typing.TypedDict` are also supported.
For :py:class:`~typing.TypedDict`, use the form: ``options: CodecOptions[MyTypedDict] = CodecOptions(...)``.

Typed BSON Decoding
-------------------
You can specify the document type returned by :mod:`bson` decoding functions by providing :class:`~bson.codec_options.CodecOptions`:

.. code-block:: python

      from typing import Any, Dict
      from bson import CodecOptions, encode, decode


      class MyDict(Dict[str, Any]):
          pass


      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"

:class:`~bson.raw_bson.RawBSONDocument` and :py:class:`~typing.TypedDict` are also supported.
For :py:class:`~typing.TypedDict`, use  the form: ``options: CodecOptions[MyTypedDict] = CodecOptions(...)``.


Troubleshooting
---------------

Client Type Annotation
~~~~~~~~~~~~~~~~~~~~~~
If you forget to add a type annotation for a :class:`~motor.motor_asyncio.AsyncIOMotorClient` object you may get the following ``mypy`` error:

.. code-block:: python

  from motor.motor_asyncio import AsyncIOMotorClient

  client = AsyncIOMotorClient()  # error: Need type annotation for "client"

The solution is to annotate the type as ``client: AsyncIOMotorClient`` or ``client: AsyncIOMotorClient[Dict[str, Any]]``.  See `Basic Usage`_.

Incompatible Types
~~~~~~~~~~~~~~~~~~
If you use the generic form of :class:`~motor.motor_asyncio.AsyncIOMotorClient` you
may encounter a ``mypy`` error like:

.. code-block:: python

   from motor.motor_asyncio import AsyncIOMotorClient


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


The solution is to use ``client: AsyncIOMotorClient[Dict[str, Any]]`` as used in
`Basic Usage`_ .

Actual Type Errors
~~~~~~~~~~~~~~~~~~

Other times ``mypy`` will catch an actual error, like the following code:

.. code-block:: python

   from motor.motor_asyncio import AsyncIOMotorClient
   from typing import Mapping


   async def main():
       client: AsyncIOMotorClient = AsyncIOMotorClient()
       await client.test.test.insert_one(
           [{}]
       )  # error: Argument 1 to "insert_one" of "Collection" has
       # incompatible type "List[Dict[<nothing>, <nothing>]]";
       # expected "Mapping[str, Any]"

In this case the solution is to use ``insert_one({})``, passing a document instead of a list.

Another example is trying to set a value on a :class:`~bson.raw_bson.RawBSONDocument`, which is read-only.:

.. code-block:: python

   from bson.raw_bson import RawBSONDocument
   from motor.motor_asyncio import AsyncIOMotorClient


   async def main():
       client = AsyncIOMotorClient(document_class=RawBSONDocument)
       coll = client.test.test
       doc = {"my": "doc"}
       await coll.insert_one(doc)
       retrieved = await 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]

.. _PyCharm: https://www.jetbrains.com/help/pycharm/type-hinting-in-product.html
.. _Visual Studio Code: https://code.visualstudio.com/docs/languages/python
.. _Sublime Text: https://github.com/sublimelsp/LSP-pyright
.. _type hints: https://docs.python.org/3/library/typing.html
.. _mypy: https://mypy.readthedocs.io/en/stable/cheat_sheet_py3.html
.. _limitations in mypy: https://github.com/python/mypy/issues/3737
.. _mypy config: https://mypy.readthedocs.io/en/stable/config_file.html
.. _test_typing module: https://github.com/mongodb/motor/blob/master/test/test_typing.py
.. _schema validation: https://www.mongodb.com/docs/manual/core/schema-validation/#when-to-use-schema-validation