sdk-connection.py

"""
Couchbase SDK connection singleton — Python
Copy and adapt for your application.
"""
from datetime import timedelta
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions, ClusterTimeoutOptions
from couchbase.auth import PasswordAuthenticator
from couchbase.exceptions import CouchbaseException

# --- Configuration ---
CB_HOST     = "localhost"          # or "cb.xxxxx.cloud.couchbase.com" for Capella
CB_USER     = "app-service-user"
CB_PASSWORD = "AppSecret123!"
CB_BUCKET   = "myapp"
CB_SCOPE    = "_default"
CB_COLLECTION = "_default"

# Use couchbases:// for TLS (required for Capella, recommended for production)
CB_SCHEME = "couchbase"  # or "couchbases"

_cluster = None
_collection = None

def get_collection():
    """Return a singleton Collection, initializing the cluster on first call."""
    global _cluster, _collection
    if _collection is not None:
        return _collection

    auth = PasswordAuthenticator(CB_USER, CB_PASSWORD)
    opts = ClusterOptions(
        auth,
        timeout_options=ClusterTimeoutOptions(
            connect_timeout=timedelta(seconds=10),
            kv_timeout=timedelta(seconds=2.5),
            query_timeout=timedelta(seconds=75),
            search_timeout=timedelta(seconds=75),
        )
    )
    _cluster = Cluster(f"{CB_SCHEME}://{CB_HOST}", opts)
    _cluster.wait_until_ready(timedelta(seconds=10))

    bucket = _cluster.bucket(CB_BUCKET)
    _collection = bucket.scope(CB_SCOPE).collection(CB_COLLECTION)
    return _collection


def get_cluster():
    """Return the singleton Cluster (needed for queries and transactions)."""
    get_collection()  # ensures cluster is initialized
    return _cluster


# --- Usage example ---
if __name__ == "__main__":
    col = get_collection()
    cluster = get_cluster()

    # KV upsert
    col.upsert("doc_1", {"type": "example", "value": 42})

    # KV get
    result = col.get("doc_1")
    print(result.content_as[dict])

    # SQL++ query
    rows = cluster.query(
        "SELECT * FROM `myapp`._default._default WHERE type = $type LIMIT 5",
        named_parameters={"type": "example"}
    )
    for row in rows:
        print(row)