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

Cloudant Python Client Library

Name: tessl/pypi-cloudant
Author: tessl

IBM Cloudant Python client library providing comprehensive interface for Cloudant and CouchDB databases. This library offers high-level abstractions for database operations including document management, indexing, querying, replication, and change feed monitoring with robust session management and authentication support.

Note: This library has been deprecated as of December 31, 2021. Users are encouraged to migrate to the newer cloudant-python-sdk for continued support.

Package Information

Package Name: cloudant
Language: Python
Version: 2.15.0
Installation: pip install cloudant
License: Apache-2.0

Core Imports

from cloudant import cloudant, cloudant_iam, couchdb
from cloudant.client import Cloudant, CouchDB
from cloudant.database import CloudantDatabase, CouchDatabase
from cloudant.document import Document
from cloudant.security_document import SecurityDocument
from cloudant.scheduler import Scheduler
from cloudant.adapters import Replay429Adapter
from cloudant.result import Result, QueryResult, ResultByKey
from cloudant.error import CloudantException

Basic Usage

Context Manager Pattern (Recommended)

from cloudant import cloudant

# Basic authentication
with cloudant(username, password, account=account_name) as client:
    # Access databases
    db = client['my_database']
    
    # Work with documents
    doc = db['my_document']
    doc['field'] = 'value'
    doc.save()
    
    # Query documents
    for document in db:
        print(document['_id'])

# IAM authentication  
from cloudant import cloudant_iam

with cloudant_iam(account_name, api_key) as client:
    db = client['my_database']
    # ... work with database

Direct Client Usage

from cloudant.client import Cloudant

client = Cloudant(username, password, account=account_name)
client.connect()

try:
    # Create database
    db = client.create_database('my_database')
    
    # Create document
    doc = db.create_document({'name': 'John', 'age': 30})
    
    # Query documents
    result = db.all_docs(include_docs=True)
    for row in result:
        print(row['doc'])
        
finally:
    client.disconnect()

Architecture

The library follows a hierarchical object model:

Client: Top-level connection to Cloudant/CouchDB server with authentication and session management
Database: Represents individual databases with document and view operations
Document: Individual JSON documents with CRUD operations and conflict resolution
DesignDocument: Special documents containing views, indexes, and other database functions
Result: Iterator collections for query results with pagination and key-based access
Feed: Streaming iterators for change feeds and database updates

Capabilities

Client Connection and Authentication

Establish connections to Cloudant and CouchDB servers with support for multiple authentication methods including basic auth, IAM, and Cloud Foundry integration.

def cloudant(user, passwd, **kwargs): ...
def cloudant_iam(account_name, api_key, **kwargs): ...
def cloudant_bluemix(vcap_services, instance_name=None, service_name=None, **kwargs): ...
def couchdb(user, passwd, **kwargs): ...

class Cloudant(CouchDB):
    def __init__(self, user, auth_token, admin_party=False, **kwargs): ...
    @classmethod
    def iam(cls, account_name, api_key, **kwargs): ...
    @classmethod  
    def bluemix(cls, vcap_services, instance_name=None, service_name=None, **kwargs): ...

Client Connection and Authentication

Database Management

Create, delete, and manage databases with support for both CouchDB and Cloudant-specific features including partitioned databases and shard information.

class CouchDatabase(dict):
    def __init__(self, client, database_name, fetch_limit=100, partitioned=False): ...
    def create(self, throw_on_exists=False): ...
    def delete(self): ...
    def exists(self): ...
    def metadata(self): ...
    def doc_count(self): ...

class CloudantDatabase(CouchDatabase):
    def shards(self): ...
    def share_database(self, username, roles=None): ...
    def unshare_database(self, username): ...

Database Management

Document Operations

Complete CRUD operations for JSON documents with attachment support, conflict resolution, and batch operations.

class Document(dict):
    def __init__(self, database, document_id=None, **kwargs): ...
    def create(self): ...
    def fetch(self): ...
    def save(self): ...
    def delete(self): ...
    def update_field(self, action, field, value, max_tries=10): ...
    def get_attachment(self, attachment, headers=None, write_to=None, attachment_type=None): ...
    def put_attachment(self, attachment, content_type, data, headers=None): ...

