Skip to main content
The Transaction class represents a database transaction.

​
Transaction lifecycle

​
commit

Commits the transaction. 
def commit() -> None
Example: 
tx = db.begin_transaction()
try:
    # Perform operations
    tx.commit()
except Exception:
    tx.rollback()
    raise

​
rollback

Rolls back the transaction. 
def rollback() -> None
Example: 
tx = db.begin_transaction()
try:
    # Perform operations
    tx.commit()
except Exception:
    tx.rollback()
    raise

​
Context manager support

Transactions automatically commit on success and rollback on exception. Example: 
with db.begin_transaction() as tx:
    doc_id = tx.insert("users", {"name": "Alice", "age": 30})
    # Transaction is automatically committed if no exception occurs
    # or rolled back if an exception is raised

​
is_active

Checks if the transaction is still active. 
def is_active() -> bool
​
bool
bool
True if the transaction is active

​
Document operations

​
insert

Inserts a document into a collection. 
def insert(collection_name: str, doc: Dict[str, Any]) -> str
​
collection_name
str
required
Name of the collection
​
doc
Dict[str, Any]
required
Document to insert
​
str
str
ID of the inserted document
Example: 
tx = db.begin_transaction()
try:
    doc_id = tx.insert("users", {
        "name": "Alice",
        "age": 30,
        "email": "alice@example.com"
    })
    print(f"Inserted document with ID: {doc_id}")
    tx.commit()
except Exception:
    tx.rollback()
    raise

​
find_by_id

Finds a document by its ID. 
def find_by_id(collection_name: str, doc_id: str) -> Optional[Dict[str, Any]]
​
collection_name
str
required
Name of the collection
​
doc_id
str
required
Document ID to find
​
Optional[Dict[str, Any]]
Optional[Dict[str, Any]]
Document if found, None otherwise
Example: 
doc = tx.find_by_id("users", "user123")
if doc:
    print(f"Found: {doc}")
else:
    print("Document not found")

​
update_by_id

Updates a document by its ID. 
def update_by_id(collection_name: str, doc_id: str, doc: Dict[str, Any]) -> None
​
collection_name
str
required
Name of the collection
​
doc_id
str
required
Document ID to update
​
doc
Dict[str, Any]
required
Updated document data
Example: 
tx.update_by_id("users", "user123", {
    "name": "Alice Smith",
    "age": 31,
    "email": "alice@example.com"
})

​
delete_by_id

Deletes a document by its ID. 
def delete_by_id(collection_name: str, doc_id: str) -> None
​
collection_name
str
required
Name of the collection
​
doc_id
str
required
Document ID to delete
Example: 
tx.delete_by_id("users", "user123")

​
find_all

Finds all documents in a collection. 
def find_all(collection_name: str) -> List[Dict[str, Any]]
​
collection_name
str
required
Name of the collection
​
List[Dict[str, Any]]
List[Dict[str, Any]]
List of all documents
Example: 
all_users = tx.find_all("users")
for user in all_users:
    print(f"User: {user}")

​
count

Counts documents in a collection. 
def count(collection_name: str) -> int
​
collection_name
str
required
Name of the collection
​
int
int
Number of documents
Example: 
count = tx.count("users")
print(f"Total users: {count}")

​
Collection management

​
create_collection

Creates a new collection. 
def create_collection(collection_name: str) -> None
​
collection_name
str
required
Name of the collection to create
Example: 
tx.create_collection("products")

​
drop_collection

Drops a collection. 
def drop_collection(collection_name: str) -> None
​
collection_name
str
required
Name of the collection to drop
Example: 
tx.drop_collection("old_data")

​
rename_collection

Renames a collection. 
def rename_collection(old_name: str, new_name: str) -> None
​
old_name
str
required
Current name of the collection
​
new_name
str
required
New name for the collection
Example: 
tx.rename_collection("users", "customers")

Build docs developers (and LLMs) love

Get started for freeTalk to us