CtrlK
BlogDocsLog inGet started
Tessl Logo

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

Overview
Eval results
Files

subdocument-operations.mddocs/

Subdocument Operations

Efficient operations on specific paths within JSON documents without retrieving or replacing entire documents. Enables atomic mutations and lookups on document fragments for improved performance and reduced network overhead.

Capabilities

Document Path Operations

Perform operations on specific JSON paths within documents.

class CBCollection:
    def lookup_in(self, key: str, spec: List[Spec], options: LookupInOptions = None) -> LookupInResult:
        """
        Perform subdocument lookup operations.

        Args:
            key (str): Document key
            spec (List[Spec]): List of lookup specifications
            options (LookupInOptions, optional): Lookup options

        Returns:
            LookupInResult: Results for each lookup operation

        Raises:
            DocumentNotFoundException: If document doesn't exist
            PathNotFoundException: If specified path doesn't exist
        """

    def mutate_in(self, key: str, spec: List[Spec], options: MutateInOptions = None) -> MutateInResult:
        """
        Perform subdocument mutation operations.

        Args:
            key (str): Document key
            spec (List[Spec]): List of mutation specifications
            options (MutateInOptions, optional): Mutation options

        Returns:
            MutateInResult: Results for each mutation operation

        Raises:
            DocumentNotFoundException: If document doesn't exist
            PathExistsException: If path already exists (for insert operations)
            PathNotFoundException: If path doesn't exist (for replace operations)
        """

Lookup Specifications

Specify paths and operations for document lookups.

import couchbase.subdocument as SD

class Spec:
    @staticmethod
    def get(path: str, xattr: bool = False) -> Spec:
        """
        Get value at path.

        Args:
            path (str): JSON path
            xattr (bool): Whether path refers to extended attribute

        Returns:
            Spec: Lookup specification
        """

    @staticmethod
    def exists(path: str, xattr: bool = False) -> Spec:
        """
        Check if path exists.

        Args:
            path (str): JSON path
            xattr (bool): Whether path refers to extended attribute

        Returns:
            Spec: Existence check specification
        """

    @staticmethod
    def count(path: str, xattr: bool = False) -> Spec:
        """
        Count elements in array at path.

        Args:
            path (str): JSON path to array
            xattr (bool): Whether path refers to extended attribute

        Returns:
            Spec: Count specification
        """

    @staticmethod
    def get_full() -> Spec:
        """
        Get entire document.

        Returns:
            Spec: Full document retrieval specification
        """

Mutation Specifications

Specify paths and operations for document mutations.

