couchbase

import os from couchbase.cluster import Cluster, ClusterOptions from couchbase.auth import PasswordAuthenticator from couchbase.exceptions import CouchbaseException # Get connection details from environment variables for security CONNECTION_STRING = os.environ.get('CB_CONNECTION_STRING', 'couchbase://localhost') USERNAME = os.environ.get('CB_USERNAME', 'Administrator') PASSWORD = os.environ.get('CB_PASSWORD', 'password') BUCKET_NAME = os.environ.get('CB_BUCKET_NAME', 'default') def main(): cluster = None try: # Connect to Couchbase Cluster auth = PasswordAuthenticator(USERNAME, PASSWORD) cluster_options = ClusterOptions(auth) cluster = Cluster(CONNECTION_STRING, cluster_options) # Wait for the cluster to be ready (optional but good practice) # cluster.wait_until_ready(timedelta(seconds=5)) # Get a bucket and default collection bucket = cluster.bucket(BUCKET_NAME) collection = bucket.default_collection() # Store a document key = 'user:123' value = {'name': 'Alice', 'age': 30, 'city': 'New York'} result = collection.upsert(key, value) print(f"Upserted document '{key}', CAS: {result.cas}") # Retrieve the document get_result = collection.get(key) print(f"Retrieved document '{key}': {get_result.content_as[dict]}") # Update a field in the document updated_value = get_result.content_as[dict] updated_value['age'] = 31 update_result = collection.upsert(key, updated_value) print(f"Updated document '{key}', new CAS: {update_result.cas}") # Delete the document collection.remove(key) print(f"Removed document '{key}'") except CouchbaseException as e: print(f"Couchbase Error: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: if cluster: # Close the cluster connection (important for resource management) cluster.disconnect() if __name__ == '__main__': main()