gcloud-aio-datastore is an asynchronous Python client for interacting with Google Cloud Datastore. It's part of the `gcloud-aio` monorepo, which provides async interfaces for various Google Cloud services (Datastore, Storage, Pub/Sub, Auth, etc.). This library wraps the official `google-cloud-datastore` client with an `asyncio` interface. The project is actively maintained, with components releasing independently but frequently.
Install & Compatibility
Where this runs
tested against v9.1.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.920 runs
installs and imports cleanly · install 0.0s · import 0.759s · 60.3MB
glibcpy 3.10–3.920 runs
installs and imports cleanly · install 5.6s · import 0.703s · 62MB
61MB installed
● package 61MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
from gcloud.aio.datastore import Datastore
from gcloud.aio.datastore import Key
from gcloud.aio.datastore import Query
This quickstart demonstrates how to initialize the Datastore client using service account credentials, create, retrieve, and query entities. Ensure `GCP_PROJECT` is set to your Google Cloud project ID and `GCP_SERVICE_KEY` is set to your service account key JSON string (or path via `GCP_SERVICE_KEY_PATH`) in the environment. The client uses `async with` for proper resource management.
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())
Upgrade
Version history
Breaking-change detection hasn't run for this library yet.
Audit
Security & dependencies
CVE tracking and dependency tree are planned for a later release.