or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authentication.mdchange-feeds.mddatabase-management.mddocument-operations.mderror-handling.mdhttp-adapters.mdindex.mdquery-indexing.mdreplication.mdscheduler-monitoring.mdsecurity-document.mdviews-design-documents.md
tile.json

tessl/pypi-cloudant

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/cloudant@2.15.x

To install, run

npx @tessl/cli install tessl/pypi-cloudant@2.15.0

index.mddocs/

Cloudant Python Client Library

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]