Typing Stubs for Google Cloud NDB

2.4.0.20260408 · active · verified Tue Apr 14

This package provides static type checking stubs for the `google-cloud-ndb` library, enabling tools like MyPy or PyRight to perform compile-time type validation of NDB code. Maintained as part of the `typeshed` project, it currently targets `google-cloud-ndb==2.4.*`. Updates are automatically released, often daily, to reflect changes in the underlying library.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates basic CRUD operations using `google-cloud-ndb` with `async/await` and context management. Installing `types-google-cloud-ndb` alongside `google-cloud-ndb` will enable your type checker to provide rich completions and error detection for this code, enhancing developer experience. Ensure you have the Cloud Datastore emulator running for local testing and `GOOGLE_CLOUD_PROJECT` environment variable set for deployment.

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())

view raw JSON →