Skip to content

Raw query usage

As ODMantic doesn't completely wrap the MongoDB API, some helpers are provided to be enhance the usability while building raw queries and interacting with raw documents.

Raw query helpers

Collection name

You can get the collection name associated to a model by using the unary + operator on the model class. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from odmantic import Model


class User(Model):
    name: str


collection_name = +User
print(collection_name)
#> user

Motor collection

The AIOEngine object can provide you directly the motor collection (AsyncIOMotorCollection) linked to the motor client used by the engine. To achieve this, you can use the AIOEngine.get_collection method.

 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
from odmantic import AIOEngine, Model


class User(Model):
    name: str


engine = AIOEngine()
motor_collection = engine.get_collection(User)
print(motor_collection)
#> AsyncIOMotorCollection(
#>     Collection(
#>         Database(
#>             MongoClient(
#>                 host=["localhost:27017"],
#>                 document_class=dict,
#>                 tz_aware=False,
#>                 connect=False,
#>                 driver=DriverInfo(name="Motor", version="2.2.0", platform="asyncio"),
#>             ),
#>             "test",
#>         ),
#>         "user",
#>     )
#> )

Key name of a field

Since some field might have some customized key names, you can get the key name associated to a field by using the unary + operator on the model class. As well, to ease the use of aggregation pipelines where you might need to reference your field ($field), you can double the operator (i.e use ++) to get the field reference name.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from odmantic import Field, Model


class User(Model):
    name: str = Field(key_name="username")


print(+User.name)
#> username

print(++User.name)
#> $username

Raw MongoDB documents

Parsing documents

You can parse MongoDB document to instances using the parse_doc method.

Tip

If the provided documents contain extra fields, ODMantic will ignore them. This can be especially useful in aggregation pipelines.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from bson import ObjectId

from odmantic import Field, Model


class User(Model):
    name: str = Field(key_name="username")


document = {"username": "John", "_id": ObjectId("5f8352a87a733b8b18b0cb27")}

user = User.parse_doc(document)
print(repr(user))
#> User(id=ObjectId('5f8352a87a733b8b18b0cb27'), name='John')

Dumping documents

You can generate a document from instances using the doc method. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from odmantic import Field, Model


class User(Model):
    name: str = Field(key_name="username")


user = User(name="John")
print(user.doc())
#> {'username': 'John', '_id': ObjectId('5f8352a87a733b8b18b0cb27')}

Advanced parsing behavior

Default values

While parsing documents, ODMantic will use the default values provided in the Models to populate the missing fields from the documents:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from bson import ObjectId

from odmantic import Model


class Player(Model):
    name: str
    level: int = 1


document = {"name": "Leeroy", "_id": ObjectId("5f8352a87a733b8b18b0cb27")}

user = Player.parse_doc(document)
print(repr(user))
#> Player(
#>     id=ObjectId("5f8352a87a733b8b18b0cb27"),
#>     name="Leeroy",
#>     level=1,
#> )

Default factories

For the field with default factories provided through the Field descriptor though, by default they wont be populated.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from datetime import datetime

from bson import ObjectId

from odmantic import Model
from odmantic.exceptions import DocumentParsingError
from odmantic.field import Field


class User(Model):
    name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)


document = {"name": "Leeroy", "_id": ObjectId("5f8352a87a733b8b18b0cb27")}

try:
    User.parse_doc(document)
except DocumentParsingError as e:
    print(e)
    #> 1 validation error for User
    #> created_at
    #>   key not found in document (type=value_error.keynotfoundindocument; key_name='created_at')
    #> (User instance details: id=ObjectId('5f8352a87a733b8b18b0cb27'))

In the previous example, using the default factories could create data inconsistencies and in this case, it would probably be more suitable to perform a manual migration to provide the correct values.

Still, the parse_doc_with_default_factories Config option can be used to allow the use of the default factories while parsing documents:

 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
from datetime import datetime

from bson import ObjectId

from odmantic import Model
from odmantic.exceptions import DocumentParsingError
from odmantic.field import Field


class User(Model):
    name: str
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        parse_doc_with_default_factories = True


document = {"name": "Leeroy", "_id": ObjectId("5f8352a87a733b8b18b0cb27")}

user = User.parse_doc(document)
print(repr(user))
#> User(
#>     id=ObjectId("5f8352a87a733b8b18b0cb27"),
#>     name="Leeroy",
#>     updated_at=datetime.datetime(2020, 11, 8, 23, 28, 19, 980000),
#> )

Aggregation example

In the following example, we will demonstrate the use of the previous helpers to build an aggregation pipeline. We will first consider a Rectangle model with two float fields (height and length). We will then fetch the rectangles with an area that is less than 10. To finish, we will reconstruct Rectangle instances from this query.

 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
from odmantic import AIOEngine, Model


class Rectangle(Model):
    length: float
    width: float


rectangles = [
    Rectangle(length=0.1, width=1),
    Rectangle(length=3.5, width=1),
    Rectangle(length=2.87, width=5.19),
    Rectangle(length=1, width=10),
    Rectangle(length=0.1, width=100),
]

engine = AIOEngine()
await engine.save_all(rectangles)

collection = engine.get_collection(Rectangle)
pipeline = []
# Add an area field
pipeline.append(
    {
        "$addFields": {
            "area": {
                "$multiply": [++Rectangle.length, ++Rectangle.width]
            }  # Compute the area remotely
        }
    }
)
# Filter only rectanges with an area lower than 10
pipeline.append({"$match": {"area": {"$lt": 10}}})
# Project to keep only the defined fields (this step is optional)
pipeline.append(
    {
        "$project": {
            +Rectangle.length: True,
            +Rectangle.width: True,
        }  # Specifying "area": False is unnecessary here
    }
)
documents = await collection.aggregate(pipeline).to_list(length=None)
small_rectangles = [Rectangle.parse_doc(doc) for doc in documents]
print(small_rectangles)
#> [
#>     Rectangle(id=ObjectId("..."), length=0.1, width=1.0),
#>     Rectangle(id=ObjectId("..."), length=3.5, width=1.0),
#> ]