Dataflow Daily Upload

catalog_upload.py

#1 Here we define the Dataflow pipeline
with beam.Pipeline() as p:
    # 2 We query all products in the queried_namespace
    query = query_pb2.Query()
    query.kind.add().name = 'Product'
    entities = p | 'Read From Datastore' >>
                    ReadFromDatastore('project_name', query, namespace='queried_namespace')

    # 3 Formatting the rows 
    products = entities | 'Format Rows' >> beam.Map(transform_entity)

    # 4 Writing the results to a GCS file
    column_order = ['id','title','availability','condition','brand',
                   'category','description','image_link','name','price','retailer_id','item_group_id','link']
    header = ','.join(column_order)
    rows | 'Write to GCS' >>
            beam.io.WriteToText(file_path_prefix='gs://filename', file_name_suffix='.csv', num_shards=1, header=header)

    p.run()