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

management-operations.mddocs/

Management Operations

Administrative operations for managing cluster resources including buckets, collections, users, roles, and indexes across all service types. These operations require appropriate administrative privileges.

Capabilities

Bucket Management

Administrative operations for managing buckets within the cluster.

class BucketManager:
    def create_bucket(self, settings: CreateBucketSettings, options: CreateBucketOptions = None) -> None:
        """
        Create a new bucket.

        Args:
            settings (CreateBucketSettings): Bucket configuration
            options (CreateBucketOptions, optional): Creation options

        Raises:
            BucketExistsException: If bucket already exists
            InvalidArgumentException: If settings are invalid
        """

    def update_bucket(self, settings: BucketSettings, options: UpdateBucketOptions = None) -> None:
        """
        Update bucket settings.

        Args:
            settings (BucketSettings): Updated bucket configuration
            options (UpdateBucketOptions, optional): Update options

        Raises:
            BucketNotFoundException: If bucket doesn't exist
        """

    def drop_bucket(self, bucket_name: str, options: DropBucketOptions = None) -> None:
        """
        Delete a bucket.

        Args:
            bucket_name (str): Name of bucket to delete
            options (DropBucketOptions, optional): Deletion options

        Raises:
            BucketNotFoundException: If bucket doesn't exist
        """

    def get_bucket(self, bucket_name: str, options: GetBucketOptions = None) -> BucketSettings:
        """
        Get bucket settings.

        Args:
            bucket_name (str): Bucket name
            options (GetBucketOptions, optional): Retrieval options

        Returns:
            BucketSettings: Current bucket configuration

        Raises:
            BucketNotFoundException: If bucket doesn't exist
        """

    def get_all_buckets(self, options: GetAllBucketsOptions = None) -> Dict[str, BucketSettings]:
        """
        Get all bucket settings.

        Args:
            options (GetAllBucketsOptions, optional): Retrieval options

        Returns:
            Dict[str, BucketSettings]: All bucket configurations
        """

    def flush_bucket(self, bucket_name: str, options: FlushBucketOptions = None) -> None:
        """
        Flush all data from bucket.

        Args:
            bucket_name (str): Bucket name
            options (FlushBucketOptions, optional): Flush options

        Raises:
            BucketNotFoundException: If bucket doesn't exist
            FeatureNotAvailableException: If flush is disabled
        """

Collection and Scope Management

Manage collections and scopes within buckets.

class CollectionManager:
    def create_scope(self, scope_name: str, options: CreateScopeOptions = None) -> None:
        """
        Create a new scope.

        Args:
            scope_name (str): Scope name
            options (CreateScopeOptions, optional): Creation options

        Raises:
            ScopeExistsException: If scope already exists
        """

    def drop_scope(self, scope_name: str, options: DropScopeOptions = None) -> None:
        """
        Delete a scope and all its collections.

        Args:
            scope_name (str): Scope name
            options (DropScopeOptions, optional): Deletion options

        Raises:
            ScopeNotFoundException: If scope doesn't exist
        """

    def create_collection(self, collection_spec: CollectionSpec, options: CreateCollectionOptions = None) -> None:
        """
        Create a new collection.

        Args:
            collection_spec (CollectionSpec): Collection configuration
            options (CreateCollectionOptions, optional): Creation options

        Raises:
            CollectionExistsException: If collection already exists
            ScopeNotFoundException: If parent scope doesn't exist
        """

    def drop_collection(self, collection_spec: CollectionSpec, options: DropCollectionOptions = None) -> None:
        """
        Delete a collection.

        Args:
            collection_spec (CollectionSpec): Collection to delete
            options (DropCollectionOptions, optional): Deletion options

        Raises:
            CollectionNotFoundException: If collection doesn't exist
        """

    def get_all_scopes(self, options: GetAllScopesOptions = None) -> List[ScopeSpec]:
        """
        Get all scopes and their collections.

        Args:
            options (GetAllScopesOptions, optional): Retrieval options

        Returns:
            List[ScopeSpec]: All scopes with their collections
        """

User and Role Management

Manage users, roles, and groups for RBAC (Role-Based Access Control).

