Choosing an API

The Couchbase Python SDK offers both asyncio and Twisted APIs for async operation.

Asyncio

To use the Python SDK with asyncio, use the acouchbase module. As opposed to the synchronous SDK methods which wait for completion and return Result objects, the acouchbase methods return a Future. You may await the success or failure of the Future. If a Future is awaited, the method awaiting the task must have the async keyword in its signature. Result objects are obtained after await`ing the `Future as seen in the examples below.

Create an acouchbase Cluster

# needed for cluster creation
from acouchbase.cluster import Cluster
from couchbase.cluster import ClusterOptions
from couchbase.auth import PasswordAuthenticator


async def get_couchbase():
    cluster = Cluster(
        "couchbase://localhost",
        ClusterOptions(PasswordAuthenticator("Administrator", "password")))
    bucket = cluster.bucket("travel-sample")
    await bucket.on_connect()

    return cluster, bucket

Asyncio Document (KV) Operations

Key-value and sub-document operations return Future objects which can then be used for await clauses. The Future’s result will always be the relevant Result object for the operation performed.

KV - Operations

async def kv_operations(collection):
    key = "hotel_10025"
    res = await collection.get(key)
    hotel = res.content_as[dict]
    print("Hotel: {}".format(hotel))

    hotel["alias"] = "Hostel"
    res = await collection.upsert(key, hotel)
    print("CAS: {}".format(res.cas))

    # handle exception
    try:
        res = await collection.get("not-a-key")
    except DocumentNotFoundException as ex:
        print("Document not found exception caught!")

Sub-document Operations

async def sub_doc_operations(collection):
    key = "hotel_10025"
    res = await collection.lookup_in(key,
                                     [SD.get("reviews[0].ratings")])

    print("Review ratings: {}".format(res.content_as[dict](0)))

    res = await collection.mutate_in(key,
                                     [SD.replace("reviews[0].ratings.Rooms", 3.5)])
    print("CAS: {}".format(res.cas))

Asyncio N1QL Operations

The API for issuing N1QL queries is almost identical to the synchronous API. The notable difference is the use of async for rather than for when iterating over the results.

N1QL Query

async def n1ql_query(cluster):
    try:
        result = cluster.query(
            "SELECT h.* FROM `travel-sample`.inventory.hotel h WHERE h.country=$country LIMIT 10",
            QueryOptions(named_parameters={"country": "United Kingdom"}))

        async for row in result:
            print("Found row: {}".format(row))
    except ParsingFailedException as ex:
        print(ex)
    except CouchbaseException as ex:
        print(ex)

Asyncio Search Operations

The API for issuing full text search queries is almost identical to the synchronous API. The notable difference is the use of async for rather than for when iterating over the results.

Search Query

async def search_query(cluster):
    try:
        result = cluster.search_query(
            "travel-sample-index", search.QueryStringQuery("swanky"))

        async for row in result:
            print("Found row: {}".format(row))
        print("Reported total rows: {}".format(
            result.metadata().metrics.total_rows))

    except CouchbaseException as ex:
        print(ex)

Asyncio Analytics Operations

The API for issuing analytics queries is almost identical to the synchronous API. The notable difference is the use of async for rather than for when iterating over the results.

Analytics Query

async def analytics_query(cluster):
    try:
        result = cluster.analytics_query(
            "SELECT a.* FROM `travel-sample`.inventory.airports a WHERE a.country = $country LIMIT 10",
            AnalyticsOptions(named_parameters={"country": "France"}))
        async for row in result:
            print("Found row: {}".format(row))
    except CouchbaseException as ex:
        print(ex)

Twisted

To use the Python SDK with the "Twisted" framework, use the txcouchbase module. As opposed to the synchronous SDK methods which wait for completion and return Result objects, the txcouchbase methods return a Twisted Deferred. You may configure Deferred with callback and errback handlers. Result objects are propagated to the callback as seen in the examples below.

Create a txcouchbase Cluster

# create a cluster object
cluster = TxCluster(
    connection_string='couchbase://localhost',
    authenticator=PasswordAuthenticator(
        'Administrator',
        'password'))

# create a bucket object
bucket = cluster.bucket('travel-sample')

Twisted Document (KV) Operations

Key-value and sub-document operations return Deferred objects which can then be configured with callback and errback handlers. The Deferred’s result will always be the relevant Result object for the operation performed.

KV - Operations

def on_kv_ok(result):
    if isinstance(result, GetResult):
        print("Document: {}".format(result.content_as[dict]))
    else:
        print("CAS: {}".format(result.cas))


def on_kv_error(error):
    print("Operation had an error.\nError: {}".format(error))


def kv_operations(collection):
    key = "hotel_10025"
    collection.get(key).addCallback(on_kv_ok).addErrback(on_kv_error)

    new_hotel = {
        "title": "Couchbase",
        "name": "The Couchbase Hotel",
        "address": "Pennington Street",
        "directions": None,
        "phone": "+44 1457 855449",
    }

    collection.upsert("hotel_98765", new_hotel).addCallback(
        on_kv_ok).addErrback(on_kv_error)

    collection.get("not-a-key").addCallback(on_kv_ok).addErrback(on_kv_error)

Sub-document Operations

def on_sd_ok(result, idx=0):
    if isinstance(result, LookupInResult):
        print("Sub-document: {}".format(result.content_as[dict](idx)))
    else:
        print("CAS: {}".format(result.cas))


def on_sd_error(error):
    print("Operation had an error.\nError: {}".format(error))


def sub_doc_operations(collection):
    key = "hotel_10025"

    collection.lookup_in(key, [SD.get("reviews[0].ratings")]).addCallback(
        on_sd_ok).addErrback(on_sd_error)

    collection.mutate_in(key, [SD.replace("reviews[0].ratings.Rooms", 3.5)]).addCallback(
        on_sd_ok).addErrback(on_sd_error)

Twisted N1QL Operations

The API for issuing N1QL queries is almost identical to the synchronous API. The notable difference is the use of a callback in order to handle iterating over results.

N1QL Query

def handle_n1ql_results(result):
    for r in result.rows():
        print("Query row: {}".format(r))


def n1ql_query(cluster):
    cluster.query(
        "SELECT h.* FROM `travel-sample`.inventory.hotel h WHERE h.country=$country LIMIT 2",
        QueryOptions(named_parameters={"country": "United Kingdom"})).addCallback(handle_n1ql_results)

Twisted Search Operations

The API for issuing full text search queries is almost identical to the synchronous API. The notable difference is the use of a callback in order to handle iterating over results.

Search Query

def handle_search_results(result):
    for r in result.rows():
        print("Search row: {}".format(r))


def search_query(cluster):
    cluster.search_query(
        "travel-sample-index", search.QueryStringQuery("swanky")).addCallback(handle_search_results)

Twisted Analytics Operations

The API for issuing analytics queries is almost identical to the synchronous API. The notable difference is the use of a callback in order to handle iterating over results.

Analytics Query

def handle_analytics_results(result):
    for r in result.rows():
        print("Analytics row: {}".format(r))


def analytics_query(cluster):
    cluster.analytics_query(
        "SELECT id, country FROM `travel-sample`.inventory.airports a WHERE a.country = $country LIMIT 10",
        AnalyticsOptions(named_parameters={"country": "France"})).addCallback(handle_analytics_results)