Skip to content

odmantic.session

Bases: AIOSessionBase, AsyncContextManager

An AsyncIO session object for ordering sequential operations.

Sessions can be created from the engine directly by using the AIOEngine.session method.

Example usage as a context manager: 

engine = AIOEngine(...)
async with engine.session() as session:
    john = await session.find(User, User.name == "John")
    john.name = "Doe"
    await session.save(john)

Example raw usage: 

engine = AIOEngine(...)
session = engine.session()
await session.start()
john = await session.find(User, User.name == "John")
john.name = "Doe"
await session.save(john)
await session.end()

Source code in odmantic/session.py 
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
class AIOSession(AIOSessionBase, AsyncContextManager):
    """An AsyncIO session object for ordering sequential operations.


    Sessions can be created from the engine directly by using the
    [AIOEngine.session][odmantic.engine.AIOEngine.session] method.

    Example usage as a context manager:
    ```python
    engine = AIOEngine(...)
    async with engine.session() as session:
        john = await session.find(User, User.name == "John")
        john.name = "Doe"
        await session.save(john)
    ```

    Example raw usage:
    ```python
    engine = AIOEngine(...)
    session = engine.session()
    await session.start()
    john = await session.find(User, User.name == "John")
    john.name = "Doe"
    await session.save(john)
    await session.end()
    ```
    """

    def __init__(self, engine: ODMEngine.AIOEngine):
        self.engine = engine
        self.session: Optional[AsyncIOMotorClientSession] = None

    @property
    def is_started(self) -> bool:
        return self.session is not None

    def get_driver_session(self) -> AsyncIOMotorClientSession:
        """Return the underlying Motor Session"""
        if self.session is None:
            raise RuntimeError("session not started")
        return self.session

    async def start(self) -> None:
        """Start the logical Mongo session."""
        if self.is_started:
            raise RuntimeError("Session is already started")
        self.session = await self.engine.client.start_session()

    async def end(self) -> None:
        """Finish the logical session."""
        if self.session is None:
            raise RuntimeError("Session is not started")
        await self.session.end_session()
        self.session = None

    async def __aenter__(self) -> "AIOSession":
        await self.start()
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        await self.end()

    def transaction(self) -> AIOTransaction:
        """Create a transaction in the existing session"""
        return AIOTransaction(self)

end() async

Finish the logical session.

Source code in odmantic/session.py 
245
246
247
248
249
250
async def end(self) -> None:
    """Finish the logical session."""
    if self.session is None:
        raise RuntimeError("Session is not started")
    await self.session.end_session()
    self.session = None

get_driver_session()

Return the underlying Motor Session

Source code in odmantic/session.py 
233
234
235
236
237
def get_driver_session(self) -> AsyncIOMotorClientSession:
    """Return the underlying Motor Session"""
    if self.session is None:
        raise RuntimeError("session not started")
    return self.session

start() async

Start the logical Mongo session.

Source code in odmantic/session.py 
239
240
241
242
243
async def start(self) -> None:
    """Start the logical Mongo session."""
    if self.is_started:
        raise RuntimeError("Session is already started")
    self.session = await self.engine.client.start_session()

transaction()

Create a transaction in the existing session

Source code in odmantic/session.py 
264
265
266
def transaction(self) -> AIOTransaction:
    """Create a transaction in the existing session"""
    return AIOTransaction(self)

Bases: AIOSessionBase, AsyncContextManager

A transaction object to aggregate sequential operations.

Transactions can be created from the engine using the AIOEngine.transaction method or they can be created during an existing session by using AIOSession.transaction.

Example usage as a context manager: 

engine = AIOEngine(...)
async with engine.transaction() as transaction:
    john = await transaction.find(User, User.name == "John")
    john.name = "Doe"
    await transaction.save(john)
    await transaction.commit()

Example raw usage: 

engine = AIOEngine(...)
transaction = engine.transaction()
await transaction.start()
john = await transaction.find(User, User.name == "John")
john.name = "Doe"
await transaction.save(john)
await transaction.commit()

Warning

MongoDB transaction are only supported on replicated clusters: either directly a replicaSet or a sharded cluster with replication enabled.

Source code in odmantic/session.py 
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
class AIOTransaction(AIOSessionBase, AsyncContextManager):
    """A transaction object to aggregate sequential operations.

    Transactions can be created from the engine using the
    [AIOEngine.transaction][odmantic.engine.AIOEngine.transaction]
    method or they can be created during an existing session by using
    [AIOSession.transaction][odmantic.session.AIOSession.transaction].

    Example usage as a context manager:
    ```python
    engine = AIOEngine(...)
    async with engine.transaction() as transaction:
        john = await transaction.find(User, User.name == "John")
        john.name = "Doe"
        await transaction.save(john)
        await transaction.commit()
    ```

    Example raw usage:
    ```python
    engine = AIOEngine(...)
    transaction = engine.transaction()
    await transaction.start()
    john = await transaction.find(User, User.name == "John")
    john.name = "Doe"
    await transaction.save(john)
    await transaction.commit()
    ```

    Warning:
        MongoDB transaction are only supported on replicated clusters: either directly a
        replicaSet or a sharded cluster with replication enabled.
    """

    def __init__(self, context: Union[ODMEngine.AIOEngine, ODMEngine.AIOSession]):
        self._session_provided = isinstance(context, ODMEngine.AIOSession)
        if self._session_provided:
            assert isinstance(context, ODMEngine.AIOSession)
            if not context.is_started:
                raise RuntimeError("provided session is not started")
            self.session = context
            self.engine = context.engine
        else:
            assert isinstance(context, ODMEngine.AIOEngine)
            self.session = AIOSession(context)
            self.engine = context

        self._transaction_started = False
        self._transaction_context: Optional[AsyncContextManager] = None

    def get_driver_session(self) -> AsyncIOMotorClientSession:
        """Return the underlying Motor Session"""
        if not self._transaction_started:
            raise RuntimeError("transaction not started")
        return self.session.get_driver_session()

    async def start(self) -> None:
        """Initiate the transaction."""
        if self._transaction_started:
            raise RuntimeError("Transaction already started")
        if not self._session_provided:
            await self.session.start()
        assert self.session.session is not None
        self._transaction_context = (
            await self.session.session.start_transaction().__aenter__()
        )
        self._transaction_started = True

    async def commit(self) -> None:
        """Commit the changes and close the transaction."""
        if not self._transaction_started:
            raise RuntimeError("Transaction not started")
        assert self.session.session is not None
        await self.session.session.commit_transaction()
        self._transaction_started = False
        if not self._session_provided:
            await self.session.end()

    async def abort(self) -> None:
        """Discard the changes and drop the transaction"""
        if not self._transaction_started:
            raise RuntimeError("Transaction not started")
        assert self.session.session is not None
        await self.session.session.abort_transaction()
        self._transaction_started = False
        if not self._session_provided:
            await self.session.end()

    async def __aenter__(self) -> "AIOTransaction":
        await self.start()
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        assert self._transaction_context is not None
        await self._transaction_context.__aexit__(exc_type, exc, traceback)
        self._transaction_started = False

abort() async

Discard the changes and drop the transaction

Source code in odmantic/session.py 
347
348
349
350
351
352
353
354
355
async def abort(self) -> None:
    """Discard the changes and drop the transaction"""
    if not self._transaction_started:
        raise RuntimeError("Transaction not started")
    assert self.session.session is not None
    await self.session.session.abort_transaction()
    self._transaction_started = False
    if not self._session_provided:
        await self.session.end()

commit() async

Commit the changes and close the transaction.

Source code in odmantic/session.py 
337
338
339
340
341
342
343
344
345
async def commit(self) -> None:
    """Commit the changes and close the transaction."""
    if not self._transaction_started:
        raise RuntimeError("Transaction not started")
    assert self.session.session is not None
    await self.session.session.commit_transaction()
    self._transaction_started = False
    if not self._session_provided:
        await self.session.end()

get_driver_session()

Return the underlying Motor Session

Source code in odmantic/session.py 
319
320
321
322
323
def get_driver_session(self) -> AsyncIOMotorClientSession:
    """Return the underlying Motor Session"""
    if not self._transaction_started:
        raise RuntimeError("transaction not started")
    return self.session.get_driver_session()

start() async

Initiate the transaction.

Source code in odmantic/session.py 
325
326
327
328
329
330
331
332
333
334
335
async def start(self) -> None:
    """Initiate the transaction."""
    if self._transaction_started:
        raise RuntimeError("Transaction already started")
    if not self._session_provided:
        await self.session.start()
    assert self.session.session is not None
    self._transaction_context = (
        await self.session.session.start_transaction().__aenter__()
    )
    self._transaction_started = True
Source code in odmantic/session.py 
 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
class AIOSessionBase(metaclass=ABCMeta):
    engine: ODMEngine.AIOEngine

    def find(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[
            QueryExpression, Dict, bool
        ],  # bool: allow using binary operators with mypy
        sort: Optional[Any] = None,
        skip: int = 0,
        limit: Optional[int] = None,
    ) -> ODMEngine.AIOCursor[ODMEngine.ModelType]:
        """Search for Model instances matching the query filter provided

        Args:
            model: model to perform the operation on
            *queries: query filter to apply
            sort: sort expression
            skip: number of document to skip
            limit: maximum number of instance fetched

        Returns:
            [odmantic.engine.AIOCursor][] of the query

        """
        return self.engine.find(
            model,
            *queries,
            sort=sort,
            skip=skip,
            limit=limit,
            session=self.engine._get_session(self),
        )

    async def find_one(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[
            QueryExpression, Dict, bool
        ],  # bool: allow using binary operators w/o plugin
        sort: Optional[Any] = None,
    ) -> Optional[ODMEngine.ModelType]:
        """Search for a Model instance matching the query filter provided

        Args:
            model: model to perform the operation on
            *queries: query filter to apply
            sort: sort expression

        Raises:
            DocumentParsingError: unable to parse the resulting document

        Returns:
            the fetched instance if found otherwise None

        <!---
        #noqa: DAR402 DocumentParsingError
        -->
        """
        return await self.engine.find_one(
            model, *queries, sort=sort, session=self.engine._get_session(self)
        )

    async def count(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[QueryExpression, Dict, bool],
    ) -> int:
        """Get the count of documents matching a query

        Args:
            model: model to perform the operation on
            *queries: query filters to apply

        Returns:
            number of document matching the query
        """
        return await self.engine.count(
            model, *queries, session=self.engine._get_session(self)
        )

    async def save(
        self,
        instance: ODMEngine.ModelType,
    ) -> ODMEngine.ModelType:
        """Persist an instance to the database

        This method behaves as an 'upsert' operation. If a document already exists
        with the same primary key, it will be overwritten.

        All the other models referenced by this instance will be saved as well.

        Args:
            instance: instance to persist

        Returns:
            the saved instance

        NOTE:
            The save operation actually modify the instance argument in place. However,
            the instance is still returned for convenience.
        """
        return await self.engine.save(instance, session=self.engine._get_session(self))

    async def save_all(
        self,
        instances: Sequence[ODMEngine.ModelType],
    ) -> List[ODMEngine.ModelType]:
        """Persist instances to the database

        This method behaves as multiple 'upsert' operations. If one of the document
        already exists with the same primary key, it will be overwritten.

        All the other models referenced by this instance will be recursively saved as
        well.

        Args:
            instances: instances to persist

        Returns:
            the saved instances

        NOTE:
            The save_all operation actually modify the arguments in place. However, the
            instances are still returned for convenience.
        """
        return await self.engine.save_all(
            instances, session=self.engine._get_session(self)
        )

    async def delete(
        self,
        instance: ODMEngine.ModelType,
    ) -> None:
        """Delete an instance from the database

        Args:
            instance: the instance to delete

        Raises:
            DocumentNotFoundError: the instance has not been persisted to the database

        <!---
        #noqa: DAR402 DocumentNotFoundError
        #noqa: DAR201
        -->
        """
        return await self.engine.delete(
            instance, session=self.engine._get_session(self)
        )

    async def remove(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[QueryExpression, Dict, bool],
        just_one: bool = False,
    ) -> int:
        """Delete Model instances matching the query filter provided

        Args:
            model: model to perform the operation on
            *queries: query filter to apply
            just_one: limit the deletion to just one document

        Returns:
            the number of instances deleted from the database.
        """
        return await self.engine.remove(
            model, *queries, just_one=just_one, session=self.engine._get_session(self)
        )

count(model, *queries) async

Get the count of documents matching a query

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filters to apply

 ()

Returns:

Type Description
int

number of document matching the query
Source code in odmantic/session.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def count(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[QueryExpression, Dict, bool],
) -> int:
    """Get the count of documents matching a query

    Args:
        model: model to perform the operation on
        *queries: query filters to apply

    Returns:
        number of document matching the query
    """
    return await self.engine.count(
        model, *queries, session=self.engine._get_session(self)
    )

delete(instance) async

Delete an instance from the database

Parameters:

Name Type Description Default
instance ModelType

the instance to delete

 required

Raises:

Type Description
DocumentNotFoundError

the instance has not been persisted to the database
Source code in odmantic/session.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def delete(
    self,
    instance: ODMEngine.ModelType,
) -> None:
    """Delete an instance from the database

    Args:
        instance: the instance to delete

    Raises:
        DocumentNotFoundError: the instance has not been persisted to the database

    <!---
    #noqa: DAR402 DocumentNotFoundError
    #noqa: DAR201
    -->
    """
    return await self.engine.delete(
        instance, session=self.engine._get_session(self)
    )

find(model, *queries, sort=None, skip=0, limit=None)

Search for Model instances matching the query filter provided

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filter to apply

 ()
sort Optional[Any]

sort expression

 None
skip int

number of document to skip

 0
limit Optional[int]

maximum number of instance fetched

 None

Returns:

Type Description
AIOCursor[ModelType]

odmantic.engine.AIOCursor of the query
Source code in odmantic/session.py 
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
def find(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[
        QueryExpression, Dict, bool
    ],  # bool: allow using binary operators with mypy
    sort: Optional[Any] = None,
    skip: int = 0,
    limit: Optional[int] = None,
) -> ODMEngine.AIOCursor[ODMEngine.ModelType]:
    """Search for Model instances matching the query filter provided

    Args:
        model: model to perform the operation on
        *queries: query filter to apply
        sort: sort expression
        skip: number of document to skip
        limit: maximum number of instance fetched

    Returns:
        [odmantic.engine.AIOCursor][] of the query

    """
    return self.engine.find(
        model,
        *queries,
        sort=sort,
        skip=skip,
        limit=limit,
        session=self.engine._get_session(self),
    )

find_one(model, *queries, sort=None) async

Search for a Model instance matching the query filter provided

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filter to apply

 ()
sort Optional[Any]

sort expression

 None

Raises:

Type Description
DocumentParsingError

unable to parse the resulting document

Returns:

Type Description
Optional[ModelType]

the fetched instance if found otherwise None
Source code in odmantic/session.py 
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
async def find_one(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[
        QueryExpression, Dict, bool
    ],  # bool: allow using binary operators w/o plugin
    sort: Optional[Any] = None,
) -> Optional[ODMEngine.ModelType]:
    """Search for a Model instance matching the query filter provided

    Args:
        model: model to perform the operation on
        *queries: query filter to apply
        sort: sort expression

    Raises:
        DocumentParsingError: unable to parse the resulting document

    Returns:
        the fetched instance if found otherwise None

    <!---
    #noqa: DAR402 DocumentParsingError
    -->
    """
    return await self.engine.find_one(
        model, *queries, sort=sort, session=self.engine._get_session(self)
    )

remove(model, *queries, just_one=False) async

Delete Model instances matching the query filter provided

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filter to apply

 ()
just_one bool

limit the deletion to just one document

 False

Returns:

Type Description
int

the number of instances deleted from the database.
Source code in odmantic/session.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
async def remove(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[QueryExpression, Dict, bool],
    just_one: bool = False,
) -> int:
    """Delete Model instances matching the query filter provided

    Args:
        model: model to perform the operation on
        *queries: query filter to apply
        just_one: limit the deletion to just one document

    Returns:
        the number of instances deleted from the database.
    """
    return await self.engine.remove(
        model, *queries, just_one=just_one, session=self.engine._get_session(self)
    )

save(instance) async

Persist an instance to the database

This method behaves as an 'upsert' operation. If a document already exists with the same primary key, it will be overwritten.

All the other models referenced by this instance will be saved as well.

Parameters:

Name Type Description Default
instance ModelType

instance to persist

 required

Returns:

Type Description
ModelType

the saved instance
NOTE

The save operation actually modify the instance argument in place. However, the instance is still returned for convenience.

Source code in odmantic/session.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def save(
    self,
    instance: ODMEngine.ModelType,
) -> ODMEngine.ModelType:
    """Persist an instance to the database

    This method behaves as an 'upsert' operation. If a document already exists
    with the same primary key, it will be overwritten.

    All the other models referenced by this instance will be saved as well.

    Args:
        instance: instance to persist

    Returns:
        the saved instance

    NOTE:
        The save operation actually modify the instance argument in place. However,
        the instance is still returned for convenience.
    """
    return await self.engine.save(instance, session=self.engine._get_session(self))

save_all(instances) async

Persist instances to the database

This method behaves as multiple 'upsert' operations. If one of the document already exists with the same primary key, it will be overwritten.

All the other models referenced by this instance will be recursively saved as well.

Parameters:

Name Type Description Default
instances Sequence[ModelType]

instances to persist

 required

Returns:

Type Description
List[ModelType]

the saved instances
NOTE

The save_all operation actually modify the arguments in place. However, the instances are still returned for convenience.

Source code in odmantic/session.py 
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
async def save_all(
    self,
    instances: Sequence[ODMEngine.ModelType],
) -> List[ODMEngine.ModelType]:
    """Persist instances to the database

    This method behaves as multiple 'upsert' operations. If one of the document
    already exists with the same primary key, it will be overwritten.

    All the other models referenced by this instance will be recursively saved as
    well.

    Args:
        instances: instances to persist

    Returns:
        the saved instances

    NOTE:
        The save_all operation actually modify the arguments in place. However, the
        instances are still returned for convenience.
    """
    return await self.engine.save_all(
        instances, session=self.engine._get_session(self)
    )

Bases: SyncSessionBase, ContextManager

A session object for ordering sequential operations.

Sessions can be created from the engine directly by using the SyncEngine.session method.

Example usage as a context manager: 

engine = SyncEngine(...)
with engine.session() as session:
    john = session.find(User, User.name == "John")
    john.name = "Doe"
    session.save(john)

Example raw usage: 

engine = SyncEngine(...)
session = engine.session()
session.start()
john = session.find(User, User.name == "John")
john.name = "Doe"
session.save(john)
session.end()

Source code in odmantic/session.py 
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
class SyncSession(SyncSessionBase, ContextManager):
    """A session object for ordering sequential operations.

    Sessions can be created from the engine directly by using the
    [SyncEngine.session][odmantic.engine.SyncEngine.session] method.

    Example usage as a context manager:
    ```python
    engine = SyncEngine(...)
    with engine.session() as session:
        john = session.find(User, User.name == "John")
        john.name = "Doe"
        session.save(john)
    ```

    Example raw usage:
    ```python
    engine = SyncEngine(...)
    session = engine.session()
    session.start()
    john = session.find(User, User.name == "John")
    john.name = "Doe"
    session.save(john)
    session.end()
    ```
    """

    def __init__(self, engine: ODMEngine.SyncEngine):
        self.engine = engine
        self.session: Optional[ClientSession] = None

    @property
    def is_started(self) -> bool:
        return self.session is not None

    def get_driver_session(self) -> ClientSession:
        """Return the underlying PyMongo Session"""
        if self.session is None:
            raise RuntimeError("session not started")
        return self.session

    def start(self) -> None:
        """Start the logical session."""
        if self.is_started:
            raise RuntimeError("Session is already started")
        self.session = self.engine.client.start_session()

    def end(self) -> None:
        """Finish the logical session."""
        if self.session is None:
            raise RuntimeError("Session is not started")
        self.session.end_session()
        self.session = None

    def __enter__(self) -> "SyncSession":
        self.start()
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        self.end()

    def transaction(self) -> SyncTransaction:
        """Create a transaction in the existing session"""
        return SyncTransaction(self)

end()

Finish the logical session.

Source code in odmantic/session.py 
589
590
591
592
593
594
def end(self) -> None:
    """Finish the logical session."""
    if self.session is None:
        raise RuntimeError("Session is not started")
    self.session.end_session()
    self.session = None

get_driver_session()

Return the underlying PyMongo Session

Source code in odmantic/session.py 
577
578
579
580
581
def get_driver_session(self) -> ClientSession:
    """Return the underlying PyMongo Session"""
    if self.session is None:
        raise RuntimeError("session not started")
    return self.session

start()

Start the logical session.

Source code in odmantic/session.py 
583
584
585
586
587
def start(self) -> None:
    """Start the logical session."""
    if self.is_started:
        raise RuntimeError("Session is already started")
    self.session = self.engine.client.start_session()

transaction()

Create a transaction in the existing session

Source code in odmantic/session.py 
608
609
610
def transaction(self) -> SyncTransaction:
    """Create a transaction in the existing session"""
    return SyncTransaction(self)

Bases: SyncSessionBase, ContextManager

A transaction object to aggregate sequential operations.

Transactions can be created from the engine using the SyncEngine.transaction method or they can be created during an existing session by using SyncSession.transaction.

Example usage as a context manager: 

engine = SyncEngine(...)
with engine.transaction() as transaction:
    john = transaction.find(User, User.name == "John")
    john.name = "Doe"
    transaction.save(john)
    transaction.commit()

Example raw usage: 

engine = SyncEngine(...)
transaction = engine.transaction()
transaction.start()
john = transaction.find(User, User.name == "John")
john.name = "Doe"
transaction.save(john)
transaction.commit()

Warning

MongoDB transaction are only supported on replicated clusters: either directly a replicaSet or a sharded cluster with replication enabled.

Source code in odmantic/session.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
class SyncTransaction(SyncSessionBase, ContextManager):
    """A transaction object to aggregate sequential operations.

    Transactions can be created from the engine using the
    [SyncEngine.transaction][odmantic.engine.SyncEngine.transaction]
    method or they can be created during an existing session by using
    [SyncSession.transaction][odmantic.session.SyncSession.transaction].

    Example usage as a context manager:
    ```python
    engine = SyncEngine(...)
    with engine.transaction() as transaction:
        john = transaction.find(User, User.name == "John")
        john.name = "Doe"
        transaction.save(john)
        transaction.commit()
    ```

    Example raw usage:
    ```python
    engine = SyncEngine(...)
    transaction = engine.transaction()
    transaction.start()
    john = transaction.find(User, User.name == "John")
    john.name = "Doe"
    transaction.save(john)
    transaction.commit()
    ```

    Warning:
        MongoDB transaction are only supported on replicated clusters: either directly a
        replicaSet or a sharded cluster with replication enabled.
    """

    def __init__(self, context: Union[ODMEngine.SyncEngine, ODMEngine.SyncSession]):
        self._session_provided = isinstance(context, ODMEngine.SyncSession)
        if self._session_provided:
            assert isinstance(context, ODMEngine.SyncSession)
            if not context.is_started:
                raise RuntimeError("provided session is not started")
            self.session = context
            self.engine = context.engine
        else:
            assert isinstance(context, ODMEngine.SyncEngine)
            self.session = SyncSession(context)
            self.engine = context

        self._transaction_started = False
        self._transaction_context: Optional[ContextManager] = None

    def get_driver_session(self) -> ClientSession:
        """Return the underlying PyMongo Session"""
        if not self._transaction_started:
            raise RuntimeError("transaction not started")
        return self.session.get_driver_session()

    def start(self) -> None:
        """Initiate the transaction."""
        if self._transaction_started:
            raise RuntimeError("Transaction already started")
        if not self._session_provided:
            self.session.start()
        assert self.session.session is not None
        self._transaction_context = self.session.session.start_transaction().__enter__()
        self._transaction_started = True

    def commit(self) -> None:
        """Commit the changes and close the transaction."""
        if not self._transaction_started:
            raise RuntimeError("Transaction not started")
        assert self.session.session is not None
        self.session.session.commit_transaction()
        self._transaction_started = False
        if not self._session_provided:
            self.session.end()

    def abort(self) -> None:
        """Discard the changes and drop the transaction."""
        if not self._transaction_started:
            raise RuntimeError("Transaction not started")
        assert self.session.session is not None
        self.session.session.abort_transaction()
        self._transaction_started = False
        if not self._session_provided:
            self.session.end()

    def __enter__(self) -> "SyncTransaction":
        self.start()
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        assert self._transaction_context is not None
        self._transaction_context.__exit__(exc_type, exc, traceback)
        self._transaction_started = False

abort()

Discard the changes and drop the transaction.

Source code in odmantic/session.py 
689
690
691
692
693
694
695
696
697
def abort(self) -> None:
    """Discard the changes and drop the transaction."""
    if not self._transaction_started:
        raise RuntimeError("Transaction not started")
    assert self.session.session is not None
    self.session.session.abort_transaction()
    self._transaction_started = False
    if not self._session_provided:
        self.session.end()

commit()

Commit the changes and close the transaction.

Source code in odmantic/session.py 
679
680
681
682
683
684
685
686
687
def commit(self) -> None:
    """Commit the changes and close the transaction."""
    if not self._transaction_started:
        raise RuntimeError("Transaction not started")
    assert self.session.session is not None
    self.session.session.commit_transaction()
    self._transaction_started = False
    if not self._session_provided:
        self.session.end()

get_driver_session()

Return the underlying PyMongo Session

Source code in odmantic/session.py 
663
664
665
666
667
def get_driver_session(self) -> ClientSession:
    """Return the underlying PyMongo Session"""
    if not self._transaction_started:
        raise RuntimeError("transaction not started")
    return self.session.get_driver_session()

start()

Initiate the transaction.

Source code in odmantic/session.py 
669
670
671
672
673
674
675
676
677
def start(self) -> None:
    """Initiate the transaction."""
    if self._transaction_started:
        raise RuntimeError("Transaction already started")
    if not self._session_provided:
        self.session.start()
    assert self.session.session is not None
    self._transaction_context = self.session.session.start_transaction().__enter__()
    self._transaction_started = True
Source code in odmantic/session.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
class SyncSessionBase(metaclass=ABCMeta):
    engine: ODMEngine.SyncEngine

    def find(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[
            QueryExpression, Dict, bool
        ],  # bool: allow using binary operators with mypy
        sort: Optional[Any] = None,
        skip: int = 0,
        limit: Optional[int] = None,
    ) -> ODMEngine.SyncCursor[ODMEngine.ModelType]:
        """Search for Model instances matching the query filter provided

        Args:
            model: model to perform the operation on
            *queries: query filter to apply
            sort: sort expression
            skip: number of document to skip
            limit: maximum number of instance fetched

        Returns:
            [odmantic.engine.SyncCursor][] of the query

        """
        return self.engine.find(
            model,
            *queries,
            sort=sort,
            skip=skip,
            limit=limit,
            session=self.engine._get_session(self),
        )

    def find_one(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[
            QueryExpression, Dict, bool
        ],  # bool: allow using binary operators w/o plugin
        sort: Optional[Any] = None,
    ) -> Optional[ODMEngine.ModelType]:
        """Search for a Model instance matching the query filter provided

        Args:
            model: model to perform the operation on
            *queries: query filter to apply
            sort: sort expression

        Raises:
            DocumentParsingError: unable to parse the resulting document

        Returns:
            the fetched instance if found otherwise None

        <!---
        #noqa: DAR402 DocumentParsingError
        -->
        """
        return self.engine.find_one(
            model, *queries, sort=sort, session=self.engine._get_session(self)
        )

    def count(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[QueryExpression, Dict, bool],
    ) -> int:
        """Get the count of documents matching a query

        Args:
            model: model to perform the operation on
            *queries: query filters to apply

        Returns:
            number of document matching the query
        """
        return self.engine.count(
            model, *queries, session=self.engine._get_session(self)
        )

    def save(
        self,
        instance: ODMEngine.ModelType,
    ) -> ODMEngine.ModelType:
        """Persist an instance to the database

        This method behaves as an 'upsert' operation. If a document already exists
        with the same primary key, it will be overwritten.

        All the other models referenced by this instance will be saved as well.

        Args:
            instance: instance to persist

        Returns:
            the saved instance

        NOTE:
            The save operation actually modify the instance argument in place. However,
            the instance is still returned for convenience.
        """
        return self.engine.save(instance, session=self.engine._get_session(self))

    def save_all(
        self,
        instances: Sequence[ODMEngine.ModelType],
    ) -> List[ODMEngine.ModelType]:
        """Persist instances to the database

        This method behaves as multiple 'upsert' operations. If one of the document
        already exists with the same primary key, it will be overwritten.

        All the other models referenced by this instance will be recursively saved as
        well.

        Args:
            instances: instances to persist

        Returns:
            the saved instances

        NOTE:
            The save_all operation actually modify the arguments in place. However, the
            instances are still returned for convenience.
        """
        return self.engine.save_all(instances, session=self.engine._get_session(self))

    def delete(
        self,
        instance: ODMEngine.ModelType,
    ) -> None:
        """Delete an instance from the database

        Args:
            instance: the instance to delete

        Raises:
            DocumentNotFoundError: the instance has not been persisted to the database

        <!---
        #noqa: DAR402 DocumentNotFoundError
        #noqa: DAR201
        -->
        """
        return self.engine.delete(instance, session=self.engine._get_session(self))

    def remove(
        self,
        model: Type[ODMEngine.ModelType],
        *queries: Union[QueryExpression, Dict, bool],
        just_one: bool = False,
    ) -> int:
        """Delete Model instances matching the query filter provided

        Args:
            model: model to perform the operation on
            *queries: query filter to apply
            just_one: limit the deletion to just one document

        Returns:
            the number of instances deleted from the database.

        """
        return self.engine.remove(
            model, *queries, just_one=just_one, session=self.engine._get_session(self)
        )

count(model, *queries)

Get the count of documents matching a query

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filters to apply

 ()

Returns:

Type Description
int

number of document matching the query
Source code in odmantic/session.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def count(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[QueryExpression, Dict, bool],
) -> int:
    """Get the count of documents matching a query

    Args:
        model: model to perform the operation on
        *queries: query filters to apply

    Returns:
        number of document matching the query
    """
    return self.engine.count(
        model, *queries, session=self.engine._get_session(self)
    )

delete(instance)

Delete an instance from the database

Parameters:

Name Type Description Default
instance ModelType

the instance to delete

 required

Raises:

Type Description
DocumentNotFoundError

the instance has not been persisted to the database
Source code in odmantic/session.py 
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
def delete(
    self,
    instance: ODMEngine.ModelType,
) -> None:
    """Delete an instance from the database

    Args:
        instance: the instance to delete

    Raises:
        DocumentNotFoundError: the instance has not been persisted to the database

    <!---
    #noqa: DAR402 DocumentNotFoundError
    #noqa: DAR201
    -->
    """
    return self.engine.delete(instance, session=self.engine._get_session(self))

find(model, *queries, sort=None, skip=0, limit=None)

Search for Model instances matching the query filter provided

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filter to apply

 ()
sort Optional[Any]

sort expression

 None
skip int

number of document to skip

 0
limit Optional[int]

maximum number of instance fetched

 None

Returns:

Type Description
SyncCursor[ModelType]

odmantic.engine.SyncCursor of the query
Source code in odmantic/session.py 
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def find(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[
        QueryExpression, Dict, bool
    ],  # bool: allow using binary operators with mypy
    sort: Optional[Any] = None,
    skip: int = 0,
    limit: Optional[int] = None,
) -> ODMEngine.SyncCursor[ODMEngine.ModelType]:
    """Search for Model instances matching the query filter provided

    Args:
        model: model to perform the operation on
        *queries: query filter to apply
        sort: sort expression
        skip: number of document to skip
        limit: maximum number of instance fetched

    Returns:
        [odmantic.engine.SyncCursor][] of the query

    """
    return self.engine.find(
        model,
        *queries,
        sort=sort,
        skip=skip,
        limit=limit,
        session=self.engine._get_session(self),
    )

find_one(model, *queries, sort=None)

Search for a Model instance matching the query filter provided

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filter to apply

 ()
sort Optional[Any]

sort expression

 None

Raises:

Type Description
DocumentParsingError

unable to parse the resulting document

Returns:

Type Description
Optional[ModelType]

the fetched instance if found otherwise None
Source code in odmantic/session.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def find_one(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[
        QueryExpression, Dict, bool
    ],  # bool: allow using binary operators w/o plugin
    sort: Optional[Any] = None,
) -> Optional[ODMEngine.ModelType]:
    """Search for a Model instance matching the query filter provided

    Args:
        model: model to perform the operation on
        *queries: query filter to apply
        sort: sort expression

    Raises:
        DocumentParsingError: unable to parse the resulting document

    Returns:
        the fetched instance if found otherwise None

    <!---
    #noqa: DAR402 DocumentParsingError
    -->
    """
    return self.engine.find_one(
        model, *queries, sort=sort, session=self.engine._get_session(self)
    )

remove(model, *queries, just_one=False)

Delete Model instances matching the query filter provided

Parameters:

Name Type Description Default
model Type[ModelType]

model to perform the operation on

 required
*queries Union[QueryExpression, Dict, bool]

query filter to apply

 ()
just_one bool

limit the deletion to just one document

 False

Returns:

Type Description
int

the number of instances deleted from the database.
Source code in odmantic/session.py 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
def remove(
    self,
    model: Type[ODMEngine.ModelType],
    *queries: Union[QueryExpression, Dict, bool],
    just_one: bool = False,
) -> int:
    """Delete Model instances matching the query filter provided

    Args:
        model: model to perform the operation on
        *queries: query filter to apply
        just_one: limit the deletion to just one document

    Returns:
        the number of instances deleted from the database.

    """
    return self.engine.remove(
        model, *queries, just_one=just_one, session=self.engine._get_session(self)
    )

save(instance)

Persist an instance to the database

This method behaves as an 'upsert' operation. If a document already exists with the same primary key, it will be overwritten.

All the other models referenced by this instance will be saved as well.

Parameters:

Name Type Description Default
instance ModelType

instance to persist

 required

Returns:

Type Description
ModelType

the saved instance
NOTE

The save operation actually modify the instance argument in place. However, the instance is still returned for convenience.

Source code in odmantic/session.py 
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
def save(
    self,
    instance: ODMEngine.ModelType,
) -> ODMEngine.ModelType:
    """Persist an instance to the database

    This method behaves as an 'upsert' operation. If a document already exists
    with the same primary key, it will be overwritten.

    All the other models referenced by this instance will be saved as well.

    Args:
        instance: instance to persist

    Returns:
        the saved instance

    NOTE:
        The save operation actually modify the instance argument in place. However,
        the instance is still returned for convenience.
    """
    return self.engine.save(instance, session=self.engine._get_session(self))

save_all(instances)

Persist instances to the database

This method behaves as multiple 'upsert' operations. If one of the document already exists with the same primary key, it will be overwritten.

All the other models referenced by this instance will be recursively saved as well.

Parameters:

Name Type Description Default
instances Sequence[ModelType]

instances to persist

 required

Returns:

Type Description
List[ModelType]

the saved instances
NOTE

The save_all operation actually modify the arguments in place. However, the instances are still returned for convenience.

Source code in odmantic/session.py 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
def save_all(
    self,
    instances: Sequence[ODMEngine.ModelType],
) -> List[ODMEngine.ModelType]:
    """Persist instances to the database

    This method behaves as multiple 'upsert' operations. If one of the document
    already exists with the same primary key, it will be overwritten.

    All the other models referenced by this instance will be recursively saved as
    well.

    Args:
        instances: instances to persist

    Returns:
        the saved instances

    NOTE:
        The save_all operation actually modify the arguments in place. However, the
        instances are still returned for convenience.
    """
    return self.engine.save_all(instances, session=self.engine._get_session(self))