class UserManager:
    def upsert_user(self, user: User, options: UpsertUserOptions = None) -> None:
        """
        Create or update a user.

        Args:
            user (User): User configuration
            options (UpsertUserOptions, optional): Upsert options

        Raises:
            InvalidArgumentException: If user configuration is invalid
        """

    def drop_user(self, username: str, options: DropUserOptions = None) -> None:
        """
        Delete a user.

        Args:
            username (str): Username to delete
            options (DropUserOptions, optional): Deletion options

        Raises:
            UserNotFoundException: If user doesn't exist
        """

    def get_user(self, username: str, options: GetUserOptions = None) -> UserAndMetadata:
        """
        Get user information.

        Args:
            username (str): Username
            options (GetUserOptions, optional): Retrieval options

        Returns:
            UserAndMetadata: User information with metadata

        Raises:
            UserNotFoundException: If user doesn't exist
        """

    def get_all_users(self, options: GetAllUsersOptions = None) -> List[UserAndMetadata]:
        """
        Get all users.

        Args:
            options (GetAllUsersOptions, optional): Retrieval options

        Returns:
            List[UserAndMetadata]: All users with metadata
        """

    def get_roles(self, options: GetRolesOptions = None) -> List[RoleAndDescription]:
        """
        Get available roles.

        Args:
            options (GetRolesOptions, optional): Retrieval options

        Returns:
            List[RoleAndDescription]: Available roles with descriptions
        """

    def upsert_group(self, group: Group, options: UpsertGroupOptions = None) -> None:
        """
        Create or update a group.

        Args:
            group (Group): Group configuration
            options (UpsertGroupOptions, optional): Upsert options
        """

    def drop_group(self, group_name: str, options: DropGroupOptions = None) -> None:
        """
        Delete a group.

        Args:
            group_name (str): Group name
            options (DropGroupOptions, optional): Deletion options
        """

    def get_group(self, group_name: str, options: GetGroupOptions = None) -> Group:
        """
        Get group information.

        Args:
            group_name (str): Group name
            options (GetGroupOptions, optional): Retrieval options

        Returns:
            Group: Group configuration
        """

    def get_all_groups(self, options: GetAllGroupsOptions = None) -> List[Group]:
        """
        Get all groups.

        Args:
            options (GetAllGroupsOptions, optional): Retrieval options

        Returns:
            List[Group]: All group configurations
        """

Query Index Management

Manage N1QL indexes for query performance optimization.

class QueryIndexManager:
    def create_index(self, bucket_name: str, index_name: str, keys: List[str], options: CreateQueryIndexOptions = None) -> None:
        """
        Create a N1QL index.

        Args:
            bucket_name (str): Bucket name
            index_name (str): Index name
            keys (List[str]): Index key fields
            options (CreateQueryIndexOptions, optional): Creation options

        Raises:
            IndexExistsException: If index already exists
        """

    def create_primary_index(self, bucket_name: str, options: CreatePrimaryQueryIndexOptions = None) -> None:
        """
        Create a primary index.

        Args:
            bucket_name (str): Bucket name
            options (CreatePrimaryQueryIndexOptions, optional): Creation options

        Raises:
            IndexExistsException: If primary index already exists
        """

    def drop_index(self, bucket_name: str, index_name: str, options: DropQueryIndexOptions = None) -> None:
        """
        Drop a N1QL index.

        Args:
            bucket_name (str): Bucket name
            index_name (str): Index name
            options (DropQueryIndexOptions, optional): Drop options

        Raises:
            IndexNotFoundException: If index doesn't exist
        """

    def drop_primary_index(self, bucket_name: str, options: DropPrimaryQueryIndexOptions = None) -> None:
        """
        Drop the primary index.

        Args:
            bucket_name (str): Bucket name
            options (DropPrimaryQueryIndexOptions, optional): Drop options

        Raises:
            IndexNotFoundException: If primary index doesn't exist
        """

    def get_all_indexes(self, bucket_name: str, options: GetAllQueryIndexesOptions = None) -> List[QueryIndex]:
        """
        Get all indexes for bucket.

        Args:
            bucket_name (str): Bucket name
            options (GetAllQueryIndexesOptions, optional): Retrieval options

        Returns:
            List[QueryIndex]: All indexes in bucket
        """

    def watch_indexes(self, bucket_name: str, index_names: List[str], options: WatchQueryIndexOptions = None) -> None:
        """
        Watch for indexes to come online.

        Args:
            bucket_name (str): Bucket name
            index_names (List[str]): Index names to watch
            options (WatchQueryIndexOptions, optional): Watch options

        Raises:
            TimeoutException: If indexes don't come online within timeout
        """

    def build_deferred_indexes(self, bucket_name: str, options: BuildDeferredQueryIndexOptions = None) -> None:
        """
        Build all deferred indexes.

        Args:
            bucket_name (str): Bucket name
            options (BuildDeferredQueryIndexOptions, optional): Build options
        """

