tessl/pypi-cloudant
IBM Cloudant Python client library providing comprehensive interface for Cloudant and CouchDB databases
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
document-operations.mddocs/
Document Operations
Complete CRUD operations for JSON documents with attachment support, conflict resolution, batch operations, and field-level update utilities.
Capabilities
Document Access
Access documents through dict-like interface on the database.
class CouchDatabase(dict):
"""Dict-like interface for document access."""
def keys(self, remote=False):
"""
Get all document IDs in the database.
Parameters:
- remote (bool): Fetch from server (default: False, uses cache)
Returns:
list[str]: Document IDs
"""
def __getitem__(self, key):
"""
Access document by ID.
Parameters:
- key (str): Document ID
Returns:
Document: Document instance
"""
def get(self, key, remote=False):
"""
Safe document access.
Parameters:
- key (str): Document ID
- remote (bool): Fetch from server if not cached
Returns:
Document | None: Document instance or None if not found
"""
def __contains__(self, key):
"""
Check if document exists.
Parameters:
- key (str): Document ID
Returns:
bool: True if document exists
"""
def __iter__(self, remote=True):
"""
Iterate over all documents.
Parameters:
- remote (bool): Fetch from server (default: True)
Yields:
Document: Document instances
"""Document Creation
Create new documents in the database.
class CouchDatabase(dict):
"""Document creation methods."""
def create_document(self, data, throw_on_exists=False):
"""
Create document with provided data.
Parameters:
- data (dict): Document data including optional _id
- throw_on_exists (bool): Raise exception if document exists
Returns:
Document: Created document instance
Raises:
CloudantDocumentException: Document creation failed
"""
def new_document(self):
"""
Create empty document with generated ID.
Returns:
Document: Empty document with auto-generated _id
"""Document Class
Core document operations with CRUD functionality.
class Document(dict):
"""
Represents a CouchDB/Cloudant document with CRUD operations.
"""
def __init__(self, database, document_id=None, **kwargs):
"""
Initialize document instance.
Parameters:
- database (CouchDatabase | CloudantDatabase): Parent database
- document_id (str): Document ID (auto-generated if None)
- **kwargs: Initial document fields
"""
def exists(self):
"""
Check if document exists in database.
Returns:
bool: True if document exists on server
"""
def create(self):
"""
Create document in database.
Returns:
dict: Creation response with _id and _rev
Raises:
CloudantDocumentException: Document already exists or creation failed
"""
def fetch(self):
"""
Fetch latest document version from database.
Returns:
None: Document data is updated in-place
Raises:
CloudantDocumentException: Document not found or fetch failed
"""
def save(self):
"""
Save document changes to database.
Returns:
dict: Save response with updated _rev
Raises:
CloudantDocumentException: Conflict or save failed
"""
def delete(self):
"""
Delete document from database.
Returns:
dict: Deletion response
Raises:
CloudantDocumentException: Deletion failed
"""
def json(self):
"""
Get JSON representation of document.
Returns:
str: JSON string of document data
"""
@property
def document_url(self):
"""Document URL endpoint"""
@property
def r_session(self):
"""HTTP request session"""Field Operations
Utilities for updating document fields with conflict resolution.
class Document(dict):
"""Field update operations."""
def update_field(self, action, field, value, max_tries=10):
"""
Update document field with conflict resolution.
Parameters:
- action (callable): Update function (list_field_append, list_field_remove, field_set)
- field (str): Field name to update
- value: Value for the update operation
- max_tries (int): Maximum retry attempts for conflicts
Returns:
dict: Update response
Raises:
CloudantDocumentException: Update failed after max retries
"""
@staticmethod
def list_field_append(doc, field, value):
"""
Helper to append value to list field.
Parameters:
- doc (dict): Document data
- field (str): Field name
- value: Value to append
Returns:
dict: Updated document
"""
@staticmethod
def list_field_remove(doc, field, value):
"""
Helper to remove value from list field.
Parameters:
- doc (dict): Document data
- field (str): Field name
- value: Value to remove
Returns:
dict: Updated document
"""
@staticmethod
def field_set(doc, field, value):
"""
Helper to set field value.
Parameters:
- doc (dict): Document data
- field (str): Field name
- value: Value to set
Returns:
dict: Updated document
"""Attachment Operations
Manage binary attachments on documents.
class Document(dict):
"""Attachment management."""
def get_attachment(self, attachment, headers=None, write_to=None, attachment_type=None):
"""
Retrieve document attachment.
Parameters:
- attachment (str): Attachment name
- headers (dict): Additional HTTP headers
- write_to (file-like): Stream attachment to file object
- attachment_type (str): Expected MIME type
Returns:
bytes | None: Attachment content (None if write_to specified)
Raises:
CloudantDocumentException: Attachment not found or retrieval failed
"""
def put_attachment(self, attachment, content_type, data, headers=None):
"""
Add or update document attachment.
Parameters:
- attachment (str): Attachment name
- content_type (str): MIME type of attachment
- data (bytes | file-like): Attachment content
- headers (dict): Additional HTTP headers
Returns:
dict: Upload response with updated _rev
Raises:
CloudantDocumentException: Upload failed
"""
def delete_attachment(self, attachment, headers=None):
"""
Delete document attachment.
Parameters:
- attachment (str): Attachment name to delete
- headers (dict): Additional HTTP headers
Returns:
dict: Deletion response with updated _rev
Raises:
CloudantDocumentException: Deletion failed
"""Context Manager Support
Automatic fetching and saving with context manager.
class Document(dict):
"""Context manager support."""
def __enter__(self):
"""
Context manager entry - fetch document if it exists.
Returns:
Document: Self reference
"""
def __exit__(self, exc_type, exc_value, traceback):
"""
Context manager exit - save document if no exception.
Parameters:
- exc_type: Exception type (if any)
- exc_value: Exception value (if any)
- traceback: Exception traceback (if any)
Returns:
bool: False (don't suppress exceptions)
"""Bulk Operations
Batch operations for multiple documents.
class CouchDatabase(dict):
"""Bulk document operations."""
def bulk_docs(self, docs):
"""
Perform bulk document operations.
Parameters:
- docs (list[dict]): List of documents to create/update/delete
Each doc can include _deleted: true for deletion
Returns:
list[dict]: List of operation results with _id, _rev, or error info
Raises:
CloudantDatabaseException: Bulk operation failed
"""Document Querying
Primary index and document iteration.
class CouchDatabase(dict):
"""Document querying."""
def all_docs(self, **kwargs):
"""
Query primary index (_all_docs).
Parameters:
- include_docs (bool): Include document content
- startkey (str): Start key for range
- endkey (str): End key for range
- keys (list[str]): Specific document IDs to retrieve
- limit (int): Maximum results
- skip (int): Number of results to skip
- descending (bool): Reverse order
- inclusive_end (bool): Include endkey in results
Returns:
Result: Query result iterator
"""
def partitioned_all_docs(self, partition_key, **kwargs):
"""
Query primary index for specific partition.
Parameters:
- partition_key (str): Partition identifier
- **kwargs: Same options as all_docs()
Returns:
Result: Query result iterator
"""Usage Examples
Basic Document Operations
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Create document with specific ID
doc_data = {
'_id': 'user123',
'name': 'John Doe',
'email': 'john@example.com',
'tags': ['developer', 'python']
}
doc = db.create_document(doc_data)
print(f"Created: {doc['_id']} with rev {doc['_rev']}")
# Create document with auto-generated ID
new_doc = db.new_document()
new_doc['name'] = 'Jane Smith'
new_doc['role'] = 'designer'
new_doc.create()
# Access existing document
user_doc = db['user123']
# Check if document exists
if user_doc.exists():
# Fetch latest version
user_doc.fetch()
print(f"User: {user_doc['name']}")
# Update document
user_doc['last_login'] = '2023-01-15'
user_doc['tags'].append('admin')
user_doc.save()
# Delete document
user_doc.delete()Context Manager Usage
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Automatic fetch and save
with db['user123'] as doc:
doc['last_updated'] = '2023-01-15'
doc['login_count'] = doc.get('login_count', 0) + 1
# Document is automatically saved on context exitField Operations with Conflict Resolution
from cloudant import cloudant
from cloudant.document import Document
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
doc = db['user123']
# Append to list field with automatic conflict resolution
doc.update_field(Document.list_field_append, 'tags', 'admin')
# Remove from list field
doc.update_field(Document.list_field_remove, 'tags', 'developer')
# Set field value
doc.update_field(Document.field_set, 'status', 'active')Attachment Operations
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
doc = db['user123']
# Add attachment
with open('profile.jpg', 'rb') as f:
attachment_data = f.read()
doc.put_attachment('profile_picture', 'image/jpeg', attachment_data)
# Retrieve attachment
image_data = doc.get_attachment('profile_picture')
# Save attachment to file
with open('downloaded_profile.jpg', 'wb') as f:
doc.get_attachment('profile_picture', write_to=f)
# Delete attachment
doc.delete_attachment('profile_picture')Bulk Operations
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Bulk create/update documents
docs = [
{'_id': 'user1', 'name': 'Alice', 'role': 'admin'},
{'_id': 'user2', 'name': 'Bob', 'role': 'user'},
{'_id': 'user3', 'name': 'Carol', 'role': 'moderator'}
]
results = db.bulk_docs(docs)
for result in results:
if 'error' in result:
print(f"Error for {result['id']}: {result['error']}")
else:
print(f"Success: {result['id']} -> {result['rev']}")
# Bulk delete documents
delete_docs = [
{'_id': 'user1', '_rev': '1-abc123', '_deleted': True},
{'_id': 'user2', '_rev': '1-def456', '_deleted': True}
]
db.bulk_docs(delete_docs)Document Iteration and Querying
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Iterate over all documents
for doc in db:
print(f"Document: {doc['_id']}")
# Query with all_docs
result = db.all_docs(include_docs=True, limit=10)
for row in result:
doc = row['doc']
print(f"ID: {doc['_id']}, Name: {doc.get('name', 'N/A')}")
# Query specific documents
user_ids = ['user1', 'user2', 'user3']
result = db.all_docs(keys=user_ids, include_docs=True)
for row in result:
if 'doc' in row:
print(f"Found user: {row['doc']['name']}")
# Range query
result = db.all_docs(
startkey='user',
endkey='user\ufff0', # Unicode high character for prefix matching
include_docs=True
)
for row in result:
print(f"User document: {row['doc']['_id']}")Partitioned Document Operations (Cloudant)
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
# Create partitioned database
db = client.create_database('partitioned_db', partitioned=True)
# Create document in partition
doc = db.create_document({
'_id': 'user123:profile', # partition_key:document_id format
'name': 'John Doe',
'partition': 'user123'
})
# Query partition
result = db.partitioned_all_docs('user123', include_docs=True)
for row in result:
print(f"Partition doc: {row['doc']['_id']}")Error Handling
Document operations can raise CloudantDocumentException:
from cloudant import cloudant
from cloudant.error import CloudantDocumentException
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
try:
# Try to fetch non-existent document
doc = db['non_existent']
doc.fetch()
except CloudantDocumentException as e:
print(f"Document not found: {e}")
try:
# Try to save document with stale revision
doc = db['user123']
doc.fetch()
doc['_rev'] = '1-old_revision' # Simulate stale revision
doc['name'] = 'Updated Name'
doc.save()
except CloudantDocumentException as e:
print(f"Conflict error: {e}")
# Fetch latest and retry
doc.fetch()
doc['name'] = 'Updated Name'
doc.save()Install with Tessl CLI
npx tessl i tessl/pypi-cloudant