The Collection class represents a non-transactional collection handle.
Collection management
close
Frees the collection handle.
Example:
coll = db.get_collection("users")
try:
# Use collection
pass
finally:
coll.close()
Context manager support
Use the with statement for automatic resource cleanup.
Example:
with db.get_collection("users") as coll:
doc_id = coll.insert({"name": "Bob", "age": 25})
# Collection is automatically closed when exiting the block
name
Gets the collection name.
Basic CRUD operations
insert
Inserts a document.
def insert(doc: Dict[str, Any]) -> str
ID of the inserted document
coll = db.get_collection("users")
doc_id = coll.insert({"name": "Alice", "age": 30})
print(f"Inserted: {doc_id}")
find_by_id
Finds a document by ID.
def find_by_id(doc_id: str) -> Optional[Dict[str, Any]]
Document if found, None otherwise
doc = coll.find_by_id("user123")
if doc:
print(f"Found: {doc}")
update_by_id
Updates a document by ID.
def update_by_id(doc_id: str, doc: Dict[str, Any]) -> None
coll.update_by_id("user123", {"name": "Alice Smith", "age": 31})
delete_by_id
Deletes a document by ID.
def delete_by_id(doc_id: str) -> None
coll.delete_by_id("user123")
find_all
Finds all documents.
def find_all() -> List[Dict[str, Any]]
all_docs = coll.find_all()
for doc in all_docs:
print(doc)
count
Counts all documents.
Example:
total = coll.count()
print(f"Total documents: {total}")
Query and filter operations
find
Finds documents matching a filter.
def find(filter_str: str) -> List[Dict[str, Any]]
List of matching documents
results = coll.find("age > 25")
for user in results:
print(user)
find_one
Finds first document matching a filter.
def find_one(filter_str: str) -> Optional[Dict[str, Any]]
First matching document, or None
user = coll.find_one("email == '[email protected]'")
if user:
print(f"Found: {user}")
update
Updates all documents matching a filter. Returns count updated.
def update(filter_str: str, update: Dict[str, Any]) -> int
Number of documents updated
count = coll.update("age > 25", {"verified": True})
print(f"Updated {count} documents")
update_one
Updates first document matching a filter. Returns True if updated.
def update_one(filter_str: str, update: Dict[str, Any]) -> bool
True if a document was updated
updated = coll.update_one("email == '[email protected]'", {"verified": True})
if updated:
print("Document updated")
delete
Deletes all documents matching a filter. Returns count deleted.
def delete(filter_str: str) -> int
Number of documents deleted
count = coll.delete("age < 18")
print(f"Deleted {count} documents")
delete_one
Deletes first document matching a filter. Returns True if deleted.
def delete_one(filter_str: str) -> bool
True if a document was deleted
deleted = coll.delete_one("email == '[email protected]'")
if deleted:
print("Document deleted")
Upsert operations
upsert_by_id
Upserts a document by ID. Returns UpsertResult with id and inserted flag.
def upsert_by_id(doc_id: str, doc: Dict[str, Any]) -> UpsertResult
Result containing id and whether it was inserted or updated
result = coll.upsert_by_id("user123", {"name": "Alice", "age": 30})
if result.inserted:
print(f"Inserted new document: {result.id}")
else:
print(f"Updated existing document: {result.id}")
upsert
Upserts a document matching a filter. Returns UpsertResult with id and inserted flag.
def upsert(filter_str: str, doc: Dict[str, Any]) -> UpsertResult
Result containing id and whether it was inserted or updated
Bulk operations
insert_many
Inserts multiple documents. Returns list of IDs.
def insert_many(docs: List[Dict[str, Any]]) -> List[str]
docs
List[Dict[str, Any]]
required
List of documents to insert
List of inserted document IDs
ids = coll.insert_many([
{"name": "Alice", "age": 30},
{"name": "Bob", "age": 25},
{"name": "Charlie", "age": 35}
])
print(f"Inserted {len(ids)} documents")
bulk_write
Executes multiple operations in a transaction.
def bulk_write(operations: List[Dict[str, Any]], ordered: bool = True) -> Dict[str, Any]
operations
List[Dict[str, Any]]
required
List of bulk operations
Whether to execute operations in order
Result with statistics for each operation type
operations = [
{"op": "insert", "doc": {"name": "Alice", "age": 30}},
{"op": "update", "filter": "age > 25", "update": {"verified": True}},
{"op": "delete", "filter": "age < 18"}
]
result = coll.bulk_write(operations)
print(f"Bulk write result: {result}")
Advanced operations
distinct
Gets distinct values for a field.
def distinct(field: str) -> List[Any]
ages = coll.distinct("age")
print(f"Distinct ages: {ages}")
count_distinct
Counts distinct values for a field.
def count_distinct(field: str) -> int
Number of distinct values
count = coll.count_distinct("city")
print(f"Number of unique cities: {count}")
search
Performs full-text search.
def search(query: str) -> List[Dict[str, Any]]
List of matching documents
results = coll.search("python database")
for doc in results:
print(doc)
count_with_query
Counts documents matching a filter.
def count_with_query(filter_str: str) -> int
Number of matching documents
count = coll.count_with_query("age > 25")
print(f"Users over 25: {count}")
aggregate
Executes an aggregation pipeline.
def aggregate(pipeline: List[Dict[str, Any]]) -> List[Dict[str, Any]]
pipeline
List[Dict[str, Any]]
required
Aggregation pipeline stages
pipeline = [
{"$match": {"age": {">=": 18}}},
{"$group": {"_id": "$city", "count": {"$sum": 1}}},
{"$sort": {"count": -1}}
]
results = coll.aggregate(pipeline)
for result in results:
print(result)
Query builder helpers
query_with_options
Executes a query with all options.
def query_with_options(
filter_str: Optional[str] = None,
sort_field: Optional[str] = None,
sort_asc: bool = True,
limit: int = 0,
skip: int = 0,
project_fields: Optional[List[str]] = None,
exclude_fields: Optional[List[str]] = None,
) -> List[Dict[str, Any]]
Maximum number of results (0 = no limit)
Number of results to skip
Fields to include in results
Fields to exclude from results
List of matching documents
results = coll.query_with_options(
filter_str="age > 25",
sort_field="age",
sort_asc=False,
limit=10,
skip=5,
project_fields=["name", "age"]
)
query_count
Counts documents with query options.
def query_count(filter_str: Optional[str] = None, skip: int = 0, limit: int = 0) -> int
Number of results to skip
Maximum number of results to count
Number of matching documents
query_first
Gets the first document matching a query.
def query_first(
filter_str: Optional[str] = None,
sort_field: Optional[str] = None,
sort_asc: bool = True,
) -> Optional[Dict[str, Any]]
First matching document, or None
oldest = coll.query_first(sort_field="age", sort_asc=False)
if oldest:
print(f"Oldest user: {oldest}")
Watch and change streams
watch
Starts watching for changes on the collection.
def watch(
callback: Callable[[str, str, Optional[Dict[str, Any]]], None],
filter_str: Optional[str] = None,
) -> WatchHandle
callback
Callable[[str, str, Optional[Dict[str, Any]]], None]
required
Callback function receiving (operation, doc_id, document)
Optional filter for watched events
def on_change(op, doc_id, doc):
print(f"{op}: {doc_id}")
if doc:
print(f" Data: {doc}")
handle = coll.watch(on_change)
# ... later ...
handle.stop()