class Spec:
    @staticmethod
    def replace(path: str, value: Any, xattr: bool = False) -> Spec:
        """
        Replace value at path.

        Args:
            path (str): JSON path
            value (Any): New value
            xattr (bool): Whether path refers to extended attribute

        Returns:
            Spec: Replace specification
        """

    @staticmethod
    def upsert(path: str, value: Any, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Insert or replace value at path.

        Args:
            path (str): JSON path
            value (Any): Value to set
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create intermediate paths if needed

        Returns:
            Spec: Upsert specification
        """

    @staticmethod
    def insert(path: str, value: Any, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Insert value at path (must not exist).

        Args:
            path (str): JSON path
            value (Any): Value to insert
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create intermediate paths if needed

        Returns:
            Spec: Insert specification
        """

    @staticmethod
    def remove(path: str, xattr: bool = False) -> Spec:
        """
        Remove value at path.

        Args:
            path (str): JSON path
            xattr (bool): Whether path refers to extended attribute

        Returns:
            Spec: Remove specification
        """

Array Operations

Specialized operations for working with JSON arrays.

class Spec:
    @staticmethod
    def array_append(path: str, *values: Any, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Append values to end of array.

        Args:
            path (str): JSON path to array
            *values: Values to append
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create array if path doesn't exist

        Returns:
            Spec: Array append specification
        """

    @staticmethod
    def array_prepend(path: str, *values: Any, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Prepend values to beginning of array.

        Args:
            path (str): JSON path to array
            *values: Values to prepend
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create array if path doesn't exist

        Returns:
            Spec: Array prepend specification
        """

    @staticmethod
    def array_insert(path: str, *values: Any, xattr: bool = False) -> Spec:
        """
        Insert values at specific array index.

        Args:
            path (str): JSON path with array index (e.g., "items[2]")
            *values: Values to insert
            xattr (bool): Whether path refers to extended attribute

        Returns:
            Spec: Array insert specification
        """

    @staticmethod
    def array_add_unique(path: str, value: Any, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Add value to array if not already present.

        Args:
            path (str): JSON path to array
            value (Any): Value to add
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create array if path doesn't exist

        Returns:
            Spec: Array add unique specification
        """

Counter Operations

Atomic counter operations on numeric values within documents.

class Spec:
    @staticmethod
    def increment(path: str, delta: int = 1, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Increment numeric value at path.

        Args:
            path (str): JSON path to numeric value
            delta (int): Increment amount (default: 1)
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create path with initial value if needed

        Returns:
            Spec: Increment specification
        """

    @staticmethod
    def decrement(path: str, delta: int = 1, xattr: bool = False, create_path: bool = False) -> Spec:
        """
        Decrement numeric value at path.

        Args:
            path (str): JSON path to numeric value
            delta (int): Decrement amount (default: 1)
            xattr (bool): Whether path refers to extended attribute
            create_path (bool): Create path with initial value if needed

        Returns:
            Spec: Decrement specification
        """

Operation Options

class LookupInOptions:
    def __init__(self, timeout: timedelta = None,
                 access_deleted: bool = False):
        """
        Options for subdocument lookup operations.

        Args:
            timeout (timedelta, optional): Operation timeout
            access_deleted (bool): Access tombstoned (deleted) documents
        """

class MutateInOptions:
    def __init__(self, timeout: timedelta = None,
                 expiry: timedelta = None,
                 durability: Durability = None,
                 cas: int = None,
                 upsert_document: bool = False,
                 access_deleted: bool = False):
        """
        Options for subdocument mutation 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
            upsert_document (bool): Create document if it doesn't exist
            access_deleted (bool): Access tombstoned (deleted) documents
        """

Result Types

class LookupInResult:
    def content_as(self, index: int, target_type: type):
        """
        Get content of lookup operation at index.

        Args:
            index (int): Operation index
            target_type (type): Target type for content

        Returns:
            Content converted to target type
        """

    def exists(self, index: int) -> bool:
        """Check if path exists for operation at index."""

    @property
    def cas(self) -> int:
        """Document CAS value."""

class MutateInResult:
    def content_as(self, index: int, target_type: type):
        """
        Get content of mutation operation at index.

        Args:
            index (int): Operation index
            target_type (type): Target type for content

        Returns:
            Content converted to target type
        """

    @property
    def cas(self) -> int:
        """New document CAS value."""

    @property
    def mutation_token(self) -> MutationToken:
        """Mutation token for consistency."""

Usage Examples

Basic Subdocument Operations

import couchbase.subdocument as SD

# Document structure
doc = {
    "name": "John Doe",
    "age": 30,
    "address": {
        "street": "123 Main St",
        "city": "San Francisco"
    },
    "hobbies": ["reading", "cycling"]
}

collection.upsert("user::123", doc)

# Lookup specific fields
result = collection.lookup_in("user::123", [
    SD.get("name"),
    SD.get("address.city"),
    SD.exists("phone"),
    SD.count("hobbies")
])

name = result.content_as(0, str)
city = result.content_as(1, str)
has_phone = result.exists(2)
hobby_count = result.content_as(3, int)

print(f"Name: {name}, City: {city}")
print(f"Has phone: {has_phone}, Hobbies: {hobby_count}")

Subdocument Mutations

# Update specific fields
collection.mutate_in("user::123", [
    SD.replace("age", 31),
    SD.upsert("address.zipcode", "94105"),
    SD.array_append("hobbies", "photography"),
    SD.increment("login_count", 1)
])

# Insert new nested object
collection.mutate_in("user::123", [
    SD.insert("preferences", {"theme": "dark", "notifications": True})
])

# Remove field
collection.mutate_in("user::123", [
    SD.remove("temporary_field")
])

Array Operations

# Working with arrays
collection.mutate_in("user::123", [
    SD.array_append("hobbies", "gaming", "cooking"),
    SD.array_prepend("hobbies", "traveling"),
    SD.array_insert("hobbies[2]", "swimming"),
    SD.array_add_unique("tags", "vip")
])

# Check array contents
result = collection.lookup_in("user::123", [
    SD.get("hobbies"),
    SD.count("hobbies"),
    SD.get("hobbies[0]")  # First element
])

all_hobbies = result.content_as(0, list)
hobby_count = result.content_as(1, int)
first_hobby = result.content_as(2, str)

Counter Operations

# Initialize counters
collection.mutate_in("stats::global", [
    SD.upsert("page_views", 0, create_path=True),
    SD.upsert("user_count", 100, create_path=True)
], MutateInOptions(upsert_document=True))

# Increment counters atomically
collection.mutate_in("stats::global", [
    SD.increment("page_views", 1),
    SD.increment("user_count", 1),
    SD.increment("api_calls", 5)
])

# Get current values
result = collection.lookup_in("stats::global", [
    SD.get("page_views"),
    SD.get("user_count")
])

views = result.content_as(0, int)
users = result.content_as(1, int)

Document Creation with Subdocument

# Create document using subdocument operations
collection.mutate_in("user::456", [
    SD.upsert("name", "Alice Smith"),
    SD.upsert("profile.bio", "Software engineer"),
    SD.upsert("profile.skills", ["Python", "JavaScript"]),
    SD.upsert("stats.login_count", 0)
], MutateInOptions(upsert_document=True))

Extended Attributes (XAttrs)

# Work with extended attributes (metadata)
collection.mutate_in("user::123", [
    SD.upsert("_metadata.created_by", "system", xattr=True),
    SD.upsert("_metadata.version", 1, xattr=True),
    SD.replace("name", "John Smith")  # Regular document update
])

# Lookup extended attributes
result = collection.lookup_in("user::123", [
    SD.get("_metadata", xattr=True),
    SD.get("name")
])

metadata = result.content_as(0, dict)
name = result.content_as(1, str)

Error Handling

from couchbase.exceptions import PathNotFoundException, PathExistsException

try:
    collection.mutate_in("user::123", [
        SD.replace("nonexistent.field", "value"),
        SD.insert("existing.field", "new_value")
    ])
except PathNotFoundException as e:
    print(f"Path not found: {e}")
except PathExistsException as e:
    print(f"Path already exists: {e}")

Install with Tessl CLI

npx tessl i tessl/pypi-couchbase

docs

analytics-operations.md

async-operations.md

cluster-operations.md

document-operations.md

index.md

management-operations.md

n1ql-queries.md

search-operations.md

subdocument-operations.md

view-operations.md

tile.json