types-google-cloud-ndb

import os from google.cloud import ndb import asyncio # For local development, point to the Datastore emulator os.environ['DATASTORE_EMULATOR_HOST'] = os.environ.get('DATASTORE_EMULATOR_HOST', 'localhost:8081') # Ensure your GOOGLE_CLOUD_PROJECT is set, e.g., in your environment or 'your-gcp-project-id' project_id = os.environ.get('GOOGLE_CLOUD_PROJECT', 'your-gcp-project-id') class User(ndb.Model): name = ndb.StringProperty() email = ndb.StringProperty() joined_date = ndb.DateTimeProperty(auto_now_add=True) async def main(): client = ndb.Client(project=project_id) async with client.context(): # Create a user user = User(name='Alice', email='alice@example.com') user_key = await user.put() print(f'Created user with key: {user_key.id()}') # Fetch the user back by key fetched_user = await user_key.get_async() if fetched_user: print(f'Fetched user: {fetched_user.name} ({fetched_user.email})') # Query for users query = User.query(User.name == 'Alice') users_with_name = await query.fetch_async(limit=1) if users_with_name: print(f'Query result: {users_with_name[0].name}') # Update user if fetched_user: fetched_user.email = 'alice.updated@example.com' await fetched_user.put() print(f'Updated user: {fetched_user.email}') # Delete user await user_key.delete_async() print(f'Deleted user with key: {user_key.id()}') if __name__ == '__main__': asyncio.run(main())