Document Operations

Query and Indexing

Query documents using Cloudant Query (Mango) with JSON indexes, text search, and pagination support.

class Query:
    def __init__(self, database, selector=None, fields=None, **kwargs): ...
    def __call__(self, **kwargs): ...

class Index:
    def __init__(self, database, design_document_id=None, name=None, partitioned=None, **kwargs): ...
    def create(self): ...
    def delete(self): ...

class TextIndex(Index): ...

Query and Indexing

Views and Design Documents

Manage MapReduce views, list functions, show functions, and update handlers through design documents.

class DesignDocument(Document):
    def __init__(self, database, document_id=None, partitioned=False): ...

class View:
    def __init__(self, design_document, view_name, **kwargs): ...
    def __call__(self, **kwargs): ...

Views and Design Documents

Replication

Set up and manage database replication between Cloudant instances with progress monitoring and state management.

class Replicator:
    def __init__(self, client): ...
    def create_replication(self, source_db=None, target_db=None, repl_id=None, **kwargs): ...
    def list_replications(self): ...
    def replication_state(self, repl_id): ...
    def stop_replication(self, repl_id): ...
    def follow_replication(self, repl_id): ...

Replication

Change Feeds and Monitoring

Monitor database changes in real-time with support for continuous feeds, filtering, and infinite monitoring.

class Feed:
    def __init__(self, source, raw_data=False, **options): ...
    def stop(self): ...
    def __iter__(self): ...

class InfiniteFeed(Feed): ...

Change Feeds and Monitoring

Error Handling

Comprehensive exception hierarchy for handling different types of errors with specific exceptions for client, database, document, and operation-specific failures.

class CloudantException(Exception): ...
class CloudantClientException(CloudantException): ...
class CloudantDatabaseException(CloudantException): ...
class CloudantDocumentException(CloudantException): ...
class CloudantArgumentError(CloudantException): ...

Error Handling

Security Document Management

Manage database access permissions and user roles through security documents with context manager support for automatic fetch and save operations.

class SecurityDocument(dict):
    def __init__(self, database): ...
    def fetch(self): ...
    def save(self): ...
    def json(self): ...
    def __enter__(self): ...
    def __exit__(self, *args): ...

Security Document Management

Scheduler Monitoring

Monitor and query replication jobs and documents using the scheduler API for visibility into active, completed, and failed replications.

class Scheduler:
    def __init__(self, client): ...
    def list_docs(self, limit=None, skip=None): ...
    def get_doc(self, doc_id): ...
    def list_jobs(self, limit=None, skip=None): ...

Scheduler Monitoring

HTTP Adapters

Custom HTTP transport adapters for handling rate limiting, retries, and connection management with automatic 429 response handling.

class Replay429Adapter(HTTPAdapter):
    def __init__(self, retries=3, initialBackoff=0.25): ...

HTTP Adapters

Common Types

# Authentication credentials
CloudFoundryService:
    username: str
    password: str  
    url: str
    iam_api_key: str

# Result collections
class Result:
    def __getitem__(self, key): ...
    def __iter__(self): ...
    def __len__(self): ...
    def all(self): ...

class QueryResult(Result): ...

class ResultByKey:
    def __init__(self, value): ...
    def __call__(self): ...

# Security document structure
SecurityDocumentData = dict[str, Any]
CloudantSection = dict[str, list[str]]  # username -> roles
CouchDBSection = dict[str, list[str]]   # keys: "names", "roles"

# Scheduler responses
DocsResponse = dict[str, Any]  # Contains total_rows, offset, docs
JobsResponse = dict[str, Any]  # Contains total_rows, offset, jobs
ReplicationDoc = dict[str, Any]
ReplicationJob = dict[str, Any]

# Feed options
FeedOptions = dict[str, Any]  # feed parameters like heartbeat, timeout, since

# Index types
JSON_INDEX_TYPE = "json"
TEXT_INDEX_TYPE = "text" 
SPECIAL_INDEX_TYPE = "special"

# HTTP adapter configuration
AdapterConfig = dict[str, Any]
RetryConfig = dict[str, Any]

Install with Tessl CLI