cloudant

from cloudant.client import Cloudant from cloudant.error import CloudantException import os # Configure credentials using environment variables CLOUDANT_USERNAME = os.environ.get('CLOUDANT_USERNAME', 'testuser_example') CLOUDANT_PASSWORD = os.environ.get('CLOUDANT_PASSWORD', 'testpass_example') CLOUDANT_URL = os.environ.get('CLOUDANT_URL', 'http://localhost:5984') # Default for local CouchDB client = None try: # Connect to the Cloudant service client = Cloudant(CLOUDANT_USERNAME, CLOUDANT_PASSWORD, url=CLOUDANT_URL, connect=True) print(f"Connected to Cloudant/CouchDB at {CLOUDANT_URL}") session = client.session() print(f"User: {session['userCtx']['name']}") db_name = 'my_sample_database' # Attempt to create a database my_database = client.create_database(db_name) if my_database.exists(): print(f"Database '{db_name}' created or already exists.") # Create a document doc_data = {'name': 'Alice', 'city': 'New York', 'age': 30} new_document = my_database.create_document(doc_data) if new_document.exists(): print(f"Document created with ID: {new_document['_id']}") print(f"Current document content: {new_document}") # Fetch the document by ID fetched_document = my_database[new_document['_id']] print(f"Fetched document content: {fetched_document}") # Update the document (requires fetching it first to get the _rev) fetched_document['age'] = 31 fetched_document['status'] = 'active' fetched_document.save() print(f"Updated document content: {my_database[new_document['_id']]}") # Delete the document fetched_document.delete() print(f"Document '{new_document['_id']}' deleted.") # Delete the database if client.get_database(db_name).exists(): client.delete_database(db_name) print(f"Database '{db_name}' deleted.") except CloudantException as ce: print(f"Cloudant Error: {ce}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: if client: client.disconnect() print("Disconnected from Cloudant service.")