Sample Code

    The Python SDK now lets you create users, assign them roles and associated privileges, and remove them from the system.

    User-Management APIs

    Users who have been assigned the Admin role for the cluster are able to create, edit, and remove users. The Python SDK provides APIs to support these activities. A high-level summary of the APIs can be found in User-Management.

    Python User-Management Example

    The following code-example demonstrates how the user-management APIs can be used. It assumes that Couchbase Server is established on localhost; that the Full Administrator username and password are Administrator and password respectively; and that the travel-sample bucket is installed. For information on installing the travel-sample bucket, see Sample Buckets.

    Use of the Full Administrator username and password gives an application complete access to all resources within the Couchbase Server-environment: use of the Full Administrator username and password may therefore be useful during code-development. However, for production purposes, the Full Administrator username and password should only be used when absolutely essential: in all other circumstances, the specified username and password should correspond to some lesser role, which appropriately delimits access to server-resources. Creation of such a role, and the establishment of its associated username and password, is demonstrated in the following code-example.
    import logging
    import sys
    from couchbase.cluster import Cluster, PasswordAuthenticator
    logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
    # open cluster and authenticate as Cluster Admin
    cluster = Cluster('couchbase://localhost:8091')
    print 'Authenticator as Administrator.'
    cluster.authenticate(PasswordAuthenticator('Administrator', 'password'))
    # Create a user and assign roles
    manager = cluster.cluster_manager()
    manager.user_upsert('cbtestuser', 'cbtestuserpwd', [
        # Roles required for the reading of data from the bucket
        ('data_reader', 'travel-sample'),
        ('query_select', 'travel-sample'),
        # Roles required for the writing of data into the bucket
        ('data_writer', 'travel-sample'),
        ('query_insert', 'travel-sample'),
        ('query_delete', 'travel-sample'),
        # Role required for the creation of indexes on the bucket
        ('query_manage_index', 'travel-sample')
        ], 'cbtestuser')
    print 'Listing current users.'
    users = manager.users_get().value
    for index, user in enumerate(users):
        print 'user {0}'.format(index)
        print 'user\'s name is {0}'.format(user.get('name'))
        print 'user\'s domain is {0}'.format(user.get('domain'))
        for role in user.get('roles'):
        print 'User has the role: {0}, applicable to bucket {1}'.format(role.get('role'), role.get('bucket_name'))
    print 'Authenticating as user.'
    cluster = Cluster('couchbase://localhost:8091')
    cluster.authenticate(PasswordAuthenticator('cbtestuser', 'cbtestuserpwd'))
    print 'Opening travel-sample bucket as user.'
    bucket = cluster.open_bucket('travel-sample')
    # Create a N1QL Primary Index (but ignore if one already exists).
    bucket.bucket_manager().create_n1ql_primary_index(defer=False, ignore_exists=True)
    print 'Reading out airline_10 document.'
    airline = bucket.get('airline_10')
    print 'Found: {0}'.format(airline.value)
    print 'Upserting new document as user.'
        {'callsign': 'MILE-AIR',
        'iata': 'Q5',
        'icao': 'MLA',
        'id': 11,
        'name': '40-Mile Air',
        'type': 'airline'})
    print 'Reading out airline_11 document.'
    airline = bucket.get('airline_11')
    print 'Found: {0}'.format(airline)
    print 'Performing query as user.'
    value = 'Query-results are:'
    result = bucket.n1ql_query('SELECT * FROM `travel-sample` LIMIT 5')
    for row in result:
    value += '\n\t{0}'.format(row)
    print value
    print 'Re-authenticating as administrator.'
    cluster = Cluster('couchbase://localhost:8091')
    cluster.authenticate(PasswordAuthenticator('Administrator', 'password'))
    print 'Removing user as administrator.'
    user_to_remove = 'cbtestuser'
    removed = cluster.cluster_manager().user_remove(user_to_remove)
    if removed.success:
        print 'Deleted user {0}.'.format(user_to_remove)
        print 'Could not delete user {0}.'.format(user_to_remove)