def run(self, config):
    storage = gcs.Gcs()
    objs = []
    if 'object' in config:
      objs.append(config['object'])
    if 'objects' in config:
      objects = config['objects']
      for o in storage.ListBucket(objects['bucket'],
                                  objects.get('prefix'),
                                  objects.get('glob')):
        objs.append(gcs.Gcs.MakeUrl(objects['bucket'], o))

    diff = len(objs) - len(config['sinks'])
    if diff < 0:
      msg = ''.join(['More sinks than objects available.',
                     'Ignoring: %s',
                     str(config['sinks'][diff:])])
      logging.warning(msg)
    elif diff > 0:
      logging.info('Found more objects than available sinks.')

    # copy any objects to sinks if the urls differ
    to_copy = zip(objs, config['sinks'])
    res = [storage.CopyObject(o[0], o[1]) for o in to_copy if o[0] != o[1]]
    for r in res:
      logging.info('Copied to %s', r['selfLink'])
Runs the stage.
dataset/ETHPy150Open GoogleCloudPlatform/Data-Pipeline/app/src/pipelines/stages/gcsinput.py/GcsInput.run