tessl/pypi-cloudant

IBM Cloudant Python client library providing comprehensive interface for Cloudant and CouchDB databases

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Client Connection and Authentication

Name: tessl/pypi-cloudant
Author: tessl

Establish connections to Cloudant and CouchDB servers with comprehensive authentication support including basic authentication, IAM tokens, and Cloud Foundry service integration.

Capabilities

Context Manager Functions

Context managers provide automatic connection and disconnection handling with proper resource cleanup.

def cloudant(user, passwd, **kwargs):
    """
    Create Cloudant session context manager with basic authentication.
    
    Parameters:
    - user (str): Username for authentication
    - passwd (str): Password or authentication token
    - account (str): Cloudant account name for URL construction
    - url (str): Fully qualified service URL (alternative to account)
    - x_cloudant_user (str): Override X-Cloudant-User header for proxy authentication
    - encoder (json.JSONEncoder): Custom JSON encoder for documents
    - connect (bool): Auto-connect on context entry (default: True)
    - auto_renew (bool): Automatic session renewal (default: False)
    - timeout (float | tuple): Request timeout in seconds
    - use_basic_auth (bool): Force basic authentication (default: False)
    
    Returns:
    Context manager yielding Cloudant client instance
    """

def cloudant_iam(account_name, api_key, **kwargs):
    """
    Create Cloudant session context manager with IAM authentication.
    
    Parameters:
    - account_name (str): Cloudant account name
    - api_key (str): IAM API key for authentication
    - iam_client_id (str): IAM client ID (optional)
    - iam_client_secret (str): IAM client secret (optional)
    - encoder (json.JSONEncoder): Custom JSON encoder
    - timeout (float | tuple): Request timeout
    
    Returns:
    Context manager yielding Cloudant client instance
    """

def cloudant_bluemix(vcap_services, instance_name=None, service_name=None, **kwargs):
    """
    Create Cloudant session from Cloud Foundry VCAP_SERVICES.
    
    Parameters:
    - vcap_services (dict | str): VCAP_SERVICES environment variable
    - instance_name (str): Specific service instance name
    - service_name (str): Service name in VCAP_SERVICES
    - encoder (json.JSONEncoder): Custom JSON encoder
    
    Returns:
    Context manager yielding Cloudant client instance
    """

def couchdb(user, passwd, **kwargs):
    """
    Create CouchDB session context manager.
    
    Parameters:
    - user (str): Username for authentication  
    - passwd (str): Password for authentication
    - url (str): CouchDB server URL
    - encoder (json.JSONEncoder): Custom JSON encoder
    
    Returns:
    Context manager yielding CouchDB client instance
    """

def couchdb_admin_party(**kwargs):
    """
    Create CouchDB session in Admin Party mode (no authentication).
    
    Parameters:
    - url (str): CouchDB server URL
    - encoder (json.JSONEncoder): Custom JSON encoder
    
    Returns:
    Context manager yielding CouchDB client instance
    """

Client Classes

Direct client instantiation for advanced use cases requiring manual connection management.

class CouchDB(dict):
    """
    CouchDB client with session and database management.
    """
    
    def __init__(self, user, auth_token, admin_party=False, **kwargs):
        """
        Initialize CouchDB client.
        
        Parameters:
        - user (str): Username for authentication
        - auth_token (str): Authentication token/password
        - admin_party (bool): Enable Admin Party mode
        - url (str): Server URL
        - encoder (json.JSONEncoder): Custom JSON encoder
        - adapter (requests.HTTPAdapter): Custom HTTP adapter
        - connect (bool): Auto-connect on initialization
        - auto_renew (bool): Automatic session renewal
        - timeout (float | tuple): Request timeout
        - use_basic_auth (bool): Force basic authentication
        - use_iam (bool): Use IAM authentication
        - iam_client_id (str): IAM client ID
        - iam_client_secret (str): IAM client secret
        """
    
    def connect(self):
        """
        Establish connection and authenticate session.
        
        Returns:
        None
        
        Raises:
        CloudantClientException: Authentication failure
        """
    
    def disconnect(self):
        """
        Close session and cleanup resources.
        
        Returns:
        None
        """
    
    def change_credentials(self, user=None, auth_token=None):
        """
        Update authentication credentials for active session.
        
        Parameters:
        - user (str): New username
        - auth_token (str): New authentication token
        
        Returns:
        None
        """
    
    def session(self):
        """
        Get current session information.
        
        Returns:
        dict: Session details including user context and authentication info
        """
    
    def session_cookie(self):
        """
        Get session authentication cookie.
        
        Returns:
        str: Session cookie value
        """
    
    def session_login(self, user=None, passwd=None):
        """
        Perform explicit session login.
        
        Parameters:
        - user (str): Username (uses instance user if None)
        - passwd (str): Password (uses instance token if None)
        
        Returns:
        dict: Login response with session details
        """
    
    def session_logout(self):
        """
        Perform explicit session logout.
        
        Returns:
        dict: Logout response
        """
    
    def basic_auth_str(self):
        """
        Generate basic authentication string.
        
        Returns:
        str: Base64 encoded basic auth header value
        """

