N1QL Read-Your-Writes with Python 2.0

ryow.py

#!/usr/bin/env python
"""
Connect to a Couchbase 4.0 cluster and issue a N1QL query.
Note that primary index creation will fail.
"""
import sys
from time import time
from pprint import pprint

from couchbase.bucket import Bucket
from couchbase.n1ql import N1QLQuery, N1QLError

try:
    CONNECTION_STRING = sys.argv[1]
except IndexError:
    CONNECTION_STRING = "couchbase://localhost/travel-sample"

TIMEVAL = time()
bkt = Bucket(CONNECTION_STRING)

# Create the primary index first
rv = bkt.n1ql_query('CREATE PRIMARY INDEX ON `travel-sample`')
try:
    # Shorthand for executing a single statement
    rv.execute()
except N1QLError as e:
    print "Index creation failed"
    pprint(e)
    pprint(rv.raw)

bkt.upsert('airline_000', {
    'id': 0,
    'type': 'airline',
    'name': 'Couchbase Airways',
    'iata': 'CBX',
    'icao': 'CBX',
    'callsign': 'COUCH-BASE',
    'country': 'United States'
})

q = N1QLQuery('SELECT * FROM `travel-sample` WHERE type=$1 AND id=$2')

# Set the placeholders
q.add_args('airline', 0)

# We might provider a nicer wrapper for this once there's a defined API name
q.set_option('scan_consistency', 'request_plus')


for row in bkt.n1ql_query(q):
    pprint(row.raw)