google-cloud-bigtable

import os from google.cloud import bigtable from google.cloud.bigtable import row from google.cloud.bigtable import column_family # Configuration PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id") # e.g., "my-project-123" INSTANCE_ID = os.environ.get("BIGTABLE_INSTANCE_ID", "your-bigtable-instance-id") # e.g., "my-instance" TABLE_ID = os.environ.get("BIGTABLE_TABLE_ID", "your-table-id") # e.g., "my-table" COLUMN_FAMILY_ID = "cf1" COLUMN_QUALIFIER = "greeting" ROW_KEY = "r1" # Initialize Bigtable client # Ensure GOOGLE_APPLICATION_CREDENTIALS environment variable is set for local dev # or gcloud auth application-default login has been run. # Use admin=True if you need to create/manage tables/instances. client = bigtable.Client(project=PROJECT_ID, admin=True) instance = client.instance(INSTANCE_ID) # Check if table exists, create if not table = instance.table(TABLE_ID) if not table.exists(): print(f"Creating table {TABLE_ID}...") table.create(column_families={COLUMN_FAMILY_ID: column_family.ColumnFamily()}) print(f"Table {TABLE_ID} created.") else: print(f"Table {TABLE_ID} already exists.") # Write a row print(f"Writing row '{ROW_KEY}'...") row_entry = row.DirectRow(ROW_KEY.encode("utf-8")) row_entry.set_cell(COLUMN_FAMILY_ID, COLUMN_QUALIFIER.encode("utf-8"), b"Hello Bigtable!", timestamp=client.timestamp()) table.mutate_rows([row_entry]) print(f"Row '{ROW_KEY}' written.") # Read a row print(f"Reading row '{ROW_KEY}'...") read_row = table.read_row(ROW_KEY.encode("utf-8")) if read_row: # Get the latest cell value for the specific column cells = read_row.cells[COLUMN_FAMILY_ID][COLUMN_QUALIFIER.encode("utf-8")] latest_value = cells[0].value.decode("utf-8") timestamp = cells[0].timestamp print(f"Read: {COLUMN_FAMILY_ID}:{COLUMN_QUALIFIER} = '{latest_value}' (at {timestamp})") else: print(f"Row '{ROW_KEY}' not found.") # Optional: Delete the table for cleanup # print(f"Deleting table {TABLE_ID}...") # table.delete() # print(f"Table {TABLE_ID} deleted.")