class Cloudant(CouchDB):
    """
    Cloudant-specific client with additional cloud features.
    """
    
    @classmethod
    def iam(cls, account_name, api_key, **kwargs):
        """
        Create Cloudant client with IAM authentication.
        
        Parameters:
        - account_name (str): Cloudant account name
        - api_key (str): IAM API key
        - iam_client_id (str): IAM client ID (optional)
        - iam_client_secret (str): IAM client secret (optional)
        
        Returns:
        Cloudant: Configured client instance
        """
    
    @classmethod
    def bluemix(cls, vcap_services, instance_name=None, service_name=None, **kwargs):
        """
        Create Cloudant client from Cloud Foundry VCAP_SERVICES.
        
        Parameters:
        - vcap_services (dict | str): VCAP_SERVICES environment variable
        - instance_name (str): Service instance name
        - service_name (str): Service name
        
        Returns:
        Cloudant: Configured client instance
        """
    
    @property
    def is_iam_authenticated(self):
        """
        Check if using IAM authentication.
        
        Returns:
        bool: True if using IAM authentication
        """

Cloud Foundry Service Utility

class CloudFoundryService:
    """
    Parse and access Cloud Foundry service credentials.
    """
    
    def __init__(self, vcap_services, instance_name=None, service_name=None):
        """
        Parse VCAP_SERVICES for Cloudant credentials.
        
        Parameters:
        - vcap_services (dict | str): VCAP_SERVICES environment variable
        - instance_name (str): Specific service instance name
        - service_name (str): Service name in VCAP_SERVICES
        """
    
    @property
    def username(self):
        """Service username"""
        
    @property
    def password(self):
        """Service password"""
        
    @property
    def url(self):
        """Service URL"""
        
    @property
    def iam_api_key(self):
        """IAM API key if available"""

Usage Examples

Basic Authentication

from cloudant import cloudant

# Using context manager (recommended)
with cloudant('myusername', 'mypassword', account='myaccount') as client:
    print(f"Connected to: {client.server_url}")
    print(f"Session: {client.session()}")
    
    # List databases
    databases = client.all_dbs()
    print(f"Databases: {databases}")

# Manual connection management
from cloudant.client import Cloudant

client = Cloudant('myusername', 'mypassword', account='myaccount')
client.connect()
try:
    # Work with client
    pass
finally:
    client.disconnect()

IAM Authentication

from cloudant import cloudant_iam

with cloudant_iam('myaccount', 'my-iam-api-key') as client:
    print(f"IAM authenticated: {client.is_iam_authenticated}")
    databases = client.all_dbs()

Cloud Foundry Integration

import os
from cloudant import cloudant_bluemix

# Automatic parsing of VCAP_SERVICES
with cloudant_bluemix(os.getenv('VCAP_SERVICES'), 'Cloudant NoSQL DB') as client:
    databases = client.all_dbs()

# Manual service parsing
from cloudant._common_util import CloudFoundryService

service = CloudFoundryService(os.getenv('VCAP_SERVICES'), service_name='cloudantNoSQLDB')
print(f"Service URL: {service.url}")
print(f"Username: {service.username}")

CouchDB Admin Party Mode

from cloudant import couchdb_admin_party

# Connect to CouchDB in Admin Party mode (no authentication)
with couchdb_admin_party(url='http://localhost:5984') as client:
    print("Connected to CouchDB in Admin Party mode")
    
    # Admin Party mode allows unrestricted access
    databases = client.all_dbs()
    print(f"Databases: {databases}")
    
    # Create database without authentication
    db = client.create_database('admin_party_test')
    
    # Create documents without authentication
    doc = db.create_document({'message': 'Admin Party enabled'})
    print(f"Created document: {doc['_id']}")

# Manual CouchDB client with Admin Party
from cloudant.client import CouchDB

client = CouchDB(None, None, admin_party=True, url='http://localhost:5984')
client.connect()
try:
    # Work with CouchDB in Admin Party mode
    all_dbs = client.all_dbs()
finally:
    client.disconnect()

Connection Options

from cloudant import cloudant

# With custom timeout and auto-renewal
with cloudant('user', 'pass', 
              account='myaccount',
              timeout=(10, 60),  # 10s connect, 60s read
              auto_renew=True) as client:
    # Long-running operations with automatic session renewal
    pass

# With custom encoder
import json

class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        # Custom encoding logic
        return super().default(obj)

with cloudant('user', 'pass', 
              account='myaccount',
              encoder=CustomEncoder) as client:
    # Documents will use custom encoder
    pass

Error Handling

Authentication-related errors are raised as CloudantClientException:

from cloudant import cloudant
from cloudant.error import CloudantClientException

try:
    with cloudant('invalid_user', 'invalid_pass', account='myaccount') as client:
        databases = client.all_dbs()
except CloudantClientException as e:
    print(f"Authentication failed: {e}")

Install with Tessl CLI