Configuration Types

Bucket Configuration

class BucketSettings:
    def __init__(self, name: str, bucket_type: BucketType, 
                 ram_quota_mb: int, **kwargs):
        """
        Bucket configuration settings.

        Args:
            name (str): Bucket name
            bucket_type (BucketType): Type of bucket
            ram_quota_mb (int): RAM quota in MB
            **kwargs: Additional configuration options
        """

    @property
    def name(self) -> str: ...
    @property
    def bucket_type(self) -> BucketType: ...
    @property
    def ram_quota_mb(self) -> int: ...
    @property
    def num_replicas(self) -> int: ...
    @property
    def replica_indexes(self) -> bool: ...
    @property
    def flush_enabled(self) -> bool: ...
    @property
    def max_expiry(self) -> int: ...
    @property
    def compression_mode(self) -> CompressionMode: ...
    @property
    def conflict_resolution_type(self) -> ConflictResolutionType: ...

class CreateBucketSettings(BucketSettings):
    """Settings for creating new buckets."""

class BucketType:
    COUCHBASE = "couchbase"
    EPHEMERAL = "ephemeral"
    MEMCACHED = "memcached"

class CompressionMode:
    OFF = "off"
    PASSIVE = "passive"
    ACTIVE = "active"

class ConflictResolutionType:
    TIMESTAMP = "lww"
    SEQUENCE_NUMBER = "seqno"

Collection Configuration

class CollectionSpec:
    def __init__(self, name: str, scope_name: str = "_default", 
                 max_expiry: int = None):
        """
        Collection specification.

        Args:
            name (str): Collection name
            scope_name (str): Parent scope name (default: "_default")
            max_expiry (int, optional): Maximum document expiry in seconds
        """

    @property
    def name(self) -> str: ...
    @property
    def scope_name(self) -> str: ...
    @property
    def max_expiry(self) -> int: ...

class ScopeSpec:
    def __init__(self, name: str, collections: List[CollectionSpec] = None):
        """
        Scope specification.

        Args:
            name (str): Scope name
            collections (List[CollectionSpec], optional): Collections in scope
        """

    @property
    def name(self) -> str: ...
    @property
    def collections(self) -> List[CollectionSpec]: ...

User Configuration

class User:
    def __init__(self, username: str, display_name: str = None,
                 password: str = None, groups: List[str] = None,
                 roles: List[Role] = None):
        """
        User configuration.

        Args:
            username (str): Username
            display_name (str, optional): Display name
            password (str, optional): User password
            groups (List[str], optional): User groups
            roles (List[Role], optional): Assigned roles
        """

    @property
    def username(self) -> str: ...
    @property
    def display_name(self) -> str: ...
    @property
    def groups(self) -> List[str]: ...
    @property
    def roles(self) -> List[Role]: ...

class Role:
    def __init__(self, name: str, bucket: str = None, scope: str = None, collection: str = None):
        """
        Role definition.

        Args:
            name (str): Role name
            bucket (str, optional): Bucket restriction
            scope (str, optional): Scope restriction  
            collection (str, optional): Collection restriction
        """

    @property
    def name(self) -> str: ...
    @property
    def bucket(self) -> str: ...
    @property
    def scope(self) -> str: ...
    @property
    def collection(self) -> str: ...

class Group:
    def __init__(self, name: str, description: str = None,
                 roles: List[Role] = None, ldap_group_reference: str = None):
        """
        Group configuration.

        Args:
            name (str): Group name
            description (str, optional): Group description
            roles (List[Role], optional): Group roles
            ldap_group_reference (str, optional): LDAP group reference
        """

    @property
    def name(self) -> str: ...
    @property
    def description(self) -> str: ...
    @property
    def roles(self) -> List[Role]: ...
    @property
    def ldap_group_reference(self) -> str: ...

