couchbase_crud_operations_and_n1ql_query_quickstart.py

1from datetime import timedelta
2
3# needed for any cluster connection
4from couchbase.auth import PasswordAuthenticator
5from couchbase.cluster import Cluster
6# need for options objects
7from couchbase.options import (ClusterOptions, QueryOptions)
8
9# Update this to your cluster details
10auth = PasswordAuthenticator('username', 'password')
11cluster = Cluster('couchbase://localhost', ClusterOptions(auth))
12
13# Wait until the cluster is ready for use.
14cluster.wait_until_ready(timedelta(seconds=5))
15
16# get a reference to our bucket
17cb = cluster.bucket('travel-sample')
18
19# get a reference to the default collection, optionally with specific scope name
20cb_coll = cb.scope('inventory').collection('airline')
21
22# upsert document
23doc = {
24    "type": "airline",
25    "id": 8091,
26    "name": "Couchbase Airways",
27    "iata": "CB",
28    "icao": "CBA",
29    "callsign": "COUCHBASE",
30    "country": "United States"
31}
32
33try:
34    # key, value
35    result = cb_coll.upsert('airline_8091', doc)
36    print(f"Upsert Result: {result.cas}")
37except Exception as e:
38    print(e)
39
40# get document
41try:
42    result = cb_coll.get('airline_8091')
43    print(f"Get Result: {result.content_as[dict]}")
44except Exception as e:
45    print(e)
46
47# query
48try:
49    sql_query = 'SELECT name FROM `travel-sample`.inventory.airline WHERE country = $country LIMIT 10'
50    result = cluster.query(sql_query, QueryOptions(named_parameters={'country': 'United States'}))
51    for row in result:
52        print(f"Query Row: {row}")
53except Exception as e:
54    print(e)