gcloud-aio-datastore

import asyncio import os from gcloud.aio.auth import build_from_service_account from gcloud.aio.datastore import Datastore, Key, Query async def main(): project = os.environ.get('GCP_PROJECT', 'your-gcp-project-id') service_account_info = os.environ.get('GCP_SERVICE_KEY') # or path via GCP_SERVICE_KEY_PATH if not project or not service_account_info: print("Please set GCP_PROJECT and GCP_SERVICE_KEY environment variables.") return # Credentials can also be built from a path using build_from_service_account_path() creds = build_from_service_account(service_account_info) async with Datastore(project=project, credentials=creds) as client: # 1. Create an entity kind = 'MyEntity' name = 'my-unique-entity-name' key = Key([kind, name], project=project) entity_data = { 'property1': 'value1', 'property2': 123 } await client.put_entity(key, entity_data) print(f"Created/updated entity: {key.path[0]['id']}") # 2. Get an entity retrieved_entity = await client.get_entity(key) if retrieved_entity: print(f"Retrieved entity: {retrieved_entity.properties}") else: print(f"Entity {key.path[0]['id']} not found.") # 3. Run a query query = Query(kind=kind, project=project) results = await client.run_query(query) print("Query results:") for entity in results: print(f" Kind: {entity.kind}, ID: {entity.id}, Properties: {entity.properties}") # 4. Delete an entity (optional cleanup) # await client.delete_entity(key) # print(f"Deleted entity: {key.path[0]['id']}") if __name__ == '__main__': asyncio.run(main())