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
database-management.mddocs/
Database Management
Create, delete, and manage databases with comprehensive support for both CouchDB and Cloudant-specific features including partitioned databases, shard information, and database sharing.
Capabilities
Database Access
Access databases through dict-like interface on the client.
class CouchDB(dict):
"""Dict-like interface for database access."""
def all_dbs(self):
"""
List all database names on the server.
Returns:
list[str]: Database names
"""
def create_database(self, dbname, partitioned=False, **kwargs):
"""
Create a new database.
Parameters:
- dbname (str): Database name
- partitioned (bool): Create as partitioned database (Cloudant only)
- **kwargs: Additional database creation options
Returns:
CouchDatabase | CloudantDatabase: Database instance
Raises:
CloudantDatabaseException: Database creation failed
"""
def delete_database(self, dbname):
"""
Delete a database.
Parameters:
- dbname (str): Database name to delete
Returns:
dict: Deletion response
Raises:
CloudantDatabaseException: Database deletion failed
"""
def keys(self, remote=False):
"""
Get database names.
Parameters:
- remote (bool): Fetch from server (default: False, uses cache)
Returns:
list[str]: Database names
"""
def __getitem__(self, key):
"""
Access database by name.
Parameters:
- key (str): Database name
Returns:
CouchDatabase | CloudantDatabase: Database instance
"""
def get(self, key, default=None, remote=False):
"""
Safe database access.
Parameters:
- key (str): Database name
- default: Default value if database not found
- remote (bool): Check existence on server
Returns:
CouchDatabase | CloudantDatabase | default: Database instance or default
"""
def __delitem__(self, key):
"""
Delete database by name.
Parameters:
- key (str): Database name
"""Database Class
Core database operations and metadata management.
class CouchDatabase(dict):
"""
Represents a CouchDB database with document and view operations.
"""
def __init__(self, client, database_name, fetch_limit=100, partitioned=False):
"""
Initialize database instance.
Parameters:
- client (CouchDB | Cloudant): Parent client
- database_name (str): Database name
- fetch_limit (int): Default result limit for queries
- partitioned (bool): Whether database is partitioned
"""
def exists(self):
"""
Check if database exists on server.
Returns:
bool: True if database exists
"""
def create(self, throw_on_exists=False):
"""
Create database on server.
Parameters:
- throw_on_exists (bool): Raise exception if database exists
Returns:
dict: Database creation response
Raises:
CloudantDatabaseException: Creation failed or database exists (if throw_on_exists=True)
"""
def delete(self):
"""
Delete database from server.
Returns:
dict: Deletion response
Raises:
CloudantDatabaseException: Deletion failed
"""
def metadata(self):
"""
Get database metadata and statistics.
Returns:
dict: Database metadata including doc_count, disk_size, etc.
"""
def doc_count(self):
"""
Get total document count in database.
Returns:
int: Number of documents
"""
def partition_metadata(self, partition_key):
"""
Get metadata for a specific partition (partitioned databases only).
Parameters:
- partition_key (str): Partition identifier
Returns:
dict: Partition metadata and statistics
Raises:
CloudantDatabaseException: Database not partitioned or partition not found
"""
def get_revision_limit(self):
"""
Get the revision limit for documents.
Returns:
int: Maximum number of revisions kept per document
"""
def set_revision_limit(self, limit):
"""
Set the revision limit for documents.
Parameters:
- limit (int): Maximum revisions to keep (minimum 1)
Returns:
dict: Response confirming limit change
"""
def view_cleanup(self):
"""
Clean up old view files and indexes.
Returns:
dict: Cleanup response
"""
@property
def database_url(self):
"""Database URL endpoint"""
@property
def r_session(self):
"""HTTP request session"""
@property
def admin_party(self):
"""Whether in Admin Party mode"""
@property
def creds(self):
"""Authentication credentials"""Cloudant Database Extensions
Additional features specific to Cloudant databases.
class CloudantDatabase(CouchDatabase):
"""
Cloudant-specific database with additional cloud features.
"""
def shards(self):
"""
Get shard information for the database.
Returns:
dict: Sharding configuration and node distribution
"""
def share_database(self, username, roles=None):
"""
Share database with another user.
Parameters:
- username (str): Username to share with
- roles (list[str]): Roles to grant (_reader, _writer, _admin, _replicator)
Returns:
dict: Sharing response
Raises:
CloudantDatabaseException: Sharing failed
"""
def unshare_database(self, username):
"""
Remove database sharing for a user.
Parameters:
- username (str): Username to remove access for
Returns:
dict: Unsharing response
Raises:
CloudantDatabaseException: Unsharing failed
"""
@property
def security_url(self):
"""Security document URL"""Server Information
Access server metadata and capabilities.
class CouchDB(dict):
"""Server information methods."""
def metadata(self):
"""
Get server metadata and version information.
Returns:
dict: Server metadata including version, features, vendor
"""
def features(self):
"""
Get server features and capabilities.
Returns:
list[str]: Available server features
"""
def db_updates(self, raw_data=False, **kwargs):
"""
Get database updates feed.
Parameters:
- raw_data (bool): Return raw response data
- **kwargs: Feed options (heartbeat, timeout, since, etc.)
Returns:
Feed: Database updates iterator
"""
class Cloudant(CouchDB):
"""Cloudant server information."""
def infinite_db_updates(self, **kwargs):
"""
Get infinite database updates feed.
Parameters:
- **kwargs: Feed options
Returns:
InfiniteFeed: Perpetual database updates iterator
"""
def shared_databases(self):
"""
List databases shared with the current user.
Returns:
list[str]: Shared database names
"""
def generate_api_key(self):
"""
Generate a new API key for authentication.
Returns:
dict: API key and password
"""
def bill(self, year=None, month=None):
"""
Get billing information.
Parameters:
- year (int): Year for billing data
- month (int): Month for billing data
Returns:
dict: Billing data and usage statistics
"""
def volume_usage(self, year=None, month=None):
"""
Get volume usage statistics.
Parameters:
- year (int): Year for usage data
- month (int): Month for usage data
Returns:
dict: Volume usage data
"""
def requests_usage(self, year=None, month=None):
"""
Get request usage statistics.
Parameters:
- year (int): Year for usage data
- month (int): Month for usage data
Returns:
dict: Request usage data
"""Usage Examples
Basic Database Operations
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
# List all databases
databases = client.all_dbs()
print(f"Databases: {databases}")
# Create database
db = client.create_database('my_new_db')
print(f"Created database: {db.database_url}")
# Access existing database
db = client['existing_db']
# Check if database exists
if db.exists():
print("Database exists")
# Get database metadata
metadata = db.metadata()
print(f"Documents: {metadata['doc_count']}")
print(f"Size: {metadata['disk_size']} bytes")
# Delete database
client.delete_database('old_db')Partitioned Databases (Cloudant)
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
# Create partitioned database
db = client.create_database('partitioned_db', partitioned=True)
# Get partition metadata
partition_info = db.partition_metadata('user123')
print(f"Partition docs: {partition_info['doc_count']}")Database Sharing (Cloudant)
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Share database with read/write access
db.share_database('other_user', roles=['_reader', '_writer'])
# Remove sharing
db.unshare_database('other_user')
# List shared databases
shared = client.shared_databases()
print(f"Shared databases: {shared}")Database Maintenance
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
db = client['my_database']
# Get current revision limit
limit = db.get_revision_limit()
print(f"Current revision limit: {limit}")
# Set revision limit to save disk space
db.set_revision_limit(10)
# Clean up old view files
cleanup_result = db.view_cleanup()
print(f"Cleanup result: {cleanup_result}")Server Information
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
# Get server metadata
server_info = client.metadata()
print(f"Server version: {server_info['version']}")
print(f"Vendor: {server_info['vendor']}")
# Get server features
features = client.features()
print(f"Features: {features}")
# Monitor database updates
for update in client.db_updates():
print(f"Database update: {update}")
break # Stop after first updateUsage Statistics (Cloudant)
from cloudant import cloudant
with cloudant('user', 'pass', account='myaccount') as client:
# Get current month billing
billing = client.bill()
print(f"Monthly cost: ${billing.get('total_cost', 0)}")
# Get volume usage
usage = client.volume_usage()
print(f"Storage used: {usage.get('storage', {}).get('used')} bytes")
# Get request usage
requests = client.requests_usage()
print(f"Requests made: {requests.get('requests', {}).get('count', 0)}")Error Handling
Database operations can raise CloudantDatabaseException:
from cloudant import cloudant
from cloudant.error import CloudantDatabaseException
with cloudant('user', 'pass', account='myaccount') as client:
try:
# Try to create database that already exists
db = client.create_database('existing_db', throw_on_exists=True)
except CloudantDatabaseException as e:
print(f"Database operation failed: {e}")
try:
# Try to delete non-existent database
client.delete_database('non_existent_db')
except CloudantDatabaseException as e:
print(f"Database deletion failed: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-cloudant