Index Configuration

class QueryIndex:
    @property
    def name(self) -> str:
        """Index name."""

    @property
    def bucket_name(self) -> str:
        """Bucket containing the index."""

    @property
    def scope_name(self) -> str:
        """Scope containing the index."""

    @property
    def collection_name(self) -> str:
        """Collection containing the index."""

    @property
    def is_primary(self) -> bool:
        """Whether this is a primary index."""

    @property
    def index_key(self) -> List[str]:
        """Index key fields."""

    @property
    def condition(self) -> str:
        """Index condition (WHERE clause)."""

    @property
    def state(self) -> str:
        """Index state (online, building, etc.)."""

    @property
    def type(self) -> str:
        """Index type (gsi, etc.)."""

Usage Examples

Bucket Management

from couchbase.management.buckets import BucketManager, BucketSettings, BucketType, CompressionMode
from datetime import timedelta

bucket_mgr = cluster.buckets()

# Create bucket
settings = BucketSettings(
    name="my-bucket",
    bucket_type=BucketType.COUCHBASE,
    ram_quota_mb=256,
    num_replicas=1,
    max_expiry=timedelta(days=30).total_seconds(),
    compression_mode=CompressionMode.ACTIVE
)
bucket_mgr.create_bucket(settings)

# Get bucket info
bucket_info = bucket_mgr.get_bucket("my-bucket")
print(f"Bucket RAM: {bucket_info.ram_quota_mb}MB")

# Update bucket
bucket_info.ram_quota_mb = 512
bucket_mgr.update_bucket(bucket_info)

# List all buckets
all_buckets = bucket_mgr.get_all_buckets()
for name, settings in all_buckets.items():
    print(f"Bucket: {name}, Type: {settings.bucket_type}")

Collection Management

from couchbase.management.collections import CollectionManager, CollectionSpec, ScopeSpec

coll_mgr = bucket.collections()

# Create scope
coll_mgr.create_scope("my-scope")

# Create collection
collection_spec = CollectionSpec("my-collection", "my-scope", max_expiry=3600)
coll_mgr.create_collection(collection_spec)

# List all scopes and collections
scopes = coll_mgr.get_all_scopes()
for scope in scopes:
    print(f"Scope: {scope.name}")
    for collection in scope.collections:
        print(f"  Collection: {collection.name}")

User Management

from couchbase.management.users import UserManager, User, Role

user_mgr = cluster.users()

# Create user with roles
user = User(
    username="developer",
    display_name="Developer User",
    password="secure_password",
    roles=[
        Role("bucket_admin", bucket="my-bucket"),
        Role("query_select", bucket="my-bucket")
    ]
)
user_mgr.upsert_user(user)

# Get user info
user_info = user_mgr.get_user("developer")
print(f"User: {user_info.user.display_name}")
for role in user_info.user.roles:
    print(f"  Role: {role.name} on {role.bucket}")

# List available roles
available_roles = user_mgr.get_roles()
for role in available_roles:
    print(f"Role: {role.role.name} - {role.description}")

Index Management

from couchbase.management.queries import QueryIndexManager
from datetime import timedelta

index_mgr = cluster.query_indexes()

# Create secondary index
index_mgr.create_index("my-bucket", "idx_type", ["type"])

# Create primary index
index_mgr.create_primary_index("my-bucket")

# Create conditional index
options = CreateQueryIndexOptions(
    condition="type = 'user'",
    deferred=True  # Build later
)
index_mgr.create_index("my-bucket", "idx_users", ["name", "email"], options)

# Build deferred indexes
index_mgr.build_deferred_indexes("my-bucket")

# Watch indexes come online
index_mgr.watch_indexes("my-bucket", ["idx_type", "idx_users"], 
                       WatchQueryIndexOptions(timeout=timedelta(minutes=5)))

# List all indexes
indexes = index_mgr.get_all_indexes("my-bucket")
for index in indexes:
    print(f"Index: {index.name} - State: {index.state}")

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