A newer version of this documentation is available.

View Latest

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)