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
views-design-documents.mddocs/
Views and Design Documents
Manage MapReduce views, list functions, show functions, and update handlers through design documents. Design documents are special documents that contain application code for database operations including views, indexes, and server-side functions.
Capabilities
Design Document Management
Access and manage design documents containing views and functions.
class CouchDatabase(dict):
"""Design document access methods."""
def design_documents(self):
"""
Get all design documents in the database.
Returns:
dict: All design documents keyed by design document ID
"""
def list_design_documents(self):
"""
List design document names.
Returns:
list[str]: Design document IDs (without _design/ prefix)
"""
def get_design_document(self, ddoc_id):
"""
Get specific design document.
Parameters:
- ddoc_id (str): Design document ID (without _design/ prefix)
Returns:
DesignDocument: Design document instance
Raises:
CloudantDesignDocumentException: Design document not found
"""DesignDocument Class
Special document type for managing views, indexes, and functions.
class DesignDocument(Document):
"""
Design document containing views, indexes, and server-side functions.
"""
def __init__(self, database, document_id=None, partitioned=False):
"""
Initialize design document.
Parameters:
- database (CouchDatabase | CloudantDatabase): Parent database
- document_id (str): Design document ID (without _design/ prefix)
- partitioned (bool): Whether views are partitioned
"""
@property
def views(self):
"""
MapReduce views in this design document.
Returns:
dict: View definitions with map/reduce functions
"""
@property
def shows(self):
"""
Show functions for formatting documents.
Returns:
dict: Show function definitions
"""
@property
def lists(self):
"""
List functions for formatting view results.
Returns:
dict: List function definitions
"""
@property
def updates(self):
"""
Update handlers for document modifications.
Returns:
dict: Update handler definitions
"""
@property
def filters(self):
"""
Filter functions for change feeds.
Returns:
dict: Filter function definitions
"""
@property
def validate_doc_update(self):
"""
Document validation function.
Returns:
str: Validation function JavaScript code
"""
@property
def rewrites(self):
"""
URL rewrite rules.
Returns:
list[dict]: Rewrite rule definitions
"""
@property
def indexes(self):
"""
Search indexes (Cloudant only).
Returns:
dict: Search index definitions
"""View Class
Individual MapReduce view management.
class View:
"""
MapReduce view within a design document.
"""
def __init__(self, design_document, view_name, **kwargs):
"""
Initialize view instance.
Parameters:
- design_document (DesignDocument): Parent design document
- view_name (str): View name within design document
- **kwargs: View options and parameters
"""
def __call__(self, **kwargs):
"""
Execute view query.
Parameters:
- key: Single key to query
- keys (list): Multiple keys to query
- startkey: Start of key range
- endkey: End of key range
- startkey_docid (str): Start document ID for pagination
- endkey_docid (str): End document ID for pagination
- limit (int): Maximum results
- skip (int): Number of results to skip
- descending (bool): Reverse sort order
- group (bool): Group results by key
- group_level (int): Group level for complex keys
- reduce (bool): Use reduce function (default: True if available)
- include_docs (bool): Include document content
- inclusive_end (bool): Include endkey in results
- update_seq (bool): Include update sequence in response
- conflicts (bool): Include conflict information
- attachments (bool): Include attachment stubs
- att_encoding_info (bool): Include attachment encoding info
Returns:
Result: View result iterator
Raises:
CloudantViewException: View query failed
"""
def custom_result(self, **options):
"""
Execute view with custom result options.
Parameters:
- **options: Result collection options
Returns:
Result: Custom result iterator
"""
@property
def result(self):
"""Default view result collection"""
@property
def map_function(self):
"""
JavaScript map function code.
Returns:
str: Map function JavaScript code
"""
@property
def reduce_function(self):
"""
JavaScript reduce function code.
Returns:
str | None: Reduce function JavaScript code or None
"""View Execution Methods
Database methods for executing views.
class CouchDatabase(dict):
"""View execution methods."""
def get_view_result(self, ddoc_id, view_name, raw_result=False, **kwargs):
"""
Execute MapReduce view and return results.
Parameters:
- ddoc_id (str): Design document ID (without _design/ prefix)
- view_name (str): View name within design document
- raw_result (bool): Return raw response without processing
- **kwargs: View query parameters (same as View.__call__)
Returns:
Result | dict: View results or raw response
Raises:
CloudantViewException: View execution failed
"""
def get_partitioned_view_result(self, partition_key, ddoc_id, view_name, raw_result=False, **kwargs):
"""
Execute view within specific partition.
Parameters:
- partition_key (str): Partition identifier
- ddoc_id (str): Design document ID
- view_name (str): View name
- raw_result (bool): Return raw response
- **kwargs: View query parameters
Returns:
Result | dict: Partitioned view results
"""List and Show Functions
Server-side formatting functions.
class CouchDatabase(dict):
"""List and show function execution."""
def get_list_function_result(self, ddoc_id, list_name, view_name, **kwargs):
"""
Execute list function to format view results.
Parameters:
- ddoc_id (str): Design document ID
- list_name (str): List function name
- view_name (str): View name to format
- **kwargs: View parameters and list options
Returns:
str: Formatted output from list function
Raises:
CloudantDesignDocumentException: List function execution failed
"""
def get_show_function_result(self, ddoc_id, show_name, doc_id):
"""
Execute show function to format document.
Parameters:
- ddoc_id (str): Design document ID
- show_name (str): Show function name
- doc_id (str): Document ID to format
Returns:
str: Formatted output from show function
Raises:
CloudantDesignDocumentException: Show function execution failed
"""Update Handlers
Server-side document update functions.
class CouchDatabase(dict):
"""Update handler execution."""
def update_handler_result(self, ddoc_id, handler_name, doc_id=None, data=None, **params):
"""
Execute update handler function.
Parameters:
- ddoc_id (str): Design document ID
- handler_name (str): Update handler name
- doc_id (str): Document ID to update (optional for create)
- data (dict | str): Request body data
- **params: Additional URL parameters
Returns:
dict: Update handler response
Raises:
CloudantDesignDocumentException: Update handler execution failed
"""Usage Examples
Creating Design Documents
from cloudant import cloudant
from cloudant.design_document import DesignDocument
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Create design document with views
ddoc = DesignDocument(db, 'user_views')
# Add MapReduce view
ddoc['views'] = {
'by_status': {
'map': '''
function(doc) {
if (doc.type === 'user') {
emit(doc.status, {
name: doc.name,
email: doc.email,
created: doc.created_date
});
}
}
'''
},
'count_by_department': {
'map': '''
function(doc) {
if (doc.type === 'user' && doc.department) {
emit(doc.department, 1);
}
}
''',
'reduce': '_count' # Built-in reduce function
}
}
# Add show function
ddoc['shows'] = {
'user_profile': '''
function(doc, req) {
if (!doc) {
return {
code: 404,
body: 'User not found'
};
}
var html = '<h1>' + doc.name + '</h1>';
html += '<p>Email: ' + doc.email + '</p>';
html += '<p>Status: ' + doc.status + '</p>';
return {
headers: {'Content-Type': 'text/html'},
body: html
};
}
'''
}
# Add list function
ddoc['lists'] = {
'user_csv': '''
function(head, req) {
var row;
start({headers: {'Content-Type': 'text/csv'}});
send('name,email,status\\n');
while (row = getRow()) {
var user = row.value;
send(user.name + ',' + user.email + ',' + row.key + '\\n');
}
}
'''
}
# Add update handler
ddoc['updates'] = {
'status_update': '''
function(doc, req) {
if (!doc) {
return [null, {
code: 404,
body: 'Document not found'
}];
}
var new_status = req.form.status;
if (new_status) {
doc.status = new_status;
doc.updated_date = new Date().toISOString();
return [doc, {
code: 200,
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({status: 'updated'})
}];
}
return [null, {
code: 400,
body: 'Status parameter required'
}];
}
'''
}
# Save design document
ddoc.save()
print(f"Created design document: {ddoc['_id']}")Querying Views
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Query view with specific key
result = db.get_view_result('user_views', 'by_status', key='active')
for row in result:
user = row['value']
print(f"Active user: {user['name']} ({user['email']})")
# Query view with key range
result = db.get_view_result(
'user_views',
'by_status',
startkey='active',
endkey='pending',
include_docs=True
)
for row in result:
print(f"User {row['key']}: {row['doc']['name']}")
# Query with multiple keys
statuses = ['active', 'pending', 'suspended']
result = db.get_view_result('user_views', 'by_status', keys=statuses)
# Reduced view (count)
result = db.get_view_result('user_views', 'count_by_department')
for row in result:
print(f"Department {row['key']}: {row['value']} users")Using View Objects
from cloudant import cloudant
from cloudant.view import View
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
ddoc = db.get_design_document('user_views')
# Create reusable view object
status_view = View(ddoc, 'by_status')
# Query active users
active_users = status_view(key='active', include_docs=True)
for row in active_users:
print(f"Active: {row['doc']['name']}")
# Query with custom result options
paginated_result = status_view.custom_result(
limit=10,
skip=0,
key='active'
)
for row in paginated_result:
print(f"User: {row['value']['name']}")Complex View Queries
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Pagination with startkey_docid
last_key = None
last_docid = None
page_size = 25
while True:
params = {
'limit': page_size + 1, # Get one extra to check for more
'include_docs': True
}
if last_key is not None:
params['startkey'] = last_key
params['startkey_docid'] = last_docid
params['skip'] = 1 # Skip the last doc from previous page
result = db.get_view_result('user_views', 'by_status', **params)
rows = list(result)
if not rows:
break
# Process current page (excluding extra row)
page_rows = rows[:page_size]
for row in page_rows:
print(f"User: {row['doc']['name']}")
# Check if there are more pages
if len(rows) <= page_size:
break
# Set pagination markers for next page
last_row = rows[page_size - 1]
last_key = last_row['key']
last_docid = last_row['id']
# Group reduce queries
result = db.get_view_result(
'user_views',
'count_by_department',
group=True,
group_level=1
)
for row in result:
print(f"Department: {row['key']}, Count: {row['value']}")List and Show Functions
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Execute show function
html_output = db.get_show_function_result(
'user_views',
'user_profile',
'user123'
)
print("HTML Profile:", html_output)
# Execute list function
csv_output = db.get_list_function_result(
'user_views',
'user_csv',
'by_status',
key='active'
)
print("CSV Export:", csv_output)Update Handlers
from cloudant import cloudant
import json
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Execute update handler
response = db.update_handler_result(
'user_views',
'status_update',
'user123',
data={'status': 'suspended'}
)
print(f"Update response: {response}")
# Update handler for new document
response = db.update_handler_result(
'user_views',
'create_user',
data=json.dumps({
'name': 'New User',
'email': 'newuser@example.com',
'status': 'pending'
})
)Partitioned Views (Cloudant)
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
# Use partitioned database
db = client['partitioned_database']
# Create partitioned design document
ddoc = db.get_design_document('partition_views')
ddoc['views'] = {
'by_activity_type': {
'map': '''
function(doc) {
if (doc.type === 'activity') {
emit(doc.activity_type, {
timestamp: doc.timestamp,
details: doc.details
});
}
}
'''
}
}
ddoc['options'] = {'partitioned': True}
ddoc.save()
# Query view within partition
result = db.get_partitioned_view_result(
'user123', # partition key
'partition_views',
'by_activity_type',
key='login',
limit=10
)
for row in result:
activity = row['value']
print(f"Login activity: {activity['timestamp']}")View Performance Optimization
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Query view without reduce for better performance on large datasets
result = db.get_view_result(
'user_views',
'count_by_department',
reduce=False, # Skip reduce function
include_docs=False # Don't fetch full documents
)
# Count manually if needed
department_counts = {}
for row in result:
dept = row['key']
department_counts[dept] = department_counts.get(dept, 0) + 1
# Use stale views for better performance (eventual consistency)
result = db.get_view_result(
'user_views',
'by_status',
stale='ok', # Don't wait for index updates
key='active'
)Error Handling
View and design document operations can raise specific exceptions:
from cloudant import cloudant
from cloudant.error import (
CloudantDesignDocumentException,
CloudantViewException
)
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
try:
# Query non-existent view
result = db.get_view_result('missing_ddoc', 'missing_view')
except CloudantViewException as e:
print(f"View query failed: {e}")
try:
# Execute non-existent show function
output = db.get_show_function_result('ddoc', 'missing_show', 'doc123')
except CloudantDesignDocumentException as e:
print(f"Show function failed: {e}")
try:
# Create design document with invalid JavaScript
ddoc = db.get_design_document('invalid_views')
ddoc['views'] = {
'bad_view': {
'map': 'invalid javascript code here'
}
}
ddoc.save()
except CloudantDesignDocumentException as e:
print(f"Design document save failed: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-cloudant