or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

analytics-operations.mdasync-operations.mdcluster-operations.mddocument-operations.mdindex.mdmanagement-operations.mdn1ql-queries.mdsearch-operations.mdsubdocument-operations.mdview-operations.md
tile.json

tessl/pypi-couchbase

Python Client for Couchbase providing comprehensive database operations including key-value, N1QL queries, search, analytics, and cluster management

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/couchbase@3.2.x

To install, run

npx @tessl/cli install tessl/pypi-couchbase@3.2.0

index.mddocs/

Couchbase Python Client

A comprehensive Python client library for Couchbase, a distributed NoSQL document database. The SDK enables developers to connect to and interact with Couchbase Server clusters, offering both synchronous and asynchronous operations through asyncio and Twisted frameworks. It supports key-value operations, SQL++ (N1QL) queries, full-text search, analytics, and management operations including bucket, user, and index management.

Package Information

  • Package Name: couchbase
  • Language: Python
  • Installation: pip install couchbase

Core Imports

from couchbase.cluster import Cluster
from couchbase.auth import PasswordAuthenticator
from couchbase.options import ClusterOptions

Asynchronous operations:

from acouchbase.cluster import ACluster

Twisted framework:

from txcouchbase.cluster import TxCluster

Basic Usage

from couchbase.cluster import Cluster
from couchbase.auth import PasswordAuthenticator
from couchbase.options import ClusterOptions
import couchbase.subdocument as SD

# Connect to cluster
auth = PasswordAuthenticator("username", "password")
cluster = Cluster("couchbase://localhost", ClusterOptions(auth))

# Get bucket and collection
bucket = cluster.bucket("travel-sample")
collection = bucket.default_collection()

# Basic document operations
doc = {"name": "John Doe", "age": 30, "city": "San Francisco"}
result = collection.upsert("user::123", doc)
print(f"CAS: {result.cas}")

# Retrieve document
get_result = collection.get("user::123")
print(f"Document: {get_result.content_as[dict]}")

# N1QL query
query = "SELECT name, age FROM `travel-sample` WHERE type = 'user' LIMIT 10"
query_result = cluster.query(query)
for row in query_result:
    print(row)

Architecture

The Couchbase Python client follows a hierarchical structure:

  • Cluster: Top-level connection and management interface
  • Bucket: Database-level operations and container for scopes
  • Scope: Logical grouping of collections (default scope: "_default")
  • Collection: Document container with key-value and subdocument operations
  • Management APIs: Administrative operations for buckets, users, indexes, etc.

The client supports multiple operation modes:

  • Synchronous: Standard blocking operations (couchbase module)
  • Asynchronous: asyncio-based non-blocking operations (acouchbase module)
  • Twisted: Twisted framework integration (txcouchbase module)

Capabilities

Cluster Connection and Authentication

Essential cluster connection, authentication, and configuration management. Supports multiple authentication methods including RBAC, certificate-based authentication, and connection pooling.

class Cluster:
    def __init__(self, connection_string: str, options: ClusterOptions = None): ...
    def bucket(self, bucket_name: str) -> Bucket: ...
    def query(self, statement: str, options: QueryOptions = None) -> QueryResult: ...

Cluster Operations

Document Operations

Core key-value operations for creating, reading, updating, and deleting documents. Includes support for atomic operations, durability requirements, and bulk operations.

class CBCollection:
    def get(self, key: str, options: GetOptions = None) -> GetResult: ...
    def upsert(self, key: str, value: Any, options: UpsertOptions = None) -> MutationResult: ...
    def insert(self, key: str, value: Any, options: InsertOptions = None) -> MutationResult: ...
    def replace(self, key: str, value: Any, options: ReplaceOptions = None) -> MutationResult: ...
    def remove(self, key: str, options: RemoveOptions = None) -> MutationResult: ...

Document Operations

Subdocument Operations

Efficient operations on specific paths within JSON documents without retrieving or replacing entire documents. Supports atomic mutations and lookups on document fragments.

class CBCollection:
    def lookup_in(self, key: str, spec: List[Spec], options: LookupInOptions = None) -> LookupInResult: ...
    def mutate_in(self, key: str, spec: List[Spec], options: MutateInOptions = None) -> MutateInResult: ...

