Skip to content

Instantly share code, notes, and snippets.

@cavedave
Last active July 5, 2025 13:12
Show Gist options
  • Save cavedave/18920b2c1a42f7ab9347fba225d28def to your computer and use it in GitHub Desktop.
Save cavedave/18920b2c1a42f7ab9347fba225d28def to your computer and use it in GitHub Desktop.
coachbase and gemini check. Code to check if my env file lets us talk to Gemini and Coachbase correctly. A hello world to make sure a connection works. before you start doing clever LLM chats to is.
# Google API Key (required)
GOOGLE_API_KEY=
# Couchbase Capella Configuration (required)
COUCHBASE_CONN_STR=couchbases://cb....
COUCHBASE_USERNAME=...
COUCHBASE_PASSWORD=...
COUCHBASE_BUCKET=travel-sample
import os
from dotenv import load_dotenv
from couchbase.cluster import Cluster, ClusterOptions
from couchbase.auth import PasswordAuthenticator
from couchbase.exceptions import CouchbaseException
from couchbase.collection import UpsertOptions
load_dotenv()
conn_str = os.getenv("COUCHBASE_CONN_STR")
username = os.getenv("COUCHBASE_USERNAME")
password = os.getenv("COUCHBASE_PASSWORD")
bucket_name = os.getenv("COUCHBASE_BUCKET")
try:
cluster = Cluster(conn_str, ClusterOptions(PasswordAuthenticator(username, password)))
bucket = cluster.bucket(bucket_name)
collection = bucket.default_collection()
# Test write
key = "test-doc"
value = {"message": "Hello, Couchbase!"}
collection.upsert(key, value)
# Test read
result = collection.get(key)
print("✅ Couchbase connection successful!")
print("Document content:", result.content_as[dict])
except CouchbaseException as e:
print("❌ Couchbase connection failed:")
print(e)
import os
from dotenv import load_dotenv
from couchbase.cluster import Cluster, ClusterOptions
from couchbase.auth import PasswordAuthenticator
from couchbase.exceptions import CouchbaseException
from couchbase.collection import UpsertOptions
load_dotenv()
conn_str = os.getenv("COUCHBASE_CONN_STR")
username = os.getenv("COUCHBASE_USERNAME")
password = os.getenv("COUCHBASE_PASSWORD")
bucket_name = os.getenv("COUCHBASE_BUCKET")
try:
cluster = Cluster(conn_str, ClusterOptions(PasswordAuthenticator(username, password)))
bucket = cluster.bucket(bucket_name)
collection = bucket.default_collection()
# Test write
key = "test-doc"
value = {"message": "Hello, Couchbase!"}
collection.upsert(key, value)
# Test read
result = collection.get(key)
print("✅ Couchbase connection successful!")
print("Document content:", result.content_as[dict])
except CouchbaseException as e:
print("❌ Couchbase connection failed:")
print(e)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment