0
# Sessions
1
2
ODMantic provides session and transaction support for both async and sync operations, enabling atomic operations, consistency guarantees, and proper resource management.
3
4
## Capabilities
5
6
### Async Session Management
7
8
Session context managers for async operations with proper resource cleanup.
9
10
```python { .api }
11
class AIOSession:
12
"""Async session context manager for MongoDB operations."""
13
14
async def __aenter__(self):
15
"""Enter async context manager."""
16
17
async def __aexit__(self, exc_type, exc_val, exc_tb):
18
"""Exit async context manager with cleanup."""
19
20
def get_driver_session(self):
21
"""
22
Get underlying motor client session.
23
24
Returns:
25
AsyncIOMotorClientSession: Motor session object
26
"""
27
28
class AIOTransaction(AIOSession):
29
"""Async transaction context manager for atomic operations."""
30
31
async def __aenter__(self):
32
"""Enter async transaction context."""
33
34
async def __aexit__(self, exc_type, exc_val, exc_tb):
35
"""Exit async transaction with commit/rollback."""
36
37
def get_driver_session(self):
38
"""
39
Get underlying motor client session.
40
41
Returns:
42
AsyncIOMotorClientSession: Motor session object
43
"""
44
```
45
46
### Sync Session Management
47
48
Session context managers for sync operations with proper resource cleanup.
49
50
```python { .api }
51
class SyncSession:
52
"""Sync session context manager for MongoDB operations."""
53
54
def __enter__(self):
55
"""Enter sync context manager."""
56
57
def __exit__(self, exc_type, exc_val, exc_tb):
58
"""Exit sync context manager with cleanup."""
59
60
def get_driver_session(self):
61
"""
62
Get underlying pymongo client session.
63
64
Returns:
65
ClientSession: Pymongo session object
66
"""
67
68
class SyncTransaction(SyncSession):
69
"""Sync transaction context manager for atomic operations."""
70
71
def __enter__(self):
72
"""Enter sync transaction context."""
73
74
def __exit__(self, exc_type, exc_val, exc_tb):
75
"""Exit sync transaction with commit/rollback."""
76
77
def get_driver_session(self):
78
"""
79
Get underlying pymongo client session.
80
81
Returns:
82
ClientSession: Pymongo session object
83
"""
84
```
85
86
### Engine Session Methods
87
88
Methods on engines to create session and transaction contexts.
89
90
```python { .api }
91
# AIOEngine methods
92
def session(self):
93
"""
94
Create async session context manager.
95
96
Returns:
97
AIOSession: Async session context manager
98
"""
99
100
def transaction(self):
101
"""
102
Create async transaction context manager.
103
104
Returns:
105
AIOTransaction: Async transaction context manager
106
"""
107
108
# SyncEngine methods
109
def session(self):
110
"""
111
Create sync session context manager.
112
113
Returns:
114
SyncSession: Sync session context manager
115
"""
116
117
def transaction(self):
118
"""
119
Create sync transaction context manager.
120
121
Returns:
122
SyncTransaction: Sync transaction context manager
123
"""
124
```
125
126
## Usage Examples
127
128
### Basic Session Usage (Async)
129
130
```python
131
from odmantic import AIOEngine, Model
132
from motor.motor_asyncio import AsyncIOMotorClient
133
134
class User(Model):
135
name: str
136
email: str
137
138
class Order(Model):
139
user_id: ObjectId
140
total: float
141
142
async def session_example():
143
client = AsyncIOMotorClient("mongodb://localhost:27017")
144
engine = AIOEngine(client, database="mydb")
145
146
# Basic session usage
147
async with engine.session() as session:
148
# All operations in this block use the same session
149
user = User(name="John", email="john@example.com")
150
await engine.save(user, session=session)
151
152
# Find operations also use the session
153
users = await engine.find(User, session=session)
154
155
# Session ensures consistent reads
156
user_count = await engine.count(User, session=session)
157
print(f"Found {user_count} users")
158
```
159
160
### Basic Session Usage (Sync)
161
162
```python
163
from odmantic import SyncEngine, Model
164
from pymongo import MongoClient
165
166
def sync_session_example():
167
client = MongoClient("mongodb://localhost:27017")
168
engine = SyncEngine(client, database="mydb")
169
170
# Basic session usage
171
with engine.session() as session:
172
# All operations use the same session
173
user = User(name="Jane", email="jane@example.com")
174
engine.save(user, session=session)
175
176
# Consistent reads within session
177
users = list(engine.find(User, session=session))
178
user_count = engine.count(User, session=session)
179
print(f"Found {user_count} users")
180
```
181
182
### Transaction Usage (Async)
183
184
```python
185
async def transaction_example():
186
client = AsyncIOMotorClient("mongodb://localhost:27017")
187
engine = AIOEngine(client, database="mydb")
188
189
try:
190
async with engine.transaction() as transaction:
191
# Create a user
192
user = User(name="Alice", email="alice@example.com")
193
await engine.save(user, session=transaction)
194
195
# Create an order for the user
196
order = Order(user_id=user.id, total=99.99)
197
await engine.save(order, session=transaction)
198
199
# Update user with additional info
200
user.model_update(total_orders=1)
201
await engine.save(user, session=transaction)
202
203
# If we reach here, transaction commits automatically
204
print("Transaction completed successfully")
205
206
except Exception as e:
207
# Transaction automatically rolls back on exception
208
print(f"Transaction failed: {e}")
209
```
210
211
### Transaction Usage (Sync)
212
213
```python
214
def sync_transaction_example():
215
client = MongoClient("mongodb://localhost:27017")
216
engine = SyncEngine(client, database="mydb")
217
218
try:
219
with engine.transaction() as transaction:
220
# Create a user
221
user = User(name="Bob", email="bob@example.com")
222
engine.save(user, session=transaction)
223
224
# Create an order for the user
225
order = Order(user_id=user.id, total=149.99)
226
engine.save(order, session=transaction)
227
228
# Transaction commits automatically if no exception
229
print("Transaction completed successfully")
230
231
except Exception as e:
232
# Transaction automatically rolls back on exception
233
print(f"Transaction failed: {e}")
234
```
235
236
### Multi-Collection Transactions
237
238
```python
239
class Account(Model):
240
user_id: ObjectId
241
balance: float
242
243
class Transaction(Model):
244
from_account: ObjectId
245
to_account: ObjectId
246
amount: float
247
timestamp: datetime
248
249
async def money_transfer(engine: AIOEngine, from_id: ObjectId, to_id: ObjectId, amount: float):
250
"""Transfer money between accounts atomically."""
251
252
async with engine.transaction() as txn:
253
# Get source account
254
from_account = await engine.find_one(Account, Account.user_id == from_id, session=txn)
255
if not from_account or from_account.balance < amount:
256
raise ValueError("Insufficient funds")
257
258
# Get destination account
259
to_account = await engine.find_one(Account, Account.user_id == to_id, session=txn)
260
if not to_account:
261
raise ValueError("Destination account not found")
262
263
# Update balances
264
from_account.model_update(balance=from_account.balance - amount)
265
to_account.model_update(balance=to_account.balance + amount)
266
267
# Save updated accounts
268
await engine.save(from_account, session=txn)
269
await engine.save(to_account, session=txn)
270
271
# Record transaction
272
transaction = Transaction(
273
from_account=from_id,
274
to_account=to_id,
275
amount=amount,
276
timestamp=datetime.utcnow()
277
)
278
await engine.save(transaction, session=txn)
279
280
print(f"Transferred ${amount} from {from_id} to {to_id}")
281
```
282
283
### Session with Error Handling
284
285
```python
286
async def robust_session_example():
287
client = AsyncIOMotorClient("mongodb://localhost:27017")
288
engine = AIOEngine(client, database="mydb")
289
290
session = None
291
try:
292
async with engine.session() as session:
293
# Perform multiple operations
294
users = []
295
for i in range(10):
296
user = User(name=f"User {i}", email=f"user{i}@example.com")
297
await engine.save(user, session=session)
298
users.append(user)
299
300
# Verify all users were created
301
total_users = await engine.count(User, session=session)
302
print(f"Created {len(users)} users, total in DB: {total_users}")
303
304
except Exception as e:
305
print(f"Session failed: {e}")
306
# Session cleanup happens automatically
307
```
308
309
### Nested Session Handling
310
311
```python
312
async def nested_operations(engine: AIOEngine, session=None):
313
"""Function that can work with or without a session."""
314
315
# Use provided session or create a new one
316
if session:
317
# Use existing session
318
user = User(name="Nested User", email="nested@example.com")
319
await engine.save(user, session=session)
320
return user
321
else:
322
# Create new session
323
async with engine.session() as new_session:
324
user = User(name="New Session User", email="newsession@example.com")
325
await engine.save(user, session=new_session)
326
return user
327
328
async def caller_example():
329
client = AsyncIOMotorClient("mongodb://localhost:27017")
330
engine = AIOEngine(client, database="mydb")
331
332
# Call without session - function creates its own
333
user1 = await nested_operations(engine)
334
335
# Call with session - function uses provided session
336
async with engine.session() as session:
337
user2 = await nested_operations(engine, session=session)
338
user3 = await nested_operations(engine, session=session)
339
340
# All operations in this session are consistent
341
count = await engine.count(User, session=session)
342
print(f"Users in session: {count}")
343
```
344
345
### Transaction Rollback Examples
346
347
```python
348
async def transaction_rollback_example():
349
client = AsyncIOMotorClient("mongodb://localhost:27017")
350
engine = AIOEngine(client, database="mydb")
351
352
# Example 1: Explicit rollback
353
try:
354
async with engine.transaction() as txn:
355
user = User(name="Test User", email="test@example.com")
356
await engine.save(user, session=txn)
357
358
# Some condition that requires rollback
359
if user.name == "Test User":
360
raise ValueError("Test users not allowed")
361
362
except ValueError as e:
363
print(f"Transaction rolled back: {e}")
364
365
# Example 2: Automatic rollback on any exception
366
try:
367
async with engine.transaction() as txn:
368
# This will all be rolled back
369
for i in range(5):
370
user = User(name=f"User {i}", email=f"user{i}@example.com")
371
await engine.save(user, session=txn)
372
373
# Simulate an error
374
raise RuntimeError("Something went wrong")
375
376
except RuntimeError as e:
377
print(f"All operations rolled back: {e}")
378
379
# Verify no users were created
380
count = await engine.count(User)
381
print(f"Total users after rollback: {count}")
382
```
383
384
### Session Configuration
385
386
```python
387
async def session_configuration_examples():
388
client = AsyncIOMotorClient("mongodb://localhost:27017")
389
engine = AIOEngine(client, database="mydb")
390
391
# Sessions automatically use appropriate read/write concerns
392
# based on MongoDB configuration and replica set setup
393
394
async with engine.session() as session:
395
# Session inherits client's read preference, write concern, etc.
396
users = await engine.find(User, session=session)
397
398
async with engine.transaction() as txn:
399
# Transactions automatically use appropriate concerns for ACID compliance
400
user = User(name="Transactional User", email="txn@example.com")
401
await engine.save(user, session=txn)
402
```