How to check if Couchbase Bucket exists and programmatically create one with Python

It seems to be a simple task, but there are several gotchas with Couchbase when one tries to check  if bucket exists and create on if it does not.

There are several approaches with Python:

  • using Python’s requests module does not work due to bug in requests library  with POST authentication
  • using Python’s urllib module doesn’t do POST auth well either
  • CURL seems to be an option, but how to avoid annoying statistics block an dhow to capture the response output as JSON?

Here it is code to implement CURL  without statistics (‘-s’ parameter in curl string) and captures response by redirecting STDOUT to PIPE and reading fron it later, loading the resulting string to JSON:

import subprocess
import json

# formulate a curl request
create_bucket_bash = ['curl', '-s', '-X', 'POST', '-u', 'Administrator:premium', 'http://127.0.0.1:8091/pools/default/buckets',
                      '-d', 'name=newBucket01',
                      '-d', 'conflictResolutionType=lww',
                      '-d', 'authType=sasl',
                      '-d', 'ramQuotaMB=512',
                      '-d', 'saslPassword=passw0rd',
                      '-d', 'bucketType=couchbase']
# execute the curl request and redirect output to pipe
proc = subprocess.Popen(create_bucket_bash, stdout=subprocess.PIPE)
# read from pipe
result = proc.communicate()[0].decode()
# interpret the result. Note that in the case of success Couchbase returns nothing
if len(result) > 0:
    result = json.loads(result)
    if 'errors' in result:
        if 'name' in result['errors']:
            # Bucket exists
            print(result['errors']['name'])
        else:
            # generic error
            print(result['errors'])
else:
    # Bucket have been just created
    print('Bucket created')
  •  Done!

Leave a Reply