from datetime import timedelta# needed for any cluster connectionfrom couchbase.auth import PasswordAuthenticatorfrom couchbase.cluster import Cluster# needed for options -- cluster, timeout, SQL++ (N1QL) query, etc.from couchbase.options import (ClusterOptions, ClusterTimeoutOptions,                               QueryOptions)# Update this to your clusterendpoint = "--your-instance--.dp.cloud.couchbase.com"username = "username"password = "Password!123"bucket_name = "travel-sample"# User Input ends here.# Connect options - authenticationauth = PasswordAuthenticator(username, password)# get a reference to our clusteroptions = ClusterOptions(auth)# Sets a pre-configured profile called "wan_development" to help avoid latency issues# when accessing Capella from a different Wide Area Network# or Availability Zone(e.g. your laptop).options.apply_profile('wan_development')cluster = Cluster('couchbases://{}'.format(endpoint), options)# Wait until the cluster is ready for use.cluster.wait_until_ready(timedelta(seconds=5))# get a reference to our bucketcb = cluster.bucket(bucket_name)cb_coll = cb.scope("inventory").collection("airline")def upsert_document(doc):    print("\nUpsert CAS: ")    try:        # key will equal: "airline_8091"        key = doc["type"] + "_" + str(doc["id"])        result = cb_coll.upsert(key, doc)        print(result.cas)    except Exception as e:        print(e)# get document functiondef get_airline_by_key(key):    print("\nGet Result: ")    try:        result = cb_coll.get(key)        print(result.content_as[str])    except Exception as e:        print(e)# query for new document by callsigndef lookup_by_callsign(cs):    print("\nLookup Result: ")    try:        inventory_scope = cb.scope('inventory')        sql_query = 'SELECT VALUE name FROM airline WHERE callsign = $1'        row_iter = inventory_scope.query(            sql_query,            QueryOptions(positional_parameters=[cs]))        for row in row_iter:            print(row)    except Exception as e:        print(e)airline = {    "type": "airline",    "id": 8091,    "callsign": "CBS",    "iata": None,    "icao": None,    "name": "Couchbase Airways",}upsert_document(airline)get_airline_by_key("airline_8091")lookup_by_callsign("CBS")

from datetime import timedelta# needed for any cluster connectionfrom couchbase.auth import PasswordAuthenticatorfrom couchbase.cluster import Cluster# needed for options -- cluster, timeout, SQL++ (N1QL) query, etc.from couchbase.options import (ClusterOptions, ClusterTimeoutOptions,                               QueryOptions)# Update this to your clusterusername = "Administrator"password = "password"bucket_name = "travel-sample"# User Input ends here.# Connect options - authenticationauth = PasswordAuthenticator(    username,    password,)# Get a reference to our cluster# NOTE: For TLS/SSL connection use 'couchbases://<your-ip-address>' insteadcluster = Cluster('couchbase://your-ip', ClusterOptions(auth))# Wait until the cluster is ready for use.cluster.wait_until_ready(timedelta(seconds=5))# get a reference to our bucketcb = cluster.bucket(bucket_name)cb_coll = cb.scope("inventory").collection("airline")# Get a reference to the default collection, required for older Couchbase server versionscb_coll_default = cb.default_collection()# upsert document functiondef upsert_document(doc):    print("\nUpsert CAS: ")    try:        # key will equal: "airline_8091"        key = doc["type"] + "_" + str(doc["id"])        result = cb_coll.upsert(key, doc)        print(result.cas)    except Exception as e:        print(e)# get document functiondef get_airline_by_key(key):    print("\nGet Result: ")    try:        result = cb_coll.get(key)        print(result.content_as[str])    except Exception as e:        print(e)# query for new document by callsigndef lookup_by_callsign(cs):    print("\nLookup Result: ")    try:        inventory_scope = cb.scope('inventory')        sql_query = 'SELECT VALUE name FROM airline WHERE callsign = $1'        row_iter = inventory_scope.query(            sql_query,            QueryOptions(positional_parameters=[cs]))        for row in row_iter:            print(row)    except Exception as e:        print(e)airline = {    "type": "airline",    "id": 8091,    "callsign": "CBS",    "iata": None,    "icao": None,    "name": "Couchbase Airways",}upsert_document(airline)get_airline_by_key("airline_8091")lookup_by_callsign("CBS")

$ brew update

$ brew install python3

$ echo 'export PATH="/usr/local/bin:"$PATH' >> ~/.zshrc

$ source ~/.zshrc

$ sudo -H python3 -m pip install couchbase

$ sudo yum install gcc gcc-c++ git python3-devel python3-pip

$ python3 -m pip install couchbase

$ sudo apt-get install git python3-dev python3-pip python3-setuptools build-essential

$ python3 -m pip install couchbase

python -m pip install couchbase

From version 4.0, the Python SDK includes Capella’s standard certificates by default, so you don’t need any additional configuration.You do need to enable TLS, which can be done by simply usingcouchbases:// in the connection string as in this example.

When accessing Capella from a different Wide Area Network or Availability Zone, you may experience latency issues with the default connection settings.SDK 4.1 introduces awan_development Configuration Profile, which provides pre-configured timeout settings suitable for working in high latency environments.Basic usage is shown in the example above, but if you want to learn more seeConstrained Network Environments.

For developing locally on the same machine as Couchbase Server, your URI can becouchbase://localhost.For production deployments, you will want to use a secure server, withcouchbases://.