Cloud Datastore single entity read test (usable in Cloud Shell)

datastore_entity_test_1.py

import os
from google.cloud import datastore
import google.cloud.exceptions

#https://stackoverflow.com/questions/1593019/is-there-any-simple-way-to-benchmark-python-script
def timereps(reps, func):
  from time import time
  start = time()
  for i in range(0, reps):
    func()
  end = time()
  
  return (end - start) / reps

def upsert(client, complete_key):
  task = datastore.Entity(key=complete_key)
  
  task.update({
      'category': 'Personal',
      'done': False,
      'priority': 4,
      'description': 'Learn Cloud Datastore'
  })
  
  client.put(task)
  
  return task

def getTest(key):
  for x in range(100): # 100 ops per call
    client.get(key, eventual=True)

def queryTest(key):
  for x in range(100): # 100 ops per call
    results = query.fetch(limit=1)
    for entity in results:
      pass

def runGetTest():
  for y in range(40): # 40 reps
    listdir_time = timereps(1, lambda: getTest(key))
    print("%d seconds for 100 get()" % (listdir_time))

def runQueryTest():
  for z in range(40): # 40 reps
    listdir_time = timereps(1, lambda: queryTest(key))
    print("%d seconds for 100 get()" % (listdir_time))
    
def runIt():
  client = datastore.Client(os.environ['DEVSHELL_PROJECT_ID']) # set project from Cloud Shell env vars
  key = client.key('Task', 'sample_task')
  upsert(client, key) # make sure entity exists
  
  #prep query
  query = client.query(kind='Task')
  query.add_filter('__key__', '=', key)
  
  #run tests
  runGetTest()
  runQueryTest()

if __name__ == '__main__':
  runIt()