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
document-operations.mddocs/
Document Operations
Core key-value operations for creating, reading, updating, and deleting documents in Couchbase collections. These operations form the foundation of data manipulation in Couchbase.
Capabilities
Basic Document Operations
Fundamental CRUD operations for document management.
class CBCollection:
def get(self, key: str, options: GetOptions = None) -> GetResult:
"""
Retrieve a document by key.
Args:
key (str): Document key
options (GetOptions, optional): Retrieval options
Returns:
GetResult: Document content and metadata
Raises:
DocumentNotFoundException: If document doesn't exist
"""
def upsert(self, key: str, value: Any, options: UpsertOptions = None) -> MutationResult:
"""
Insert or update a document.
Args:
key (str): Document key
value (Any): Document content
options (UpsertOptions, optional): Upsert options
Returns:
MutationResult: Operation result with CAS and mutation token
Raises:
TimeoutException: If operation times out
"""
def insert(self, key: str, value: Any, options: InsertOptions = None) -> MutationResult:
"""
Insert a new document.
Args:
key (str): Document key
value (Any): Document content
options (InsertOptions, optional): Insert options
Returns:
MutationResult: Operation result with CAS and mutation token
Raises:
DocumentExistsException: If document already exists
"""
def replace(self, key: str, value: Any, options: ReplaceOptions = None) -> MutationResult:
"""
Replace an existing document.
Args:
key (str): Document key
value (Any): New document content
options (ReplaceOptions, optional): Replace options
Returns:
MutationResult: Operation result with CAS and mutation token
Raises:
DocumentNotFoundException: If document doesn't exist
CASMismatchException: If CAS value doesn't match
"""
def remove(self, key: str, options: RemoveOptions = None) -> MutationResult:
"""
Remove a document.
Args:
key (str): Document key
options (RemoveOptions, optional): Remove options
Returns:
MutationResult: Operation result with CAS and mutation token
Raises:
DocumentNotFoundException: If document doesn't exist
CASMismatchException: If CAS value doesn't match
"""Document Metadata Operations
Operations for working with document metadata without retrieving content.
class CBCollection:
def exists(self, key: str, options: ExistsOptions = None) -> ExistsResult:
"""
Check if a document exists.
Args:
key (str): Document key
options (ExistsOptions, optional): Existence check options
Returns:
ExistsResult: Existence status and CAS value
"""
def touch(self, key: str, expiry: timedelta, options: TouchOptions = None) -> MutationResult:
"""
Update document expiration time.
Args:
key (str): Document key
expiry (timedelta): New expiration time
options (TouchOptions, optional): Touch options
Returns:
MutationResult: Operation result with new CAS
Raises:
DocumentNotFoundException: If document doesn't exist
"""
def get_and_touch(self, key: str, expiry: timedelta, options: GetAndTouchOptions = None) -> GetResult:
"""
Retrieve document and update expiration.
Args:
key (str): Document key
expiry (timedelta): New expiration time
options (GetAndTouchOptions, optional): Operation options
Returns:
GetResult: Document content with updated CAS
Raises:
DocumentNotFoundException: If document doesn't exist
"""
def get_and_lock(self, key: str, lock_time: timedelta, options: GetAndLockOptions = None) -> GetResult:
"""
Retrieve and lock a document.
Args:
key (str): Document key
lock_time (timedelta): Lock duration
options (GetAndLockOptions, optional): Lock options
Returns:
GetResult: Document content with lock CAS
Raises:
DocumentNotFoundException: If document doesn't exist
DocumentLockedException: If document is already locked
"""
def unlock(self, key: str, cas: int, options: UnlockOptions = None) -> None:
"""
Unlock a locked document.
Args:
key (str): Document key
cas (int): CAS value from get_and_lock
options (UnlockOptions, optional): Unlock options
Raises:
DocumentNotFoundException: If document doesn't exist
CASMismatchException: If CAS value doesn't match
"""
def get_any_replica(self, key: str, options: GetAnyReplicaOptions = None) -> GetReplicaResult:
"""
Retrieve a document from any available replica.
Args:
key (str): Document key
options (GetAnyReplicaOptions, optional): Replica retrieval options
Returns:
GetReplicaResult: Document content from replica with replica flag
Raises:
DocumentNotFoundException: If document doesn't exist on any replica
"""
def get_all_replicas(self, key: str, options: GetAllReplicasOptions = None) -> List[GetReplicaResult]:
"""
Retrieve a document from all available replicas.
Args:
key (str): Document key
options (GetAllReplicasOptions, optional): Replica retrieval options
Returns:
List[GetReplicaResult]: Document content from all replicas
Raises:
DocumentNotFoundException: If document doesn't exist
"""Binary Operations
Operations for working with binary data and counter documents.
class BinaryCollection:
def append(self, key: str, value: bytes, options: AppendOptions = None) -> MutationResult:
"""
Append binary data to document.
Args:
key (str): Document key
value (bytes): Binary data to append
options (AppendOptions, optional): Append options
Returns:
MutationResult: Operation result
Raises:
DocumentNotFoundException: If document doesn't exist
"""
def prepend(self, key: str, value: bytes, options: PrependOptions = None) -> MutationResult:
"""
Prepend binary data to document.
Args:
key (str): Document key
value (bytes): Binary data to prepend
options (PrependOptions, optional): Prepend options
Returns:
MutationResult: Operation result
Raises:
DocumentNotFoundException: If document doesn't exist
"""
def increment(self, key: str, options: IncrementOptions = None) -> CounterResult:
"""
Increment a counter document.
Args:
key (str): Counter document key
options (IncrementOptions, optional): Increment options
Returns:
CounterResult: New counter value and CAS
Raises:
DocumentNotFoundException: If document doesn't exist and no initial value set
"""
def decrement(self, key: str, options: DecrementOptions = None) -> CounterResult:
"""
Decrement a counter document.
Args:
key (str): Counter document key
options (DecrementOptions, optional): Decrement options
Returns:
CounterResult: New counter value and CAS
Raises:
DocumentNotFoundException: If document doesn't exist and no initial value set
"""Multi-Document Operations
Batch operations for improved performance when working with multiple documents.
class CBCollection:
def get_multi(self, keys: List[str], options: GetOptions = None) -> MultiGetResult:
"""
Retrieve multiple documents.
Args:
keys (List[str]): List of document keys
options (GetOptions, optional): Retrieval options
Returns:
MultiGetResult: Results for each key
"""
def upsert_multi(self, docs: Dict[str, Any], options: UpsertOptions = None) -> MultiMutationResult:
"""
Upsert multiple documents.
Args:
docs (Dict[str, Any]): Key-value pairs to upsert
options (UpsertOptions, optional): Upsert options
Returns:
MultiMutationResult: Results for each key
"""
def remove_multi(self, keys: List[str], options: RemoveOptions = None) -> MultiMutationResult:
"""
Remove multiple documents.
Args:
keys (List[str]): List of document keys to remove
options (RemoveOptions, optional): Remove options
Returns:
MultiMutationResult: Results for each key
"""Operation Options
Basic Operation Options
class GetOptions:
def __init__(self, timeout: timedelta = None,
with_expiry: bool = False,
project: List[str] = None):
"""
Options for get operations.
Args:
timeout (timedelta, optional): Operation timeout
with_expiry (bool): Include expiration time in result
project (List[str], optional): Fields to project
"""
class UpsertOptions:
def __init__(self, timeout: timedelta = None,
expiry: timedelta = None,
durability: Durability = None,
cas: int = None):
"""
Options for upsert operations.
Args:
timeout (timedelta, optional): Operation timeout
expiry (timedelta, optional): Document expiration
durability (Durability, optional): Durability requirements
cas (int, optional): CAS value for optimistic locking
"""
class InsertOptions:
def __init__(self, timeout: timedelta = None,
expiry: timedelta = None,
durability: Durability = None):
"""
Options for insert operations.
Args:
timeout (timedelta, optional): Operation timeout
expiry (timedelta, optional): Document expiration
durability (Durability, optional): Durability requirements
"""
class ReplaceOptions:
def __init__(self, timeout: timedelta = None,
expiry: timedelta = None,
durability: Durability = None,
cas: int = None):
"""
Options for replace operations.
Args:
timeout (timedelta, optional): Operation timeout
expiry (timedelta, optional): Document expiration
durability (Durability, optional): Durability requirements
cas (int, optional): CAS value for optimistic locking
"""
class RemoveOptions:
def __init__(self, timeout: timedelta = None,
durability: Durability = None,
cas: int = None):
"""
Options for remove operations.
Args:
timeout (timedelta, optional): Operation timeout
durability (Durability, optional): Durability requirements
cas (int, optional): CAS value for optimistic locking
"""Counter Operation Options
class IncrementOptions:
def __init__(self, timeout: timedelta = None,
expiry: timedelta = None,
durability: Durability = None,
delta: int = 1,
initial: int = None):
"""
Options for increment operations.
Args:
timeout (timedelta, optional): Operation timeout
expiry (timedelta, optional): Document expiration
durability (Durability, optional): Durability requirements
delta (int): Increment amount (default: 1)
initial (int, optional): Initial value if document doesn't exist
"""
class DecrementOptions:
def __init__(self, timeout: timedelta = None,
expiry: timedelta = None,
durability: Durability = None,
delta: int = 1,
initial: int = None):
"""
Options for decrement operations.
Args:
timeout (timedelta, optional): Operation timeout
expiry (timedelta, optional): Document expiration
durability (Durability, optional): Durability requirements
delta (int): Decrement amount (default: 1)
initial (int, optional): Initial value if document doesn't exist
"""Result Types
class GetResult:
@property
def content_as(self) -> ContentProxy:
"""Access document content with type conversion."""
@property
def cas(self) -> int:
"""Document CAS value."""
@property
def expiry_time(self) -> datetime:
"""Document expiration time (if requested)."""
class MutationResult:
@property
def cas(self) -> int:
"""New CAS value after mutation."""
@property
def mutation_token(self) -> MutationToken:
"""Mutation token for consistency."""
class ExistsResult:
@property
def exists(self) -> bool:
"""Whether document exists."""
@property
def cas(self) -> int:
"""Document CAS value if exists."""
class CounterResult(MutationResult):
@property
def content(self) -> int:
"""New counter value."""
class GetReplicaResult:
@property
def content_as(self) -> ContentProxy:
"""Access document content with type conversion."""
@property
def cas(self) -> int:
"""Document CAS value."""
@property
def is_replica(self) -> bool:
"""True if document came from replica, False if from active."""
class MultiGetResult:
def __getitem__(self, key: str) -> GetResult:
"""Get result for specific key."""
def __iter__(self) -> Iterator[Tuple[str, GetResult]]:
"""Iterate over key-result pairs."""
def __len__(self) -> int:
"""Number of results."""
class MultiMutationResult:
def __getitem__(self, key: str) -> MutationResult:
"""Get mutation result for specific key."""
def __iter__(self) -> Iterator[Tuple[str, MutationResult]]:
"""Iterate over key-result pairs."""
def __len__(self) -> int:
"""Number of results."""
class ContentProxy:
def __getitem__(self, key: type):
"""Get content as specific type (dict, list, str, etc.)."""Usage Examples
Basic Operations
from datetime import timedelta
# Basic get/upsert
doc = {"name": "John", "age": 30}
result = collection.upsert("user::123", doc)
print(f"CAS: {result.cas}")
get_result = collection.get("user::123")
print(f"Name: {get_result.content_as[dict]['name']}")
# With expiration
collection.upsert("temp::data", {"value": 42},
UpsertOptions(expiry=timedelta(hours=1)))
# Conditional replace
try:
collection.replace("user::123", {"name": "Jane", "age": 25},
ReplaceOptions(cas=get_result.cas))
except CASMismatchException:
print("Document was modified by another process")Counter Operations
# Initialize counter
collection.binary().increment("counter::page_views",
IncrementOptions(initial=0))
# Increment by 5
result = collection.binary().increment("counter::page_views",
IncrementOptions(delta=5))
print(f"New value: {result.content}")Multi-Document Operations
# Batch get
keys = ["user::1", "user::2", "user::3"]
results = collection.get_multi(keys)
for key, result in results.results.items():
if result.success:
print(f"{key}: {result.content_as[dict]}")
# Batch upsert
docs = {
"user::100": {"name": "Alice", "age": 28},
"user::101": {"name": "Bob", "age": 35}
}
collection.upsert_multi(docs)Install with Tessl CLI
npx tessl i tessl/pypi-couchbase