0
# Beanie
1

2
An asynchronous Python object-document mapper (ODM) for MongoDB that provides a modern, Pydantic-based approach to database interactions. Built on top of PyMongo's async client and Pydantic models, Beanie offers a clean, type-safe API for document operations including CRUD operations, aggregation pipelines, and complex queries with pythonic syntax.
3

4
## Package Information
5

6
- **Package Name**: beanie
7
- **Language**: Python
8
- **Installation**: `pip install beanie`
9

10
## Core Imports
11

12
```python
13
import beanie
14
```
15

16
Common patterns for working with documents:
17

18
```python
19
from beanie import Document, init_beanie
20
```
21

22
Import specific components:
23

24
```python
25
from beanie import (
26
    Document, DocumentWithSoftDelete, View, UnionDoc,
27
    init_beanie, PydanticObjectId, Indexed, Link, BackLink,
28
    before_event, after_event, Insert, Save, Update, Delete,
29
    BulkWriter, iterative_migration, free_fall_migration
30
)
31
```
32

33
## Basic Usage
34

35
```python
36
import asyncio
37
from beanie import Document, init_beanie
38
from motor.motor_asyncio import AsyncIOMotorClient
39
from pydantic import BaseModel
40

41
# Define a document model
42
class User(Document):
43
    name: str
44
    email: str
45
    age: int = 0
46
    
47
    class Settings:
48
        collection = "users"
49

50
# Initialize Beanie
51
async def init_db():
52
    client = AsyncIOMotorClient("mongodb://localhost:27017")
53
    database = client.myapp
54
    
55
    await init_beanie(database=database, document_models=[User])
56

57
# Basic CRUD operations
58
async def main():
59
    await init_db()
60
    
61
    # Create a new user
62
    user = User(name="John Doe", email="john@example.com", age=30)
63
    await user.insert()
64
    
65
    # Find users
66
    users = await User.find_all().to_list()
67
    user = await User.find_one(User.email == "john@example.com")
68
    
69
    # Update user
70
    await user.update({"$set": {"age": 31}})
71
    
72
    # Delete user
73
    await user.delete()
74

75
if __name__ == "__main__":
76
    asyncio.run(main())
77
```
78

79
## Architecture
80

81
Beanie follows a layered architecture built on MongoDB and Pydantic:
82

83
- **Document Models**: Core ODM classes extending Pydantic models with MongoDB integration
84
- **Field System**: Type-safe field definitions with validation, indexing, and relationship support
85
- **Query Interface**: Fluent query API with aggregation pipeline support
86
- **Event System**: Pre/post hooks for document lifecycle events
87
- **Bulk Operations**: Efficient batch processing with transaction support
88
- **Migration System**: Schema migration tools for data evolution
89

90
This design provides full MongoDB feature access while maintaining Python type safety and async performance, making it ideal for modern FastAPI applications and microservices.
91

92
## Capabilities
93

94
### Document Models
95

96
Core document mapping classes for different MongoDB collection patterns, providing CRUD operations, query interfaces, and lifecycle management.
97

98
```python { .api }
99
class Document(BaseModel):
100
    id: Optional[PydanticObjectId] = None
101
    revision_id: Optional[UUID] = None
102
    
103
    # Core CRUD operations
104
    async def insert(self, session: Optional[AsyncClientSession] = None, **kwargs) -> "Document": ...
105
    async def save(self, session: Optional[AsyncClientSession] = None, **kwargs) -> "Document": ...
106
    async def save_changes(self, session: Optional[AsyncClientSession] = None, **kwargs) -> Optional["Document"]: ...
107
    async def update(self, *args, session: Optional[AsyncClientSession] = None, **kwargs) -> None: ...
108
    async def replace(self, session: Optional[AsyncClientSession] = None, **kwargs) -> "Document": ...
109
    async def delete(self, session: Optional[AsyncClientSession] = None, **kwargs) -> None: ...
110
    
111
    # Class methods
112
    @classmethod
113
    async def get(cls, document_id: Any, **kwargs) -> Optional["Document"]: ...
114
    @classmethod
115
    async def find_one(cls, filter_query: Optional[Dict] = None, **kwargs) -> Optional["Document"]: ...
116
    @classmethod
117
    def find(cls, filter_query: Optional[Dict] = None, **kwargs) -> "FindInterface": ...
118
    @classmethod
119
    def aggregate(cls, pipeline: List[Dict], **kwargs) -> "AggregateInterface": ...
120
    @classmethod
121
    async def insert_many(cls, documents: Iterable["Document"], **kwargs) -> InsertManyResult: ...
122
    @classmethod
123
    async def delete_all(cls, **kwargs) -> DeleteResult: ...
124
    @classmethod
125
    async def distinct(cls, key: str, **kwargs) -> List[Any]: ...
126
    
127
    # State management
128
    @property
129
    def is_changed(self) -> bool: ...
130
    def get_changes(self) -> Dict[str, Any]: ...
131
    def rollback(self) -> None: ...
132

133
class DocumentWithSoftDelete(Document):
134
    deleted_at: Optional[datetime] = None
135
    
136
    async def delete(self) -> None: ...
137
    async def hard_delete(self) -> None: ...
138
    def is_deleted(self) -> bool: ...
139

140
class View(BaseModel):
141
    @classmethod
142
    def find(cls, *args, **kwargs) -> "FindInterface": ...
143
    @classmethod
144
    def aggregate(cls, pipeline: List[Dict]) -> "AggregateInterface": ...
145

146
class UnionDoc:
147
    @classmethod
148
    async def register_doc(cls, doc_model: Type[Document]) -> None: ...
149
    @classmethod
150
    def find(cls, *args, **kwargs) -> "FindInterface": ...
151
```
152