Subdocument Operations

N1QL Queries

SQL++ (N1QL) query execution with support for prepared statements, parameterized queries, and various consistency levels. Enables complex data analysis and retrieval operations.

class Cluster:
    def query(self, statement: str, options: QueryOptions = None) -> QueryResult: ...

class QueryResult:
    def __iter__(self) -> Iterator[dict]: ...
    def metadata(self) -> QueryMetaData: ...

N1QL Queries

Full-Text Search

Advanced search capabilities with support for complex queries, faceting, sorting, and highlighting. Enables powerful text search across document collections.

class Cluster:
    def search_query(self, index: str, query: SearchQuery, options: SearchOptions = None) -> SearchResult: ...

class SearchQuery:
    @staticmethod
    def match(match: str) -> MatchQuery: ...
    @staticmethod
    def term(term: str) -> TermQuery: ...

Full-Text Search

Analytics

Analytics query execution for complex data analysis and reporting. Supports large-scale analytical workloads with integration to external data sources.

class Cluster:
    def analytics_query(self, statement: str, options: AnalyticsOptions = None) -> AnalyticsResult: ...

class AnalyticsResult:
    def __iter__(self) -> Iterator[dict]: ...
    def metadata(self) -> AnalyticsMetaData: ...

Analytics Operations

Management Operations

Administrative operations for managing cluster resources including buckets, collections, users, roles, and indexes across all service types.

class Cluster:
    def buckets(self) -> BucketManager: ...
    def users(self) -> UserManager: ...
    def query_indexes(self) -> QueryIndexManager: ...
    def analytics_indexes(self) -> AnalyticsIndexManager: ...
    def search_indexes(self) -> SearchIndexManager: ...
    def eventing_functions(self) -> EventingFunctionManager: ...

Management Operations

View Operations

Traditional MapReduce view queries for data indexing and querying. Provides compatibility with legacy Couchbase applications using design documents.

class Bucket:
    def view_query(self, design_doc: str, view_name: str, options: ViewOptions = None) -> ViewResult: ...
    def view_indexes(self) -> ViewIndexManager: ...

View Operations

Asynchronous Operations

Asyncio-based asynchronous operations for high-performance, non-blocking applications. Provides the same API surface as synchronous operations with async/await support.

class ACluster:
    async def bucket(self, bucket_name: str) -> ABucket: ...
    async def query(self, statement: str, options: QueryOptions = None) -> QueryResult: ...

class AsyncCBCollection:
    async def get(self, key: str, options: GetOptions = None) -> AsyncGetResult: ...
    async def upsert(self, key: str, value: Any, options: UpsertOptions = None) -> AsyncMutationResult: ...

Asynchronous Operations

Common Types

class ClusterOptions:
    def __init__(self, authenticator: Authenticator = None): ...

class PasswordAuthenticator:
    def __init__(self, username: str, password: str): ...

class GetResult:
    @property
    def content_as(self) -> ContentProxy: ...
    @property
    def cas(self) -> int: ...

class MutationResult:
    @property
    def cas(self) -> int: ...
    @property
    def mutation_token(self) -> MutationToken: ...

class ContentProxy:
    def __getitem__(self, t: Type[T]) -> T: ...

class MutationToken:
    @property
    def bucket_name(self) -> str: ...
    @property
    def partition_id(self) -> int: ...
    @property
    def partition_uuid(self) -> int: ...
    @property
    def sequence_number(self) -> int: ...

enum Durability:
    NONE = 0
    MAJORITY = 1
    MAJORITY_AND_PERSIST_TO_ACTIVE = 2
    PERSIST_TO_MAJORITY = 3

enum ServiceType:
    KV = "kv"
    QUERY = "n1ql"
    SEARCH = "fts"
    ANALYTICS = "cbas"
    MANAGEMENT = "mgmt"
    VIEWS = "views"
    EVENTING = "eventing"

class CouchbaseException(Exception): ...
class DocumentNotFoundException(CouchbaseException): ...
class DocumentExistsException(CouchbaseException): ...