File size: 1,653 Bytes
40e9b67 312bffc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
from ibmcloudant.cloudant_v1 import CloudantV1, Document
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
def cloudant_retrieve_documents(
client: CloudantV1,
db_name: str,
selectors: dict,
fields,
sort: dict = None,
limit: int = 100,
):
"""Retrieve documents from the cloudant database."""
cloudant = client
sort_by = sort or {
"sort": [
{"_id": "desc"},
]
}
fields_to_retrieve = (
{"fields": fields} if isinstance(fields, list) else fields
)
if not isinstance(fields, (list, dict)):
raise ValueError("fields must be a list of strings or a dict")
retrieved_docs = cloudant.post_find(
db=db_name,
selector=selectors,
fields=fields_to_retrieve["fields"],
sort=sort,
limit=limit,
).get_result()
return retrieved_docs
def ensure_database_exists(client: CloudantV1, db_name: str) -> bool:
"""
Check if database exists and create it if it doesn't.
Args:
client (CloudantV1): Initialized Cloudant client
db_name (str): Database name to check/create
Returns:
bool: True if database exists or was created successfully
"""
try:
# Try to get database info - will raise error if doesn't exist
client.get_database_information(db=db_name)
return True
except Exception:
try:
# Database doesn't exist, try to create it
client.put_database(db=db_name)
return True
except Exception as e:
print(f"Failed to create database {db_name}: {str(e)}")
raise |