153
[Document Models](./documents.md)
154

155
### Field Types & Relationships
156

157
Type-safe field definitions with MongoDB-specific types, indexing capabilities, and document relationships with lazy loading support.
158

159
```python { .api }
160
class PydanticObjectId(ObjectId):
161
    def __init__(self, value: Union[str, ObjectId, None] = None): ...
162
    def __str__(self) -> str: ...
163

164
BeanieObjectId = PydanticObjectId
165

166
def Indexed(annotation, index_type=None, unique=False, sparse=False, **kwargs):
167
    """Create indexed field annotation for automatic index creation."""
168
    ...
169

170
class Link(Generic[T]):
171
    def __init__(self, ref: DBRef, document_class: Type[T]): ...
172
    async def fetch(self, fetch_links: bool = False) -> Union[T, "Link[T]"]: ...
173
    @classmethod
174
    async def fetch_list(cls, links: List[Union["Link[T]", T]], fetch_links: bool = False) -> List[Union[T, "Link[T]"]]: ...
175
    def to_dict(self) -> Dict[str, str]: ...
176

177
class BackLink(Generic[T]):
178
    def __init__(self, document_class: Type[T], original_field: str): ...
179
```
180

181
[Fields & Types](./fields-types.md)
182

183
### Event System & Actions
184

185
Pre/post event hooks for document lifecycle events with action type constants for flexible business logic integration.
186

187
```python { .api }
188
# Event decorators
189
def before_event(*actions):
190
    """Decorator to register functions that run before document events."""
191
    ...
192

193
def after_event(*actions):
194
    """Decorator to register functions that run after document events."""
195
    ...
196

197
# Action type constants
198
class Insert: ...
199
class Replace: ...
200
class Save: ...
201
class SaveChanges: ...
202
class ValidateOnSave: ...
203
class Delete: ...
204
class Update: ...
205

206
# Action direction constants  
207
class Before: ...
208
class After: ...
209
```
210

211
[Events & Actions](./events-actions.md)
212

213
### Bulk Operations
214

215
Efficient bulk write operations with transaction support for high-performance batch processing.
216

217
```python { .api }
218
class BulkWriter:
219
    def __init__(self, *args, **kwargs): ...
220
    async def __aenter__(self) -> "BulkWriter": ...
221
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
222
    
223
    def add_operation(self, operation: Dict) -> None: ...
224
    async def commit(self) -> Any: ...
225
```
226

227
[Bulk Operations](./bulk-operations.md)
228

229
### Database Migrations
230

231
Schema migration tools for iterative and free-form data migrations with batch processing support.
232

233
```python { .api }
234
def iterative_migration(document_models: List[Type[Document]], batch_size: int = 1000):
235
    """Decorator for creating iterative data migrations with batching."""
236
    ...
237

238
def free_fall_migration(document_models: List[Type[Document]]):
239
    """Decorator for creating free-form migration functions."""
240
    ...
241
```
242

243
[Migrations](./migrations.md)
244

245
### Query Operations & Responses
246

247
Query response configuration and utility enumerations for sorting and document merging strategies.
248

249
```python { .api }
250
class UpdateResponse(Enum):
251
    UPDATE_RESULT = "update_result"
252
    OLD_DOCUMENT = "old_document"
253
    NEW_DOCUMENT = "new_document"
254

255
class SortDirection(Enum):
256
    ASCENDING = 1
257
    DESCENDING = -1
258

259
class MergeStrategy(Enum):
260
    local = "local"
261
    remote = "remote"
262
```
263

264
[Query Operations](./query-operations.md)
265

266
### Time Series Collections
267

268
MongoDB time series collection configuration with granularity control and TTL support.
269

270
```python { .api }
271
class TimeSeriesConfig(BaseModel):
272
    time_field: str
273
    meta_field: Optional[str] = None
274
    granularity: Optional[Granularity] = None
275
    expire_after_seconds: Optional[int] = None
276

277
class Granularity(Enum):
278
    seconds = "seconds"
279
    minutes = "minutes"
280
    hours = "hours"
281
```
282

283
[Time Series](./time-series.md)
284

285
### Custom Types
286

287
Pydantic-compatible custom field types for MongoDB-specific data types like Decimal128 and Binary.
288

289
```python { .api }
290
DecimalAnnotation: TypeAlias
291
"""Pydantic annotation for handling BSON Decimal128 types."""
292

293
BsonBinary: TypeAlias
294
"""Pydantic-compatible BSON binary field type."""
295
```
296

297
[Custom Types](./custom-types.md)
298

299
### Initialization
300

301
Database and ODM initialization with connection management and model registration.
302

303
```python { .api }
304
async def init_beanie(
305
    database,
306
    document_models: List[Union[Type[Document], Type[View], str]],
307
    connection_string: Optional[str] = None,
308
    allow_index_dropping: bool = False,
309
    recreate_views: bool = False,
310
    skip_indexes: bool = False
311
) -> None:
312
    """Initialize Beanie ODM with database connection and document models."""
313
    ...
314
```
315

316
[Initialization](./initialization.md)
317

318
## Types
319

320
```python { .api }
321
# Common type aliases
322
DocumentType = TypeVar("DocumentType", bound=Document)
323
FindInterface = TypeVar("FindInterface")
324
AggregateInterface = TypeVar("AggregateInterface")
325

326
# Rule enumerations
327
class WriteRules(Enum):
328
    DO_NOTHING = "DO_NOTHING"
329
    WRITE = "WRITE"
330

331
class DeleteRules(Enum):
332
    DO_NOTHING = "DO_NOTHING" 
333
    DELETE_LINKS = "DELETE_LINKS"
334
```