Function main

examples/couchbase/couchbase_query_operations.py:18–119

()

def main():
    cluster = Cluster.connect('couchbase://localhost',
                              ClusterOptions(PasswordAuthenticator('Administrator', 'password')))
    bucket = cluster.bucket("travel-sample")
    collection = bucket.default_collection()

    # basic query
    try:
        result = cluster.query(
            "SELECT * FROM `travel-sample` LIMIT 10;", QueryOptions(metrics=True))

        for row in result.rows():
            print(f'Found row: {row}')

        metrics = result.metadata().metrics()
        print(f'Query execution time: {metrics.execution_time()}')

    except ParsingFailedException as ex:
        import traceback
        traceback.print_exc()

    # positional params
    q_str = "SELECT ts.* FROM `travel-sample` ts WHERE ts.`type`=$1 LIMIT 10"
    result = cluster.query(q_str, "hotel")
    rows = [r for r in result]

    # positional params via QueryOptions
    result = cluster.query(q_str, QueryOptions(positional_parameters=["hotel"]))
    rows = [r for r in result]

    # named params
    q_str = "SELECT ts.* FROM `travel-sample` ts WHERE ts.`type`=$doc_type LIMIT 10"
    result = cluster.query(q_str, doc_type='hotel')
    rows = [r for r in result]

    # name params via QueryOptions
    result = cluster.query(q_str, QueryOptions(named_parameters={'doc_type': 'hotel'}))
    rows = [r for r in result]

    # iterate over result/rows
    q_str = "SELECT ts.* FROM `travel-sample` ts WHERE ts.`type`='airline' LIMIT 10"
    result = cluster.query(q_str)

    # iterate over rows
    for row in result:
        # each row is an serialized JSON object
        name = row["name"]
        callsign = row["callsign"]
        print(f'Airline name: {name}, callsign: {callsign}')

    # query metrics
    result = cluster.query("SELECT 1=1", QueryOptions(metrics=True))
    # ignore results
    result.execute()

    print("Execution time: {}".format(
        result.metadata().metrics().execution_time()))