A Python library that offers an easy-to-use Repository pattern for MongoDB, supporting both synchronous and asynchronous operations. It simplifies working with databases by providing a clear interface for CRUD (Create, Read, Update, Delete) operations using Pydantic models. With built-in data validation and serialization from Pydantic, it helps manage your MongoDB data safely.
- Asynchronous and Synchronous support
- Pydantic models integration
- Type-safe MongoDB operations
- Cursor-based pagination
pip install pydantic-mongofrom bson import ObjectId
from pydantic import BaseModel
from pydantic_mongo import AbstractRepository, PydanticObjectId
from pymongo import MongoClient
from typing import Optional, List
# Define your models
class Foo(BaseModel):
count: int
size: float = None
class Bar(BaseModel):
apple: str = 'x'
banana: str = 'y'
class Spam(BaseModel):
# PydanticObjectId is an alias to Annotated[ObjectId, ObjectIdAnnotation]
id: Optional[PydanticObjectId] = None
foo: Foo
bars: List[Bar]
# Create a repository
class SpamRepository(AbstractRepository[Spam]):
class Meta:
collection_name = 'spams'
# Connect to database
client = MongoClient("mongodb://localhost:27017")
database = client["example"]
repo = SpamRepository(database)# Create a new document
spam = Spam(foo=Foo(count=1, size=1.0), bars=[Bar()])
# Create a document with predefined ID
spam_with_predefined_id = Spam(
id=ObjectId("611827f2878b88b49ebb69fc"),
foo=Foo(count=2, size=2.0),
bars=[Bar()]
)
# Save a single document
repo.save(spam) # spam.id is now set to an ObjectId
# Save multiple documents
repo.save_many([spam, spam_with_predefined_id])# Find by ID
result = repo.find_one_by_id(spam.id)
# Find by ID using string
result = repo.find_one_by_id(ObjectId('611827f2878b88b49ebb69fc'))
assert result.foo.count == 2
# Find one by custom query
result = repo.find_one_by({'foo.count': 1})
# Find multiple documents by query
results = repo.find_by({'foo.count': {'$gte': 1}})# Get first page
edges = repo.paginate({'foo.count': {'$gte': 1}}, limit=10)
# Get next page using the last cursor
more_edges = repo.paginate(
{'foo.count': {'$gte': 1}},
limit=10,
after=list(edges)[-1].cursor
)# Delete a document
repo.delete(spam)
# Delete by ID
repo.delete_by_id(ObjectId("..."))For asynchronous applications, you can use AsyncAbstractRepository which provides the same functionality as AbstractRepository but with async/await support:
from pymongo import AsyncMongoClient
from pydantic import BaseModel
from pydantic_mongo import AsyncAbstractRepository
class User(BaseModel):
id: str
name: str
email: str
class UserRepository(AsyncAbstractRepository[User]):
class Meta:
collection_name = 'users'
# Initialize database connection
client = AsyncMongoClient('mongodb://localhost:27017')
database = client["mydb"]
# Create repository instance
user_repo = UserRepository(database)
# Example usage
user = User(name='John Doe', email='[email protected]')
await user_repo.save(user)
user = await user_repo.find_one_by_id(user_id)MIT License
