def run(self, config):
    storage = gcs.Gcs()
    gql = config.get('gql')
    if not gql:
      with contextlib.closing(
          storage.OpenObject(url=config['object'])) as stream:
        with contextlib.closing(StringIO.StringIO()) as gql_buf:
          while True:
            buf = stream.read(gcs.Gcs.READ_CHUNK_SIZE)
            if buf and len(buf):
              gql_buf.write(buf)
            else:
              break
          gql = gql_buf.getvalue()

    qkwargs = {}
    consistency = None
    keys_only = False
    projection = None
    if 'params' in config:
      params = config['params']
      qkwargs = params.get('values', {})
      consistency = params.get('consistency')
      if 'consistency' in params and params['consistency'] is 'eventual':
        consistency = ndb.EVENTUAL_CONSISTENCY
      keys_only = params.get('keysOnly', False)
      projection = params.get('projection')

    # for now just emit a dumb CSV
    # TODO(user): better way to decide output type.
    # TODO(user): make robust - don't always want to spool into memory...
    writer = None
    with contextlib.closing(StringIO.StringIO()) as buf:
      query = ndb.gql(gql, **qkwargs)
      for entity in query.iter(read_policy=consistency,
                               keys_only=keys_only,
                               projection=projection):
        if not projection:
          # pylint: disable=protected-access
          projection = entity._properties.keys()
        if not writer:
          writer = csv.DictWriter(buf, projection)
          headers = dict((p, p) for p in projection)
          writer.writerow(headers)
        writer.writerow(entity.to_dict())

      # TODO(user): what to do with multiple sinks?
      buf.seek(0)
      storage.InsertObject(buf, url=config['sinks'][0])
Runs the stage.
dataset/ETHPy150Open GoogleCloudPlatform/Data-Pipeline/app/src/pipelines/stages/datastoreinput.py/DatastoreInput.run