#NEWLINE# def run( self , config ) : #NEWLINE# #INDENT# 'Runs the stage.\n\n    Args:\n      config: Specifies the either an inline GQL query or a GCS object that\n        contains a GQL query.\n    ' #NEWLINE# storage = gcs . Gcs ( ) #NEWLINE# gql = config . get ( 'gql' ) #NEWLINE# if ( not gql ) : #NEWLINE# #INDENT# with contextlib . closing ( storage . OpenObject ( url = config [ 'object' ] ) ) as stream : #NEWLINE# #INDENT# with contextlib . closing ( StringIO . StringIO ( ) ) as gql_buf : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# buf = stream . read ( gcs . Gcs . READ_CHUNK_SIZE ) #NEWLINE# if ( buf and len ( buf ) ) : #NEWLINE# #INDENT# gql_buf . write ( buf ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# gql = gql_buf . getvalue ( ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# qkwargs = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# consistency = None #NEWLINE# keys_only = False #NEWLINE# projection = None #NEWLINE# if ( 'params' in config ) : #NEWLINE# #INDENT# params = keys_only [ 'params' ] #NEWLINE# qkwargs = params . get ( 'values' , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# consistency = params . get ( 'consistency' ) #NEWLINE# if ( ( 'consistency' in params ) and ( params [ 'consistency' ] is 'eventual' ) ) : #NEWLINE# #INDENT# consistency = ndb . EVENTUAL_CONSISTENCY #NEWLINE# #UNINDENT# keys_only = params . get ( 'keysOnly' , False ) #NEWLINE# projection = params . get ( 'projection' ) #NEWLINE# #UNINDENT# writer = None #NEWLINE# with contextlib . closing ( StringIO . StringIO ( ) ) as buf : #NEWLINE# #INDENT# query = ndb . gql ( gql , ** qkwargs ) #NEWLINE# for entity in query . iter ( read_policy = consistency , keys_only = keys_only , projection = projection ) : #NEWLINE# #INDENT# if ( not projection ) : #NEWLINE# #INDENT# projection = entity . _properties . keys ( ) #NEWLINE# #UNINDENT# if ( not writer ) : #NEWLINE# #INDENT# writer = csv . DictWriter ( buf , projection ) #NEWLINE# headers = dict ( ( ( p , p ) for p in projection ) ) #NEWLINE# writer . writerow ( headers ) #NEWLINE# #UNINDENT# writer . writerow ( entity . to_dict ( ) ) #NEWLINE# #UNINDENT# buf . seek ( 0 ) #NEWLINE# storage . InsertObject ( buf , url = config [ 'sinks' ] [ 0 ] )