tessl/pypi-couchbase
Python Client for Couchbase providing comprehensive database operations including key-value, N1QL queries, search, analytics, and cluster management
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
view-operations.mddocs/
View Operations
Traditional MapReduce view queries for data indexing and querying. Views provide a way to create secondary indexes on document data using JavaScript MapReduce functions, offering compatibility with legacy Couchbase applications.
Capabilities
View Query Execution
Execute queries against MapReduce views defined in design documents.
class Bucket:
def view_query(self, design_doc: str, view_name: str, options: ViewOptions = None) -> ViewResult:
"""
Execute a view query.
Args:
design_doc (str): Name of the design document
view_name (str): Name of the view within the design document
options (ViewOptions, optional): Query options
Returns:
ViewResult: Query results with rows and metadata
Raises:
ViewException: If view execution fails
DesignDocumentNotFoundException: If design document not found
"""View Query Options
Configure view query execution parameters including keys, ranges, and result formatting.
class ViewOptions:
def __init__(self):
"""Create view query options."""
def key(self, key: Any) -> ViewOptions:
"""Query documents with specific key."""
def keys(self, keys: List[Any]) -> ViewOptions:
"""Query documents with specific keys."""
def start_key(self, start_key: Any) -> ViewOptions:
"""Set start key for range queries."""
def end_key(self, end_key: Any) -> ViewOptions:
"""Set end key for range queries."""
def start_key_doc_id(self, doc_id: str) -> ViewOptions:
"""Set start document ID for pagination."""
def end_key_doc_id(self, doc_id: str) -> ViewOptions:
"""Set end document ID for pagination."""
def inclusive_end(self, inclusive: bool) -> ViewOptions:
"""Include end key in results."""
def skip(self, skip: int) -> ViewOptions:
"""Skip number of results."""
def limit(self, limit: int) -> ViewOptions:
"""Limit number of results."""
def scan_consistency(self, consistency: ViewScanConsistency) -> ViewOptions:
"""Set scan consistency level."""
def reduce(self, reduce: bool) -> ViewOptions:
"""Enable/disable reduce function."""
def group(self, group: bool) -> ViewOptions:
"""Enable/disable grouping."""
def group_level(self, level: int) -> ViewOptions:
"""Set grouping level."""
def descending(self, descending: bool) -> ViewOptions:
"""Sort results in descending order."""
def development(self, development: bool) -> ViewOptions:
"""Query development or production views."""
def debug(self, debug: bool) -> ViewOptions:
"""Enable debug mode."""
def timeout(self, timeout: timedelta) -> ViewOptions:
"""Set query timeout."""View Results
Process view query results including individual rows and metadata.
class ViewResult:
def __iter__(self) -> Iterator[ViewRow]:
"""Iterate over view result rows."""
def rows(self) -> List[ViewRow]:
"""Get all result rows."""
def metadata(self) -> ViewMetaData:
"""Get query metadata."""
class ViewRow:
@property
def id(self) -> Optional[str]:
"""Document ID."""
@property
def key(self) -> Any:
"""View key."""
@property
def value(self) -> Any:
"""View value."""
@property
def document(self) -> Optional[dict]:
"""Full document (if include_docs=True)."""
class ViewMetaData:
@property
def total_rows(self) -> int:
"""Total number of rows in view."""
@property
def debug_info(self) -> Optional[dict]:
"""Debug information (if debug=True)."""View Index Management
Manage design documents containing view definitions.
class ViewIndexManager:
def get_design_document(self, design_doc_name: str, namespace: DesignDocumentNamespace = None,
options: GetDesignDocumentOptions = None) -> DesignDocument:
"""
Retrieve a design document.
Args:
design_doc_name (str): Name of the design document
namespace (DesignDocumentNamespace, optional): Development or production namespace
options (GetDesignDocumentOptions, optional): Additional options
Returns:
DesignDocument: The design document definition
Raises:
DesignDocumentNotFoundException: If design document not found
"""
def get_all_design_documents(self, namespace: DesignDocumentNamespace = None,
options: GetAllDesignDocumentsOptions = None) -> List[DesignDocument]:
"""Get all design documents."""
def upsert_design_document(self, design_doc: DesignDocument, namespace: DesignDocumentNamespace = None,
options: UpsertDesignDocumentOptions = None) -> None:
"""Create or update a design document."""
def drop_design_document(self, design_doc_name: str, namespace: DesignDocumentNamespace = None,
options: DropDesignDocumentOptions = None) -> None:
"""Delete a design document."""
def publish_design_document(self, design_doc_name: str, options: PublishDesignDocumentOptions = None) -> None:
"""Publish a design document from development to production."""Design Documents
Define MapReduce views within design documents.
class DesignDocument:
def __init__(self, name: str, views: Dict[str, View] = None):
"""
Create a design document.
Args:
name (str): Design document name
views (dict, optional): Dictionary of view name to View objects
"""
@property
def name(self) -> str:
"""Design document name."""
@property
def views(self) -> Dict[str, View]:
"""Dictionary of views in this design document."""
class View:
def __init__(self, map_function: str, reduce_function: str = None):
"""
Create a view definition.
Args:
map_function (str): JavaScript map function
reduce_function (str, optional): JavaScript reduce function
"""
@property
def map(self) -> str:
"""JavaScript map function."""
@property
def reduce(self) -> Optional[str]:
"""JavaScript reduce function."""
enum DesignDocumentNamespace:
DEVELOPMENT = "dev"
PRODUCTION = "prod"
enum ViewScanConsistency:
NOT_BOUNDED = "ok"
REQUEST_PLUS = "false"
UPDATE_AFTER = "update_after"Usage Examples
Basic View Query
from couchbase.cluster import Cluster
from couchbase.auth import PasswordAuthenticator
from couchbase.options import ClusterOptions, ViewOptions
# Connect to cluster
auth = PasswordAuthenticator("username", "password")
cluster = Cluster("couchbase://localhost", ClusterOptions(auth))
bucket = cluster.bucket("travel-sample")
# Query a view
view_options = ViewOptions()
view_options.limit(10)
view_options.descending(True)
result = bucket.view_query("travel", "by_country", view_options)
# Process results
for row in result:
print(f"Key: {row.key}, Value: {row.value}, Document ID: {row.id}")
# Get metadata
metadata = result.metadata()
print(f"Total rows: {metadata.total_rows}")View Query with Key Range
from couchbase.options import ViewOptions
# Query views within a key range
view_options = ViewOptions()
view_options.start_key("A")
view_options.end_key("C")
view_options.inclusive_end(False)
result = bucket.view_query("travel", "by_country", view_options)
for row in result:
print(f"Country: {row.key}, Count: {row.value}")Creating and Managing Design Documents
from couchbase.management.views import View, DesignDocument, DesignDocumentNamespace
# Create a view definition
map_function = """
function(doc, meta) {
if (doc.type === "hotel" && doc.country) {
emit(doc.country, 1);
}
}
"""
reduce_function = "_count"
# Create design document
view = View(map_function, reduce_function)
design_doc = DesignDocument("hotels", {"by_country": view})
# Get view index manager
view_mgr = bucket.view_indexes()
# Create design document in development namespace
view_mgr.upsert_design_document(design_doc, DesignDocumentNamespace.DEVELOPMENT)
# Test the view in development
dev_options = ViewOptions()
dev_options.development(True)
dev_options.limit(5)
result = bucket.view_query("hotels", "by_country", dev_options)
for row in result:
print(f"Development - Country: {row.key}, Count: {row.value}")
# Publish to production when ready
view_mgr.publish_design_document("hotels")Install with Tessl CLI
npx tessl i tessl/pypi-couchbase